前回のチュートリアルでは、ログイン／ログアウトの処理を作成しました。

このチュートリアルではログインするまでは、不定期にデータを受信する必要性はあまりないかと考えています。よってログインが成功した時点で読み取りを行うスレッドと書き込みを行うスレッドを起動するようにしてみたいと思います。同様にログアウトを行った場合、または送受信において例外が発生した場合には読み取り、および書き込みスレッドを停止するようにしたいと思います。

今回のポイントとしては、Sample2DRPG.io名前空間に追加した３つのクラスを確認して欲しいと思います。この３つのクラスは今後、このチュートリアルで話を進める上でのエンジンとなるクラスです。ログインのタイミングでスレッドを起動して、ログアウトをする際にスレッドを停止する処理を記述しています。

また、今回のチュートリアルでは、サーバーから受け取った Command オブジェクトをフォーム上のリストボックスに追加しています。この処理をSTEP.02に当てはめるとネットワークを経由して他のプレイヤーにアクションを伝えることができると思います。

※今回もフォームクラスで色々やっていますが、あまり重要ではありません。スレッドの起動と終了あたりを中心にご確認ください。

• まずはじめに Sample2DRPG_server フォルダに含まれる start.bat をダブルクリックしてください。（サーバーアプリケーションのコンパイルと起動を行います。）
• 次に ClientEmulator に含まれる ClientEmulator.sln をダブルクリックして Visual Studio 2005 Express Edition を起動します。
• メニューバーの デバッグ(D) から デバッグ開始(S) を選択してデバッグを行ってください。（F5でもOK）
• デバッグ開始を行うと ClientEmulator\ClientEmulator\bin\Debug フォルダに ClientEmulator.exe という実行ファイルが作成されます。
• 作成された ClientEmulator.exe を複数起動してください。
• 起動したクライアントをそれぞれ、Connect→ログインとしてログインしてください。（パスワードは未チェックのため空白でOK!!）
• 複数のクライアントがログインした状態で Up、Right、Down、Left、Stopのボタンを押すと各クライアントのリストボックスが更新されます。
• また、ボタンを押したときのサーバーの動きをコマンドプロンプトを見ながら確認してください。

STEP.05 のソリューション設定は次の図のようになっています。

io (ネットワークI/Oを別スレッドで実行するクラスを作成します。)

• NetworkThread.cs
• ReaderThread.cs
• WriterThread.cs

message (キーボード情報、ネットワーク切断要求のメッセージを追加します。)

• CloseRequest.cs
• KeyDownMessage.cs
• KeyUpMessage.cs

message (キーボードのキーコードをマッピングするクラスを作成します。)

• KeyUpMessage.cs

package sample2DRPG.command;

public class AddPlayerCommand implements IPlayerCommand
{
    /**
     * メッセージ識別番号
     */
    public short getCode()
    {
        return 7;
    }

    // メッセージの文字列表現
    public String toString()
    {
        StringBuilder buf = new StringBuilder();
        buf.append("name=");
        buf.append(this.getClass().getSimpleName());
        buf.append(",code=");
        buf.append(this.getCode());
        buf.append(",id=");
        buf.append(this.id);
        buf.append(",name=");
        buf.append(this.name);
        buf.append(",type=");
        buf.append(this.type);
        return buf.toString();
    }

    // プレイヤー識別番号
    public int id;

    // プレイヤー名称
    public String name;

    // キャラクタタイプ
    public int type;
}

package sample2DRPG.command;

public class RemPlayerCommand implements IPlayerCommand
{
    /**
     * メッセージ識別番号
     */
    public short getCode()
    {
        return 8;
    }

    // メッセージの文字列表現
    public String toString()
    {
        StringBuilder buf = new StringBuilder();
        buf.append("name=");
        buf.append(this.getClass().getSimpleName());
        buf.append(",code=");
        buf.append(this.getCode());
        buf.append(",id=");
        buf.append(this.id);
        return buf.toString();
    }

