引き続きWebSocket。

クライアント側のサーバーから切断されてしまった時の再接続機能をさくっと作ってみた。
基本的にはクローズイベントかエラー発生のイベントで再接続をすれば良い。ただ、APIによって仕様が違ったりする。将来的には実装されるAPIまで標準化されることでしょう。

ここでは画面上にサーバーが送信してくる時刻を常に表示し続けるクライアントを作成します。
画面

C# – WCF WebSockets版

using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Drawing;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Windows.Forms;

using System.ServiceModel.WebSockets;

namespace WsForm_WCF
{
    public partial class Form1 : Form
    {

        /// <summary>
        /// WebSocketsインスタンス
        /// </summary>
        private WebSocket _ws = null;

        /// <summary>
        /// リトライカウンタ
        /// </summary>
        private int _retryCount = 0;

        public Form1()
        {
            InitializeComponent();
        }

        private void Form1_Load(object sender, EventArgs e)
        {
            /// サーバー接続開始
            Connect();
        }

        /// <summary>
        /// サーバーへ接続する
        /// </summary>
        private void Connect()
        {

            /// listBoxへ文字列を挿入する。
            Action<string> AddText = (s) =>
                {
                    listBox1.Items.Add(s);
                    listBox1.TopIndex = listBox1.Items.Count - 1;
                };

            if (_ws == null)
            {
                AddText("サーバー接続を開始します。");
            }
            _ws = new WebSocket("ws://192.168.1.2:12345");

            /// 文字列受信
            _ws.OnData += (s, e) =>
            {
                AddText(e.TextData);
            };

            /// サーバー接続完了
            _ws.OnOpen += (s, e) =>
            {
                _retryCount = 0;
                AddText("サーバーに接続しました。");
            };

            /// 接続断の発生
            _ws.OnClose += (s, e) =>
            {
                AddText("サーバー接続中..リトライ" + (++_retryCount).ToString() + "回目");

                /// 再接続を試行する
                Connect();
            };

            /// サーバー接続開始
            _ws.Open();

        }
    }
}

C# – WebSocket4Net版

using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Data;
using System.Drawing;
using System.Linq;
using System.Text;
using System.Threading.Tasks;
using System.Windows.Forms;

using WebSocket4Net;

namespace WsForm_WebSocket4Net
{
    public partial class Form1 : Form
    {
        /// <summary>
        /// WebSocketsインスタンス
        /// </summary>
        private WebSocket _ws = null;

        /// <summary>
        /// リトライカウンタ
        /// </summary>
        private int _retryCount = 0;

        public Form1()
        {
            InitializeComponent();
        }

        private void Form1_Load(object sender, EventArgs e)
        {
            /// サーバー接続開始
            Connect();
        }

        /// <summary>
        /// サーバーへ接続する
        /// </summary>
        private void Connect()
        {

            /// listBoxへ文字列を挿入する。
            Action<string> AddText = (s) =>
            {
                this.Invoke(new Action(() =>
                    {
                        listBox1.Items.Add(s);
                        listBox1.TopIndex = listBox1.Items.Count - 1;
                    }));
            };

            if (_ws == null)
            {
                AddText("サーバー接続を開始します。");
            }
            _ws = new WebSocket("ws://192.168.1.2:12345");

            /// 文字列受信
            _ws.MessageReceived += (s, e) =>
            {
                AddText(e.Message);
            };

            /// サーバー接続完了
            _ws.Opened += (s, e) =>
            {
                _retryCount = 0;
                AddText("サーバーに接続しました。");
            };

            /// 接続断の発生
            _ws.Error += (s,e)=>
            {
                AddText("サーバー接続中..リトライ" + (++_retryCount).ToString() + "回目");
                /// 再接続を試行する
                Connect();
            };

            /// 接続断の発生
            _ws.Closed += (s, e) =>
            {
                AddText("サーバー接続中..リトライ" + (++_retryCount).ToString() + "回目");
                /// 再接続を試行する
                Connect();
            };

            /// サーバー接続開始
            _ws.Open();

        }
    }
}

