

SignalR(.NET Client)を使用して、イベントの通知とデータの取得を行うプログラムを作りました。  
やりたいことはとりあえず出来たのですが、あまり処理を綺麗に書けませんでした。  
マルチスレッドの問題やDisposable漏れなどに考慮して、もっと処理をスマートに書けるものでしょうか？  
Rxなどを使うともっと綺麗に書けたりするものなのかもしれませんが、自分の力では、これをどうRxで記述すれば良いかイメージができませんでした。

なにかアイディアがあれば力をお貸しください。

■やりたいこと

  * サーバ側は一定間隔でクライアントにデータの発生を通知する。
  * データはキューに保持する。

  * クライアントはデータ発生のイベントを受信すると、サーバからデータ取得を行う。

  * データ発生のイベントだけでなく、接続直後にもデータの取得を行う。
  * データ取得は、サーバのキューが空になるまで続けて呼び出す。
  * クライアントが接続に失敗した場合、一定時間後に再接続を試みる。
  * クライアントは任意の時点で開始/停止ができる。

■クライアント/サーバ共通ソース

[code]

    public class Data
    {
        public int Id { get; set; }
        public string Text { get; set; }
    }
    
[/code]

■サーバ側ソース

[code]

    // Hub
    [HubName("test")]
    public class TestHub : Hub
    {
        // データ取得処理
        public Data Query()
        {
            return Service.Deque();
        }
    }
    
    // 通知およびキューの管理、事前にStartしておく
    public static class Service
    {
        private static readonly ConcurrentQueue<Data> queue = new ConcurrentQueue<Data>();
    
        private static int nextId = 1;
    
        private static Timer timer;
    
        // 一定期間毎にキューにデータを入れて、クライアントに通知
        public static void Start(IHubConnectionContext<dynamic> clients)
        {
            timer = new Timer(state =>
            {
                nextId++;
                queue.Enqueue(new Data { Id = nextId, Text = String.Format("Data-{0}", nextId) });
    
                clients.All.Update();
            }, null, 3000, 3000);
        }
    
        public static void Stop()
        {
            timer.Dispose();
        }
    
        public static Data Deque()
        {
            Data data;
            queue.TryFetchAndAction(out data);
            return data;
        }
    }
    
[/code]

■クライアント側ソース(もっとスマートにしたい部分)

[code]

    public class Client
    {
        // hub操作用の同期オブジェクト
        private readonly object sync = new object();
    
        // 実行状態、Closed時の再接続と、データ取得の連続処理中断に使用
        private readonly ManualResetEvent running = new ManualResetEvent(false);
    
        // 再接続用のタイマー
        private Timer timer;
    
        private HubConnection hub;
    
        private IHubProxy proxy;
    
        private void Start()
        {
            if (hub != null)
            {
                return;
            }
    
            // 実行状態に移行
            running.Set();
    
            hub = new HubConnection("http://127.0.0.1:10080");
            hub.Closed += () =>
            {
                // 旧タイマーを片付け
                var disposable = Interlocked.Exchange(ref timer, null);
                if (disposable != null)
                {
                    disposable.Dispose();
                }
    
                if (running.WaitOne(0))
                {
                    // 一定時間後に再接続の試行
                    Interlocked.Exchange(ref timer, new Timer(state =>
                    {
                        lock (sync)
                        {
                            if (hub != null)
                            {
                                ConnectHubAsync();
                            }
                        }
                    }, null, 5000, 0));
                }
            };
    
            proxy = hub.CreateHubProxy("test");
            proxy.On("Update", FetchAndAction);
    
            ConnectHubAsync();
        }
    
        // 接続と、接続とのデータ取得
        private void ConnectHubAsync()
        {
            hub.Start().ContinueWith(t =>
            {
                if (hub.State == ConnectionState.Connected)
                {
                    FetchAndAction();
                }
            });
        }
    
        // データ取得の連続処理
        private async void FetchAndAction()
        {
            while (running.WaitOne(0))
            {
                Data data;
                try
                {
                    data = await proxy.Invoke<Data>("Query");
                }
                catch (Exception ex)
                {
                    // Invoke中にHubConnection.Stop()するとInvalidOperationException発生
                    break;
                }
    
                // もうデータがない
                if (data == null)
                {
                    break;
                }
    
                // ここでDataを元に処理を行う
            }
        }
    
        private void Stop()
        {
            if (hub == null)
            {
                return;
            }
    
            // 停止状態に移行
            running.Reset();
    
            lock (sync)
            {
                var disposable = Interlocked.Exchange(ref timer, null);
                if (disposable != null)
                {
                    disposable.Dispose();
                }
    
                hub.Stop();
                hub = null;
            }
        }
    }
    