    // プレイヤー識別番号
    public int id;
}

package sample2DRPG.command;

public class UpdPlayerCommand implements IPlayerCommand
{
    /**
     * メッセージ識別番号
     */
    public short getCode()
    {
        return 9;
    }

    // メッセージの文字列表現
    public String toString()
    {
        StringBuilder buf = new StringBuilder();
        buf.append("name=");
        buf.append(this.getClass().getSimpleName());
        buf.append(",code=");
        buf.append(this.getCode());
        buf.append(",id=");
        buf.append(this.id);
        buf.append(",action=");
        buf.append(this.action);
        return buf.toString();
    }

    // プレイヤー識別番号
    public int id;

    // アクション
    public int action;
}

package sample2DRPG;

import java.io.EOFException;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.channels.SocketChannel;

import sample2DRPG.message.IMessage;
import sample2DRPG.message.MessageManager;

public class Connection {

    private static final int MIN_ID = 100000;
    private static final int MAX_ID = 199999;

    /**
     * コンストラクタ
     */
    public Connection(SocketChannel channel)
    {
        connid = nextId();
        this.channel = channel;
        buffer = ByteBuffer.allocate(8192);            
        buffer.order(ByteOrder.LITTLE_ENDIAN);
    }
    
    /**
     * チャネルからバイト配列を取得
     * @param channel
     * @return
     * @throws Exception
     */
    public boolean read() throws EOFException, IOException
    {
        // リミットを最大に設定
        buffer.limit(buffer.capacity());
        
        int bytesread = channel.read(buffer);

        if (bytesread == -1) {
            throw new EOFException();
        }
        
        if (buffer.position() < 6)
        {
            // 必要最低限のデータが取得できていない
            return false;
        }
        buffer.flip();
        int length = buffer.getInt(0);
        if (length > buffer.limit()) {
            // １メッセージ分のデータが取得できていない
            buffer.position(buffer.limit());
            return false;
        }
        return true;
    }
    
    /**
     * メッセージ読み込み
     */
    public IMessage receive()
    {
        buffer.mark();
        
        if (buffer.remaining() < 6) {
            // 必要最低限(length + code)の容量が確保できていない
            return null;
        }
        
        // データ長を取得
        int length = buffer.getInt();
        if (length > buffer.remaining() + 4) {
            // １メッセージ分のデータが取得できていない
            buffer.reset();
            return null;
        }
        
        // メッセージマネージャからマスタメッセージを取得
        IMessage msg = MessageManager.get(buffer);

        // デバッグ用
        if (msg != null)
            System.out.println("RECV <<< " + connid + " [" + msg + "]");
        return msg;
    }
    
    /**
     * メッセージ読み込み後の処理
     */
    public void endReceive()
    {
        // 現在のポジションを取得
        int position = buffer.position();

        // 新しい末尾を算出
        int newLimit = buffer.limit() - position;

        // バッファを切り詰める
        buffer.compact();
        
        // ポジションを末尾へ
        buffer.position(newLimit);
    }

    /**
     * コネクションの文字列表現
     */
    public String toString()
    {
        return "connid=" + connid;
    }
    
    /**
     * 次のコネクション識別番号を発行する
     * @return コネクション識別番号
     */
    private static int nextId()
    {
        currentId++;
        if (currentId > MAX_ID) {
            currentId = MIN_ID;
        }
        return currentId;
    }

    // コネクション識別番号
    protected int connid;
    
    // コネクションごとのチャネル
    protected SocketChannel channel;

    // 読み取り用バッファ
    protected ByteBuffer buffer;

    // コネクション識別番号の連番管理
    protected static int currentId = MIN_ID;
}

package sample2DRPG;

import java.nio.*;
import java.io.*;
import java.nio.channels.*;
import java.net.*;
import java.util.*;

import sample2DRPG.command.*;
import sample2DRPG.message.*;
import sample2DRPG.util.*;