WCF WebSocketsとWebSocket4Netの違いはほぼないものの、エラーハンドリングがあるだけWebSocket4Netはちょっと分かりやすいかもしれません。
WCF WebSocketsのエラー内容はOnClose()のsenderから得ることができます。

            ws.OnClose += (s, e) =>
            {
                Console.WriteLine("Last Error:{0}.", ((WebSocket)s).LastError.ToString());
            };

実行画面

気になったのはエラー発生後、サーバーへの接続を行うsleepは指定していませんが、どれも1秒おきくらいにやっているように見えることです。ちなみに、ブラウザ(Javascript版)では実行環境の違いによって試行までの時間(他のPC環境では10秒おきくらい)が違ったりしました。

おまけ
JavaScript版

<!DOCTYPE html>
<html>
<head>
    <meta charset="utf-8" />
    <title>WebSocket Sample</title>
</head>
<body>
    <div class="page">
        <header>
            <div id="title">
                <h1>WebSocket アプリケーション</h1>
            </div>
        </header>
		<div id="log"></div>
		<script type="text/javascript">
			var ws;
			var retryCount = 0;
			var logCount = 0;
			var connect = function(){
				
				if ("WebSocket" in window) {
					ws = new WebSocket("ws://192.168.1.2:12345");
				} else if ("MozWebSocket" in window) {
					ws = new MozWebSocket("ws://192.168.1.2:12345");
				}

				ws.onmessage = function(event){
				    output(event.data);
				}
				
				if( retryCount > 0 ){
					output("サーバー接続中..リトライ" + retryCount.toString() + "回目");
				}

				
				ws.onopen = function(event){
				    output("サーバーに接続しました。");
				    retryCount = 0;
				}

				ws.onclose = function(event){
					retryCount++;
					connect();
				}

				function output(str){
					document.getElementById("log").innerHTML += str + "<hr />";
					logCount++;
					scroll(0,logCount*100);
				}

				function disconnect(){
					ws.close();
					ws = null;
				}
			}
			
			connect();
		</script>
        </section>
        <footer>
        </footer>
    </div>
</body>
</html>

ブロードキャストエコーサーバー
WebSocketのテスト用。
送信時にちょっと工夫。ユーザーに送信する際にParallel.ForEach()を使用して並列化してみた。これを使うと複数ユーザーに送信する処理を並列化できる。例えば大きいデータなどを分割して送りたい処理などに有効。但し、リソースの消費量が大きくなるので注意が必要。Parallel.ForEach()はマルチコアCPUをフル活用するからパソコンのファンがブーンって鳴る。

並列化の効用。
複数の画像を各クライアントにプッシュ配信するサーバーの例。合計1.77MBの画像データを8KBパケットに分割して送信するとします。実行画面では1フォルダにつき1クライアントが動作しています。

以下、シンプルブロードキャストサーバーのソース。
サーバーソース

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

using SuperSocket.Common;
using SuperSocket.SocketBase;
using SuperSocket.SocketBase.Config;
using SuperSocket.SocketEngine;
using SuperWebSocket;

