SynchronizationContext sample

SingleThreadSynchronizationContext.cs

using Reactive.Bindings;
using System;
using System.Collections;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Reactive.Concurrency;
using System.Reactive.Linq;
using System.Threading;
using System.Windows.Input;

namespace ConsoleApplication1
{
    /// <summary>Provides a SynchronizationContext that's single-threaded.</summary>
    class SingleThreadSynchronizationContext : SynchronizationContext
    {
        /// <summary>The queue of work items.</summary>
        private readonly BlockingCollection<KeyValuePair<SendOrPostCallback, object>> m_queue =
            new BlockingCollection<KeyValuePair<SendOrPostCallback, object>>();
        /// <summary>The processing thread.</summary>
        private readonly Thread m_thread = Thread.CurrentThread;

        /// <summary>Dispatches an asynchronous message to the synchronization context.</summary>
        /// <param name="d">The System.Threading.SendOrPostCallback delegate to call.</param>
        /// <param name="state">The object passed to the delegate.</param>
        public override void Post(SendOrPostCallback d, object state)
        {
            if (d == null) throw new ArgumentNullException("d");
            m_queue.Add(new KeyValuePair<SendOrPostCallback, object>(d, state));
        }

        /// <summary>Not supported.</summary>
        public override void Send(SendOrPostCallback d, object state)
        {
            throw new NotSupportedException("Synchronously sending is not supported.");
        }

        /// <summary>Runs an loop to process all queued work items.</summary>
        public void RunOnCurrentThread()
        {
            foreach (var workItem in m_queue.GetConsumingEnumerable())
                workItem.Key(workItem.Value);
        }

        /// <summary>Notifies the context that no more work will arrive.</summary>
        public void Complete() { m_queue.CompleteAdding(); }
    }

    class ViewModel
    {
        ReactiveCommand m_command;
        public ICommand Command
        {
            get
            {
                if(m_command== null)
                {
                    m_command = new ReactiveCommand();
                    m_command.Subscribe(_ =>
                    {
                        Console.WriteLine("Command");
                    });
                }
                return m_command;
            }
        }
    }

    class Program
    {

        static void Main(string[] args)
        {
            // contextを作ってセットする
            var context = new SingleThreadSynchronizationContext();
            SynchronizationContext.SetSynchronizationContext(context);

            // 本体
            _Main(args);

            Observable.Start(() =>
            {
                // エンターで停止・・・
                Console.ReadLine();
            })
            .Subscribe(x =>
            {
                // コンテキストを終了させる
                context.Complete();
            });

            // ブロッキング
            context.RunOnCurrentThread();
        }

        static void _Main(string[] args)
        {
            var vm = new ViewModel();

            Console.WriteLine("ThreadID: " + Thread.CurrentThread.ManagedThreadId);
            var subscription =
                Observable.Interval(TimeSpan.FromSeconds(2))
                .Take(1)
                .Do(x =>
                {
                    Console.WriteLine("ThreadID: " + Thread.CurrentThread.ManagedThreadId);
                })
                .ObserveOn(SynchronizationContext.Current)
                .Do(x =>
                {
                    Console.WriteLine("ThreadID: " + Thread.CurrentThread.ManagedThreadId);
                })
                .Subscribe(x =>
                {
                    vm.Command.Execute(null);
                }
                , ex=>
                {
                }
                , ()=>
                {
                })
                ;
        }
    }
}