public class Server implements Runnable
{
    // デフォルトポート番号
    public static final int DEFAULT_PORT = 65000;

    // キー選択処理のタイムアウトを定義（ミリ秒）
    protected static final int SELECT_TIME = 100;
    
    // ポート番号
    protected int port;

    // バッファ
    protected ByteBuffer buffer;
    
    // 終了予定の接続
    protected LinkedList<SelectionKey> closeq 
        = new LinkedList<SelectionKey>();

    // ログインユーザーリスト
    protected HashMap<Integer, Connection> usrmap 
        = new HashMap<Integer, Connection>();

    // セレクタ
    private Selector selector;
    
    // サーバーソケットチャネル
    private ServerSocketChannel serverSocketChannel;

    /**
     * コンストラクタ
     * コンストラクタでポートを指定しなかった場合はデフォルトポート番号(65000)が使用されます。
     */
    public Server()
    {
        this(DEFAULT_PORT);
    }
    
    public Server(int port)
    {
        this.port = port;
                
        // 書き込み用バッファを初期化
        buffer = ByteBuffer.allocate(8192);            
        buffer.order(ByteOrder.LITTLE_ENDIAN);

        // メッセージマネージャ生成
        MessageManager.create();
    }

    /**
     * メイン処理
     */
    public void run()
    {
        try {
            selector = Selector.open();
    
            serverSocketChannel = ServerSocketChannel.open();
            serverSocketChannel.configureBlocking(false);
            serverSocketChannel.socket().bind(new InetSocketAddress(port));
            serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);

            System.out.println("Startup server [port=" + port + "]");
            
            // ループ処理
            while (true) {
                selector.select(SELECT_TIME);
                Iterator<SelectionKey> it = selector.selectedKeys().iterator();
                while (it.hasNext()) {
                    SelectionKey key = it.next();
                    it.remove();

                    if (key.isAcceptable()) {
                        System.out.println("ACCEPTABLE");
                        handleAcceptable(key);
                    }

                    if (key.isConnectable()) {
                        System.out.println("CONNECTABLE");
                    }

                    if (key.isReadable()) {
                        System.out.println("READABLE");
                        handleReadable(key);
                    }
                    
                    if (key.isWritable()) {
                        System.out.println("WRITABLE");
                    }

                    while ((key = closeq.poll()) != null) {
                        close(key);
                    }
                }
            }
            
        } catch(IOException ioe) {
            System.out.println("例外が発生したため終了します。[exception=" + ioe + "]");
        }
    }
    
    /**
     * 接続受入
     */
    protected void handleAcceptable(SelectionKey key)
    {
        try {
            SocketChannel channel = serverSocketChannel.accept();
            channel.configureBlocking(false);
    
            // 読み取り可能なキーを登録してコネクションを付属させる
            SelectionKey selKey = channel.register(selector, SelectionKey.OP_READ);
            selKey.attach(new Connection(channel));

        } catch(IOException e) {
            System.out.println("例外が発生しました。[exception=" + e + "]");
            e.printStackTrace();
        }
    }
    
    /**
     * 読込処理
     */
    protected void handleReadable(SelectionKey key)
    {
        try {
            Connection conn = (Connection)key.attachment();

            // チャネルからメッセージの受信を行います。
            // １メッセージ分のデータが受信できた場合にはバイト配列からメッセージ
            // を生成します。受信したデータが１メッセージ未満の場合は、次回読み込
            // み後に再度、１メッセージ分の受信ができたかを確認します。読み込み中
            // のバッファは接続ごとにConnectionクラスで管理されます。
            if (conn.read())
            {
                IMessage msg;
                while((msg = conn.receive()) != null)
                {
                    if (msg instanceof KeyDownMessage) {
                        KeyDownMessage keydown = (KeyDownMessage) msg;
                        
                        // プレイヤー更新コマンド発行
                        UpdPlayerCommand cmd = new UpdPlayerCommand();
                        cmd.id = conn.connid;
                        
                        // 押下されたキーによりアクションを設定
                        if (keydown.key == Keyboard.Up) {
                            cmd.action = 0;
                        } else if (keydown.key == Keyboard.Right) {
                            cmd.action = 1;
                        } else if (keydown.key == Keyboard.Down) {
                            cmd.action = 2;
                        } else if (keydown.key == Keyboard.Left) {
                            cmd.action = 3;                            
                        }
                        
                        sendToLoginUser(cmd);
                        
                    } else if (msg instanceof KeyUpMessage) {
                        // プレイヤー更新コマンド発行
                        UpdPlayerCommand cmd = new UpdPlayerCommand();
                        cmd.id = conn.connid;
                        cmd.action = -1;
                        sendToLoginUser(cmd);
                        
                    } else if (msg instanceof LoginRequest) {
                        LoginRequest req = (LoginRequest) msg;
                        
                        // ユーザーマップを更新
                        usrmap.put(conn.connid, conn);
                        
                        // ログイン応答を返す
                        LoginResponse res = new LoginResponse();
                        res.connid = conn.connid;
                        send(conn, res);
                        
                        // プレイヤー追加コマンド発行
                        // 本来データベース等から読み込んだ情報を返すべきですが、
                        // チュートリアルでは、キャラクタタイプは便宜上、0〜4の
                        // 数値を返しています。
                        AddPlayerCommand cmd = new AddPlayerCommand();
                        cmd.id = conn.connid;
                        cmd.name = req.username;
                        cmd.type = conn.connid % 5;
                        sendToLoginUser(cmd);
                        
                    } else if (msg instanceof LogoutRequest) {                        
                        // ユーザーマップを更新
                        usrmap.remove(conn.connid);

                        // ログイン応答を返す
                        LogoutResponse res = new LogoutResponse();
                        send(conn, res);

                        // プレイヤー削除コマンド発行
                        RemPlayerCommand cmd = new RemPlayerCommand();
                        cmd.id = conn.connid;
                        sendToLoginUser(cmd);
                        
                    } else if (msg instanceof CloseRequest) {
                        System.out.println("接続を閉じます。[" + conn + "]");
                        closeq.add(key);
                    }
                }
                conn.endReceive();
            }

        } catch(EOFException eof) {
            System.out.println("接続が切れました。[" + key.attachment() + "]");
            closeq.add(key);

        } catch(IOException ioe) {
            ioe.printStackTrace();
            closeq.add(key);

        } catch(Exception e) {
            System.out.println("例外が発生しました。[exception=" + e + "]");
            e.printStackTrace();
        }
    }
    
    /**
     * ログイン中のユーザーにコマンドを発行
     * @param cmd
     */
    protected void sendToLoginUser(IPlayerCommand cmd)
    {
        Iterator<Connection> it = usrmap.values().iterator();
        while (it.hasNext()) {
            try {
                Connection conn = it.next();
                if (conn.channel.isConnected()) {
                    send(conn, cmd);
                } else {
                    it.remove();
                }
                
            } catch(IOException ioe) {
                System.out.println("例外が発生しました。[exception=" + ioe + "]");
                ioe.printStackTrace();
                it.remove();
            }
        }
    }

    /**
     * メッセージ送信
     * @param channel 送信対象
     * @param msg 送信するメッセージ
     */
    protected void send(Connection conn, IMessage msg) throws IOException
    {
        MessageManager.set(buffer, msg);
        buffer.rewind();
        conn.channel.write(buffer);
        System.out.println("SEND >>> " + conn.connid + " [" + msg + "]");
            
    }

    /**
     * 接続終了
     */
    protected void close(SelectionKey key)
    {
        try {
            key.cancel();
            key.channel().close();

        } catch(IOException e) {
            System.out.println("例外が発生しました。[exception=" + e + "]");
            e.printStackTrace();
        }
    }
    
    /**
     * エントリーポイント
     * @param argv
     * @throws Exception
     */
    public static void main(String[] argv) throws Exception
    {
        Server server = new Server();
        server.run();
    }
}