namespace WsEchoServer
{
    class Program
    {
        static void Main(string[] args)
        {
            /// WebSocket初期化
            var server = new SuperWebSocket.WebSocketServer();
            var rootConfig = new SuperSocket.SocketBase.Config.RootConfig();
            var serverConfig = new SuperSocket.SocketBase.Config.ServerConfig()
            {
                Port = 12345,
                Ip = "Any",
                MaxConnectionNumber = 100,       /// 最大ユーザセッション数
                Mode = SocketMode.Sync,
                Name = "Echo Server"
            };

            /// クライアントセッションを格納
            var clinet_sessions = new List<WebSocketSession>();

            /// セッションの接続
            server.NewSessionConnected += (s) =>
                {
                    Console.WriteLine("{0}:New Session:{1}", DateTime.Now.ToString(), s.RemoteEndPoint.Address.ToString());

                    /// 新クライアントとして格納
                    clinet_sessions.Add(s);
                };

            /// セッションのクローズ
            server.SessionClosed += (s, r) =>
                {
                    Console.WriteLine("{0}:Session Close:{1}", DateTime.Now.ToString(), s.RemoteEndPoint.Address.ToString());

                    /// 該当クライアントの除外
                    clinet_sessions.Remove(s);
                };

            /// データ受信(バイナリ)
            server.NewDataReceived += (s, d) =>
                {
                    Console.WriteLine("{0}:Binary Received:{1}", DateTime.Now.ToString(), s.RemoteEndPoint.Address.ToString());
                    Console.WriteLine("Length={0}", d.Length);
                    
                    /// 全ユーザに送信
                    Parallel.ForEach(clinet_sessions, p => p.SendResponse(d));
                };

            /// データ受信(文字列)
            server.NewMessageReceived += (s, m) =>
                {
                    Console.WriteLine("{0}:String Received:{1}",DateTime.Now.ToString(), s.RemoteEndPoint.Address.ToString());
                    Console.WriteLine("Message={0}", m);

                    /// 全ユーザに送信
                    Parallel.ForEach(clinet_sessions, p => p.SendResponse(m));
                };


            /// サーバセットアップ
            server.Setup(rootConfig, serverConfig, SocketServerFactory.Instance);
            /// サーバ起動
            server.Start();

            Console.WriteLine("{0}:Server start.",DateTime.Now.ToString());
        }
    }
}

クライアントにはWebSocket4Netを使用。
これまでクライアントにはWCF WebSocketsを使っていたけど、SuperWebSocketsに含まれていたWebSocket4Netを使用してみる。
ほぼWCF WebSocketsと使用感は変わらない。

WebSocket4Net
http://websocket4net.codeplex.com/

クライアントソース

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

using WebSocket4Net;

namespace WsClient
{
    class Program
    {
        static void Main(string[] args)
        {
            var ws = new WebSocket("ws://192.168.1.2:12345");

            /// 文字列受信
            ws.MessageReceived += (s, e) =>
                {
                    Console.WriteLine("{0}:String Received:{1}", DateTime.Now.ToString(), e.Message);
                };

            /// バイナリ受信
            ws.DataReceived +=(s,e)=>
                {
                    Console.WriteLine("{0}:Binary Received Length:{1}", DateTime.Now.ToString(), e.Data.Length);
                };

            /// サーバ接続完了
            ws.Opened += (s, e) =>
            {
                Console.WriteLine("{0}:Server connected.", DateTime.Now.ToString());
            };

            /// サーバ接続開始
            ws.Open();

            /// 送受信ループ
            while (true)
            {
                var str = Console.ReadLine();
                if (str == "END") break;

                if (ws.State == WebSocketState.Open)
                {
                    ws.Send(str);
                }
                else
                {
                    Console.WriteLine("{0}:wait...", DateTime.Now.ToString());
                }
            }


            /// ソケットを閉じる
            ws.Close();

        }
    }
}

実行画面

今回作成したソースはこちら。

とうとうできたWebSocketのC#サーバ♪
WCFのWebSocketではサーバ側プログラムがどうにもこうにも動作できないでいたので、別のコンポーネントを試してみた。

SuperWebSocket, a .NET WebSocket Server
http://superwebsocket.codeplex.com/

こちらの方のサイトをちょっと参考。感謝感謝。
つくるよ
http://blog.livedoor.jp/tukuruyo/tag/SuperWebSocket

簡単な動作テストを行った所、動作が確認できたので、感動。でも、SuperWebSocketの機能をどうやって使用していいのかよくわかりません。全部英語だし。検索してみても使っている人はまだあまりいなそう。

Pythonのサーバをコーディングし直し。
以前作成したチャットソフトのサーバ側プログラムをフルスクラッチで作成。今までVMでPythonやっていたけど、VM起動する必要なくなった。

◇サーバプログラム・・・C#, SuperWebSocket
できること。
・Pythonで作ったサーバプログラムと同等の機能。
・ソケットのデータフォーマットにはMessagePackを使用。
・なんか、動いているのがわかるようなビューを作成。

