Наблюдаемые <string>обновленных событий?

Question

Я просто пытаюсь создать простой обработчик событий для строковой переменной, чтобы в случае изменения строки она выполняла Console.WriteLine (используя новую библиотеку Reactive от MS (Rx))

Проблема, с которой я столкнулся, заключается в том, что он будет отображать первый бит, когда я создаю экземпляр класса («RandomGuid: Mine?»), Но после этого ни одна из вещей, которые я изменяю, впоследствии не выплевывает что-либо на консоль.

Я прошел через HOL с веб-сайта MS, но он идет от определения Observable к чтению значений из текстового поля, когда все, что я хочу сделать, это посмотреть, была ли изменена строка.

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;

namespace MynahBirds
{
    class Program
    {
        static void Main(string[] args)
        {
            List<Minah> minahs = new List<Minah>();

            for (int i = 0; i < 10; i++) {
                minahs.Add(new Minah());
            }

            foreach (var item in minahs) {
                item.peers = minahs;
            }

            minahs.ForEach(m => m.s = Observable.Return<string>("Mine"));
            minahs.ForEach(m => m.s = Observable.Return<string>("Whee"));
            minahs.ForEach(m => m.s = Observable.Return<string>("Argh"));

            Console.ReadLine();
        }
    }
    class Minah
    {
        Guid Id;
        public List<Minah> peers;
        IDisposable subscription;

        public IObservable<string> s = Observable.Return<string>("Mine?");

        public Minah()
        {
            try {
                this.Id = Guid.NewGuid();

                subscription = s.Subscribe((string a) => {
                    Console.WriteLine("{0} : {1}", this.Id, a);
                },
                (Exception ex) => {
                    Console.WriteLine("Error {0} hit", ex.ToString());
                },
                () => { });

            } catch (Exception ex) {
                Console.WriteLine(ex.ToString());
                Console.ReadLine();
                throw;
            }
        }
    }
}

Answer 1

Когда вы присваиваете m.s в ForEach, вы не обновляете существующую наблюдаемую (на которую вы подписались) новым значением, вместо этого вы создаете новые наблюдаемые, что и делает Observable.Return.Код ниже делает то, что, я думаю, вы ожидаете:

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;

namespace MynahBirds
{
    class Program
    {
        static void Main(string[] args)
        {
            List<Minah> minahs = new List<Minah>();

            for (int i = 0; i < 10; i++)
            {
                minahs.Add(new Minah());
            }

            foreach (var item in minahs)
            {
                item.peers = minahs;
            }

            minahs.ForEach(m => m.s.OnNext("Mine"));
            minahs.ForEach(m => m.s.OnNext("Whee"));
            minahs.ForEach(m => m.s.OnNext("Argh"));

            Console.ReadLine();
        }
    }
    class Minah
    {
        Guid Id;
        public List<Minah> peers;
        IDisposable subscription;

        public ISubject<string> s = new Subject<string>();

        public Minah()
        {
            try
            {
                this.Id = Guid.NewGuid();

                subscription = s.Subscribe((string a) =>
                {
                    Console.WriteLine("{0} : {1}", this.Id, a);
                },
                (Exception ex) =>
                {
                    Console.WriteLine("Error {0} hit", ex.ToString());
                },
                () => { });

            }
            catch (Exception ex)
            {
                Console.WriteLine(ex.ToString());
                Console.ReadLine();
                throw;
            }
        }
    }
}

Вместо использования Observable.Return<T>(), здесь я использую Subject, который является одновременно наблюдателем и наблюдаемой последовательностью.Он обновляет все свои подписки с каждым значением, которое он наблюдает.Таким образом, когда OnNext вызывается для субъекта, он направляется всем подпискам.

Если вам нужно начальное значение (Mine?), вы можете добавить s.OnNext("Mine?"); в конце конструктора Minah.

Answer 2

using System;
using System.Collections.Generic;
using System.Threading;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Diagnostics;
using System.Collections.Concurrent;

namespace MynahBirds
{
    class Program
    {
        static void Main(string[] args)
        {
            ThreadPool.SetMaxThreads(100, 100);

            ConcurrentBag<Minah> minahs = new ConcurrentBag<Minah>();
            Stopwatch ti = new Stopwatch();

            ti.Start();

            Task.Factory.StartNew(() => {
                for (int i = 1; i < 2501; i++) {
                    minahs.Add(new Minah(i));
                };
            });

            Task.Factory.StartNew(() => {
                for (int i = 1; i < 2501; i++) {
                    minahs.Add(new Minah(i));
                };
            });

            Task.Factory.StartNew(() => {
                for (int i = 1; i < 2501; i++) {
                    minahs.Add(new Minah(i));
                };
            });

            Task.Factory.StartNew(() => {
                for (int i = 1; i < 2501; i++) {
                    minahs.Add(new Minah(i));
                };
            });

            Task.WaitAll();

            string[] alpha = { "Alpha", "Bravo", "Charlie", "Delta", "Eagle", "Foxtrot", "Gulf", "Hotel" };

            foreach (string s in alpha) {
                Console.WriteLine(s);
                Task.Factory.StartNew(() => minahs.AsParallel().ForAll(m => m.RepeatWord = s)).Wait();
            }

            minahs.AsParallel().ForAll(m => m.s.OnCompleted());

            ti.Stop();

            Console.WriteLine("{1} birds : {0} seconds", ti.Elapsed.TotalSeconds, minahs.Count);
            Console.ReadLine();
        }
    }
    class Minah
    {
        Guid Id;
        IDisposable subscription;

        public ISubject<string> s = new Subject<string>();

        private string _RepeatWord;
        public string RepeatWord
        {
            get
            {
                return _RepeatWord;
            }
            set
            {
                this.s.OnNext(value);
                _RepeatWord = value;
            }
        }

        public Minah(int i)
        {
            try {
                this.Id = Guid.NewGuid();

                subscription = s.Subscribe((string a) => {
                    Console.WriteLine("{0} : {1}", i, a);
                },
                (Exception ex) => {
                    Console.WriteLine("Error {0} hit", ex.ToString());
                },
                () => { /* Console.WriteLine("{0} : Completed", this.Id); */ });

            } catch (Exception ex) {
                Console.WriteLine(ex.ToString());
                Console.ReadLine();
                throw;
            }
        }
    }
}

Это то, что я закончил делать с помощью Маркуса.Больше играть с параллелизмом.Интересно, что если я удалю .Wait () из конца .ForAll (... RepeatWord = s), он будет выполнять только последнее слово в последовательности.Я предполагаю, что это закрытие, но я не слишком обеспокоен этим.