using System;
using System.Collections.Generic;
using System.Text;
using System.Windows.Forms;

namespace ClientEmulator.command
{
    public class AddPlayerCommand : PlayerCommand
    {
        // メッセージ識別コード
        public short Code
        {
            get { return 7; }
        }

        /// <summary>
        /// コンストラクタ
        /// </summary>
        public AddPlayerCommand() { }

        public AddPlayerCommand(int id, string name, int type)
        {
            this.id = id;
            this.name = name;
            this.type = type;
        }

        /// <summary>
        /// コマンド実行
        /// </summary>
        public void Execute(ListBox list)
        {
            // エミュレータではリストボックスに
            // コマンドを追加しています。
            list.Items.Add(this);
            list.SelectedItem = this;
        }

        /// <summary>
        /// 文字列表現
        /// </summary>
        /// <returns></returns>
        public override string ToString()
        {
            StringBuilder buf = new StringBuilder();
            buf.Append("ADD [code=");
            buf.Append(this.Code);
            buf.Append(",id=");
            buf.Append(this.id);
            buf.Append(",name=");
            buf.Append(this.name);
            buf.Append(",type=");
            buf.Append(this.type);
            buf.Append("]");
            return buf.ToString();
        }

        // プレイヤー識別番号
        public int id;

        // プレイヤー名称
        public String name;