◇クライアントプログラム・・・C#, WCF WebSockets
・以前作成したチャットソフトと同じもの。

クライアントプログラム　コーディング
サーバプログラムを作成するにあたり、共通部分についてはクラスライブラリ化した。なので、改修箇所はその部分のみ。WebSocketは規格なので、ほかにいじるところなし。

サーバプログラム コーディング
キモな部分だけ抽出。

ユーザソケットのセッション管理について
Pythonでコーディングした時、結構ややこしい処理になっていたところをちょと反省。ちゃんとデータクラスを作成して管理することに。
基本的に下記のSessionInfoクラスのリストをグローバルに置いて、各処理がユーザセッションにアクセスできるようにした。
WebSocketSession というのがユーザソケットのインスタンスになる。

        /// <summary>
        /// ユーザセッション管理クラス
        /// </summary>
        public class SessionInfo
        {
            /// <summary>
            /// WebSocketSession
            /// </summary>
            public WebSocketSession Session { get; set; }

            /// <summary>
            /// User GUID
            /// </summary>
            public string Guid { get; set; }

            /// <summary>
            /// UserData
            /// </summary>
            public Interfaces.UserInfo UserInfo { get; set; }

            public SessionInfo(WebSocketSession session, string guid, Interfaces.UserInfo userinfo)
            {
                Session = session;
                Guid = guid;
                UserInfo = userinfo;
            }
        }
        /// <summary>
        /// ユーザセッション管理
        /// </summary>
        private List<SessionInfo> _sessions = new List<SessionInfo>();

サーバ起動処理

                /// WebSocket初期化
                _server = new SuperWebSocket.WebSocketServer();
                var rootConfig = new SuperSocket.SocketBase.Config.RootConfig();
                var serverConfig = new SuperSocket.SocketBase.Config.ServerConfig()
                {
                    Port = int.Parse(textBoxPort.Text),
                    Ip = "Any",
                    MaxConnectionNumber = 100,       /// 最大ユーザセッション数
                    Mode = SocketMode.Sync,
                    Name = "Chat Server"
                };

                _server.NewSessionConnected += HNewSessionConnected;
                _server.SessionClosed += HCloseSession;
                _server.NewDataReceived += HReceiveMessage;
                _server.Setup(rootConfig, serverConfig,SocketServerFactory.Instance);
                /// サーバ起動
                _server.Start();

だいぶイミフwだけど、なかなかシンプル。
なんたらConfigに関してはほぼわからない。MaxConnectionNumberは試してみたらちゃんと同時接続数であることを確認できた。あとのパラメータはあとで調べる。

以下の部分がユーザ接続の新規接続、クローズ、受信部分のイベント。NewDataReceivedはバイナリデータ受信時、文字列はNewMessageReceivedになる。このチャットソフトはMessagePackを使用しているのでバイナリしか扱わない。

_server.NewSessionConnected += HNewSessionConnected;
_server.SessionClosed += HCloseSession;
_server.NewDataReceived += HReceiveMessage;

新規接続(NewSessionConnected)
今回のチャットソフトはソケットのセッション確立後、ユーザからログイン情報（GUID、名前、アイコン)を送信してくるので、今回ほぼ未使用。

ユーザセッションクローズ(SessionClosed)
クローズされたSessionを受け取ることができるので、自分で管理している情報にアクセスし、全ユーザに抜けたユーザ情報を送信できるような仕組みにした。

HCloseSession

        /// <summary>
        /// セッションクローズ要求
        /// </summary>
        /// <param name="session"></param>
        /// <param name="e"></param>
        private void HCloseSession(SuperWebSocket.WebSocketSession session, SuperSocket.SocketBase.CloseReason e)
        {
            Trace.WriteLine("HReceiveMessage");

            /// 該当ユーザのログアウトを知らせる
            RemoveSession(session);
        }