[/code]


CC BY-SA 3.0
2015-03-23T10:21:56.410
8264
2016-11-14T10:58:10.623
8965
question
c# .net
SignalR(.NET Client)の使用方法


クライアント側の同期機構が`lock
(sync)`、`Interlocked.Exchange`、`ManualResetEvent`と三種類も存在するのが気になります。ロック時間が特別にクリティカルでもない限り単純に`lock
(sync)`で制御すれば大分シンプルになるのではないでしょうか。

また

[code]

    var disposable = Interlocked.Exchange(ref timer, null);
    if (disposable != null)
    {
        disposable.Dispose();
    }
    
[/code]

は`using`ステートメントで

[code]

    using(Interlocked.Exchange(ref timer, null)) { }
    
[/code]

と書けます。


CC BY-SA 3.0
2015-03-23T11:24:38.197
8269
2015-03-23T11:24:38.197
5750
8264
answer


何らかのリソースを抱えるクラスは、IDisposableインターフェースを実装します。  
例えば、Clientクラスでは、Timer、ManualResetEventといったものがリソースと見なされます。  
それらの特徴として、いずれもIDisposableを実装しています。

Clientクラスのインスタンスが破棄されるとき、それらオブジェクトのDisposeメソッドを呼び出す必要があります。

Clientクラスを今後、継承して利用する必要がなければ、sealed を指定しておいたほうが良いでしょう。  
次にIDisposable パターンを実装した例を示します。

[code]

    public sealed class Client : IDisposable {
        private int disposed = 0;
        private void Dispose( bool disposing ) {
            if( Interlocked.CompareExchange( ref disposed , 1 , 0 ) == 0 ) {
                // ここでリソースを解放する、つまり timer.Dispose(); などを呼び出す。
                GC.SuppressFinalize( this );
            }
        }
    
        public void Dispose() {
            Dispose( true );
        }
    
        ~Client(){
            Dispose( false );
        }
    }
    
[/code]

~Client() の部分は、デストラクタを定義しています。  
こうすることで、仮にDisposeが呼ばれないケースがあったとしても、GCがそのうちDisposeを呼び出してくれます。  
GC.SuppressFinalize( this ) は、GCにデストラクタを呼び出す必要がないことを通知します。  
しかし、デストラクタは、デストラクタを実装しないオブジェクトに比べ、回収されにくくにります。(後回しにされるイメージ)  
そのため、無闇にデストラクタを実装することは、推奨されませんが、今回のようなケースでは、有効な方法だと思われます。

もし、Clientクラスを派生させる場合は、 void Dispose( bool disposing )メソッドを  
virtual指定し、オーバーライドして下さい。

また非同期パターンにおけるロック機構として、 **System.Threading.SemaphoreSlim** を使用する方法があります。  
詳しくは、[非同期：awaitを含むコードをロックするには？（SemaphoreSlim編）](http://www.atmarkit.co.jp/ait/articles/1411/11/news117.html)

を参考にして下さい。


CC BY-SA 3.0
2015-05-22T12:47:22.927
10510
2015-05-22T12:47:22.927
2207
8264
answer