        // キャラクタタイプ
        public int type;
    }
}

using System;
using System.Collections.Generic;
using System.Text;
using System.Windows.Forms;

namespace ClientEmulator.command
{
    public class RemPlayerCommand : PlayerCommand
    {
        // メッセージ識別コード
        public short Code
        {
            get { return 8; }
        }

        /// <summary>
        /// コンストラクタ
        /// </summary>
        public RemPlayerCommand() { }

        public RemPlayerCommand(int id)
        {
            this.id = id;
        }

        /// <summary>
        /// コマンド実行
        /// </summary>
        public void Execute(ListBox list)
        {
            // エミュレータではリストボックスに
            // コマンドを追加しています。
            list.Items.Add(this);
            list.SelectedItem = this;
        }

        /// <summary>
        /// 文字列表現
        /// </summary>
        /// <returns></returns>
        public override string ToString()
        {
            StringBuilder buf = new StringBuilder();
            buf.Append("REM [code=");
            buf.Append(this.Code);
            buf.Append(",id=");
            buf.Append(this.id);
            buf.Append("]");
            return buf.ToString();
        }

        // プレイヤー識別番号
        public int id;
    }
}

using System;
using System.Collections.Generic;
using System.Text;
using System.Windows.Forms;

namespace ClientEmulator.command
{
    public class UpdPlayerCommand : PlayerCommand
    {
        // メッセージ識別コード
        public short Code
        {
            get { return 9; }
        }

        /// <summary>
        /// コンストラクタ
        /// </summary>
        public UpdPlayerCommand() { }

        public UpdPlayerCommand(int id, int action)
        {
            this.id = id;
            this.action = action;
        }

        /// <summary>
        /// コマンド実行
        /// </summary>
        public void Execute(ListBox list)
        {
            // エミュレータではリストボックスに
            // コマンドを追加しています。
            list.Items.Add(this);
            list.SelectedItem = this;
        }

        /// <summary>
        /// 文字列表現
        /// </summary>
        /// <returns></returns>
        public override string ToString()
        {
            StringBuilder buf = new StringBuilder();
            buf.Append("UPD [code=");
            buf.Append(this.Code);
            buf.Append(",id=");
            buf.Append(this.id);
            buf.Append(",action=");
            buf.Append(this.action);
            buf.Append("]");
            return buf.ToString();
        }