RemoveSession関数

        /// <summary>
        /// セッションのログアウト処理
        /// </summary>
        /// <param name="session"></param>
        private void RemoveSession(SuperWebSocket.WebSocketSession session)
        {
            /// 接続中のユーザ情報を検索
            var sessionInfo = _sessions.Where(p => p.Session == session);
            if (sessionInfo.Any())
            {
                var delSession = sessionInfo.First();

                /// 抜けたユーザ以外にユーザが抜けたことを知らせる。
                var sendIf = new Interfaces.CmdInfo();
                sendIf._1CMDNO = (int)Interfaces.COMMAND.CMDLOGOUT;
                sendIf._2GUID = delSession.Guid;
                sendIf._3TIME = DateTime.Now.ToString();
                SendMessage(session, sendIf, SENDTO.OTHER);

                AddLog(LOGCMD.SYSTEM, "ADMIN", delSession.Guid + " IS LOGOUT");

                _sessions.Remove(sessionInfo.First());

            }
            SetUserCount();
        }

ソケット受信処理
例によってbyte[]をMessagePackでクラスに変換するので、解析超楽。もう僕の中でパターン化しつつある。

HReceiveMessage

        /// <summary>
        /// 受信要求
        /// </summary>
        /// <param name="session"></param>
        /// <param name="message"></param>
        private void HReceiveMessage(SuperWebSocket.WebSocketSession session, byte[] message)
        {
            Trace.WriteLine("HReceiveMessage");

            /// コマンド解析
            var rcvIf = message.MessagePack2Object<Interfaces.CmdInfo>();
            
            /// 送信要求を行う処理を格納
            var sendIfs = new List<Action>();
            /// ブロードキャスト(全てのユーザ)に送信する必要あるならフラグ
            var bcastFlag = false;
            /// 送信要求パケット作成
            var sendIf = new Interfaces.CmdInfo();
            sendIf._1CMDNO = rcvIf._1CMDNO;
            sendIf._2GUID = rcvIf._2GUID;
            sendIf._3TIME = DateTime.Now.ToString();

            AddLog(LOGCMD.RECEIVE, sendIf._2GUID, Enum.GetName(typeof(Interfaces.COMMAND), (Interfaces.COMMAND)rcvIf._1CMDNO));

            switch ((Interfaces.COMMAND)rcvIf._1CMDNO)
            {
                /// ユーザログイン要求
                case Interfaces.COMMAND.CMDLOGIN:
                    {
                        /// 受信パケットからユーザデータ(Interfaces.UserInfo)を取り出す。
                        var newsession = new SessionInfo(session, rcvIf._2GUID, rcvIf._4SENDDATA.MessagePack2Object<Interfaces.UserInfo>());
                        _sessions.Add(newsession);

                        AddLog(LOGCMD.SYSTEM, "ADMIN", newsession.Guid + " IS LOGIN");

                        /// 新規ユーザに現在ログイン中のすべてのユーザ情報を送信
                        Action alluserinfo = new Action(() =>
                            {
                                var sendCmdIf = new Interfaces.CmdInfo();
                                sendCmdIf._1CMDNO = (int)Interfaces.COMMAND.CMDALLUSERINFO;
                                sendCmdIf._2GUID = "ADMIN";
                                sendCmdIf._3TIME = DateTime.Now.ToString();
                                var infos = new List<Interfaces.UserInfo>();
                                _sessions.ForEach(p => infos.Add(p.UserInfo));
                                sendCmdIf._4SENDDATA = CommonLibrary.Extensions.ExtensionsClass.Object2MessagePack1Serializer(infos.ToArray());
                                SendMessage(session, sendCmdIf, SENDTO.ONE);
                            });

                        /// 送信処理に追加
                        sendIfs.Add(alluserinfo);
                        /// 全員に知らせる
                        bcastFlag = true;
                    }
                    break;
                /// チャットメッセージ受信
                case Interfaces.COMMAND.CMDSENDSTRING:

                    var rcvmsg = rcvIf._4SENDDATA.MessagePack2Object<Interfaces.MessageInfo>();
                    
                    /// 送信先が空なら宛先全員
                    if (rcvmsg._1TOGUID == "")
                    {
                        bcastFlag = true;
                    }
                    /// 宛先が特定ユーザ
                    else
                    {
                        /// 接続中ユーザへのメッセージ送信処理
                        var toUserInfo = _sessions.Where(p => p.Guid == rcvmsg._1TOGUID);
                        if (toUserInfo.Any())
                        {
                            Action sendMsg = new Action(() =>
                                {
                                    sendIf._4SENDDATA = rcvIf._4SENDDATA;

                                    /// 送信先->送信元ユーザ
                                    SendMessage(session, sendIf, SENDTO.ONE);
                                    /// 送信先->特定ユーザ
                                    SendMessage(toUserInfo.First().Session, sendIf, SENDTO.ONE);
                                });
                            sendIfs.Add(sendMsg);
                        }

                        /// 他の人には送らない
                        bcastFlag = false;
                    }
                    break;
                /// その他のコマンド
                default:
                    {
                        /// ブロードキャスト
                        bcastFlag = true;
                    }
                    break;
            }

            /// 全員宛のパケット作成
            if (bcastFlag)
            {
                var sendAll = new Action(() =>
                    {
                        sendIf._4SENDDATA = rcvIf._4SENDDATA;
                        SendMessage(session, sendIf, SENDTO.ALL);   /// 送信先=全員
                    });
                sendIfs.Add(sendAll);
            }

            /// 送信処理
            sendIfs.ForEach(p => p.DynamicInvoke());

            SetUserCount();
        }