        // プレイヤー識別番号
        public int id;

        // アクション
        public int action;
    }
}

using System;
using System.Collections.Generic;
using System.IO;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading;
using ClientEmulator.command;
using ClientEmulator.messsage;

namespace ClientEmulator.io
{
    abstract class NetworkThread
    {
        /// <summary>
        /// 実行状態
        /// </summary>
        public bool isRunning
        {
            get
            {
                lock (locker) { return running; }
            }
        }

        /// <summary>
        /// 接続状態
        /// </summary>
        public bool isConnect
        {
            get
            {
                if (tcpClient == null) return false;
                return tcpClient.Connected;
            }
        }

        /// <summary>
        /// ループ処理を開始
        /// </summary>
        public void start()
        {
            running = true;
            thread = new Thread(new ThreadStart(run));
            thread.IsBackground = true;
            thread.Start();
        }

        /// <summary>
        /// スレッドを終了
        /// </summary>
        public void shutdown()
        {
            lock (locker)
            {
                running = false;
            }
        }

        /// <summary>
        /// スレッドを結合
        /// </summary>
        public void join(int timeout)
        {
            // スレッドが null の場合、終了する
            if (thread == null) return;

            // スレッドが止まっていたら終了する
            if (thread.ThreadState == ThreadState.Stopped) return;

            if (this.thread != Thread.CurrentThread)
            {
                // タイムアウト設定指定
                thread.Join(timeout);
            }
        }

        /// <summary>
        /// スレッドを実行
        /// </summary>
        protected void run()
        {
            try
            {
                while (running) process();
            }
            catch (Exception e)
            {
                // エラーイベントを発生
                if (ErrorProcess != null)
                {
                    ErrorProcess(this, e);
                }
                shutdown();
            }
            Console.WriteLine("shutdown [name={0}]",this.GetType().Name);
        }

        /// <summary>
        /// メイン処理
        /// </summary>
        protected abstract void process();

        // フォーム
        protected ClientEmulator emu;

        // 排他ロック用オブジェクト
        protected object locker = new object();

        // 実行状態
        protected bool running = false;

        // ループ処理を実行するスレッド
        protected Thread thread;

        // バッファストリーム
        protected BufferedStream stream;

        // TCPクライアント
        protected TcpClient tcpClient;

        // 送受信処理でエラーが発生した場合のデリゲート
        public delegate void ErrorProcessDelegate(NetworkThread sender, Exception e);

        // 送受信スレッドで発生したエラー
        public static ErrorProcessDelegate ErrorProcess;

        // バッファ初期サイズ
        public static readonly int INITIAL_BUFFER_SIZE = 8192;
    }
}

using System;
using System.Collections.Generic;
using System.IO;
using System.Net;
using System.Net.Sockets;
using System.Text;
using ClientEmulator.command;
using ClientEmulator.messsage;

namespace ClientEmulator.io
{
    class ReaderThread : NetworkThread
    {
        public ReaderThread(ClientEmulator emu, TcpClient tcpClient)
        {
            // フォーム設定
            this.emu = emu;

            // TCPクライアント設定
            this.tcpClient = tcpClient;

            // ストリーム作成
            stream = new BufferedStream(tcpClient.GetStream(), INITIAL_BUFFER_SIZE);
        }
        
        /// <summary>
        /// 受信処理
        /// </summary>
        public IMessage receive()
        {
            // 接続されていない場合、受信処理を行わない
            if (!isConnect) return null;

            int bytesread = stream.Read(header, 0, header.Length);

            // Read が 0 を返すのは、ストリーム内にそれ以上データがない場合とそれ以上読
            // み込み可能な対象を予期していない場合だけです。[下記のURLを参照]
            // http://msdn.microsoft.com/ja-jp/library/system.io.stream.read(VS.80).aspx
            if (bytesread == 0)
            {
                shutdown();
                return null;
            }
            // データ長をデコード
            int length = BitConverter.ToInt32(header, 0);

            // メッセージ識別コード以降を取得
            byte[] dst = new byte[length - header.Length];
            stream.Read(dst, 0, length - header.Length);

            // メッセージを取得して返す
            return MessageManager.get(dst);
        }

        /// <summary>
        /// メイン処理
        /// </summary>
        protected override void process()
        {
            IMessage msg = receive();

            if (msg is PlayerCommand)
            {
                // プレイヤーコマンド処理
                emu.dispatchCommand(msg as PlayerCommand);
            }
        }

        private byte[] header = new byte[4];
    }
}

using System;
using System.Collections.Generic;
using System.IO;
using System.Net;
using System.Net.Sockets;
using System.Text;
using System.Threading;
using ClientEmulator.command;
using ClientEmulator.messsage;

namespace ClientEmulator.io
{
    class WriterThread : NetworkThread
    {
        /// <summary>
        /// コンストラクタ
        /// </summary>
        public WriterThread(ClientEmulator emu, TcpClient tcpClient)
        {
            // フォーム設定
            this.emu = emu;

            // TCPクライアント設定
            this.tcpClient = tcpClient;

            // 送信リストを生成
            msgq = new Queue<IMessage>();

            // ストリーム作成
            stream = new BufferedStream(tcpClient.GetStream(), INITIAL_BUFFER_SIZE);
        }

        /// <summary>
        /// 送信リストにメッセージを追加
        /// </summary>
        /// <param name="msg"></param>
        public void postMessage(IMessage msg)
        {
            lock (msgq) msgq.Enqueue(msg);
        }

        /// <summary>
        /// ログインの送信
        /// スレッドが起動していない状態で実行可能
        /// </summary>
        public void loginRequest(LoginRequest msg)
        {
            // タイムスタンプを更新
            timestamp = DateTime.Now.Ticks;

            send((IMessage)msg);

            stream.Flush();
        }

        /// <summary>
        /// ログアウトの送信
        /// スレッドが起動していない状態で実行可能
        /// </summary>
        public void logoutRequest(LogoutRequest msg)
        {
            lock (msgq)
            {
                // 送信キューにメッセージ登録
                msgq.Enqueue(msg);

                // メッセージを全て送信
                process();
            }
        }

        /// <summary>
        /// 送信処理
        /// </summary>
        protected void send(IMessage msg)
        {
            // 接続されていない場合、送信処理を行わない
            if (!isConnect) return;

            byte[] dst = MessageManager.set(msg);
            if (dst != null)
            {
                stream.Write(dst, 0, dst.Length);
                Console.WriteLine("Send [byteArray={0}]", BitConverter.ToString(dst));
            }
        }

        /// <summary>
        /// メイン処理
        /// </summary>
        protected override void process()
        {
            lock (msgq)
            {
                // メッセージ分書き込み処理を行う
                while (msgq.Count > 0)
                {
                    IMessage msg = msgq.Dequeue();
                    send(msg);
                }

                // 全てのメッセージをバッファ書き込んだ後
                // Flushを行いサーバーへ転送を行う。
                stream.Flush();
            }

            // 指定したタイミングまで停止
            sleep();
        }

        /// <summary>
        /// スレッド停止
        /// </summary>
        private void sleep()
        {
            // 前回実行からの処理時間を取得
            int time = (int)(DateTime.Now.Ticks - timestamp);

            // 設定した時間より短い場合、その差分を停止する
            if (SLEEP_TIME > time) Thread.Sleep(SLEEP_TIME - time);

            // タイムスタンプ更新
            timestamp = DateTime.Now.Ticks;
        }

        // 送信メッセージキュー
        protected Queue<IMessage> msgq;

        // 最終送信タイムスタンプ
        protected long timestamp;

        // １秒間に最大で 20回 の送信機会を設ける（50ミリ秒待機）
        protected static readonly int SLEEP_TIME = 50;
    }
}

次回予定