これでもPythonのサーバと同じ処理のつもり。。。WebSocketsはどんなタイミングでも送信できるから楽だなぁ。
SendMessage関数がデータをパックして送信する処理になる。

SendMessage関数

        /// <summary>
        /// 送信処理
        /// </summary>
        /// <param name="session"></param>
        /// <param name="cmdif"></param>
        /// <param name="sendto"></param>
        /// <param name="delSession"></param>
        private void SendMessage(SuperWebSocket.WebSocketSession session, Interfaces.CmdInfo cmdif,SENDTO sendto,bool delSession = true)
        {
            /// 異常セッション回収用
            var delsession = new List<SuperWebSocket.WebSocketSession>();

            /// ログ文字生成
            var str = string.Format("{0}  FROM {1}  SENDTO {2}",
                Enum.GetName(typeof(Interfaces.COMMAND),(Interfaces.COMMAND)cmdif._1CMDNO),
                cmdif._2GUID,
                Enum.GetName(typeof(SENDTO),sendto));

            AddLog(LOGCMD.SEND, "ADMIN", str);
                
            /// 各セッション宛に送信処理を行う
            for (int i = 0; i < _sessions.Count(); i++)
            {
                var current = _sessions[i];

                /// 送信先指定==自分自身,current != 自分以外　OR
                /// 送信先指定==自分以外,current == 自分  の場合は抜ける
                if ( ((sendto == SENDTO.ONE) && (current.Session != session)) ||
                     ((sendto == SENDTO.OTHER) && (current.Session == session))
                    )
                {
                    continue;
                }

                try
                {
                    /// これが送信処理
                    current.Session.SendResponse(cmdif.Object2MessagePack());
                }
                catch
                {
                    /// 異常セッション回収
                    delsession.Add(current.Session);
                }

                /// 送信先指定==自分の場合は処理終了
                if (sendto == SENDTO.ONE) break;
            }

            /// 異常終了セッションをログアウト通知
            if (delsession.Any())
            {
                for (int i = 0; i < delsession.Count; i++)
                {
                    try
                    {
                        RemoveSession(delsession[i]);
                    }
                    catch { }
                }
            }
        }

送信処理は特定ユーザのみにデータを送信する場合があるので、ちょっとわかりづらいコードになってしまったなう。SuperWebSocket.WebSocketSessionクラスのSendResponse()が送信処理本体で、byte[]とstringの2種類がある。

実行画面

まとめ
さくっとできたところはよかっと思います。

今回作成したソースはこちらから。