技術めも

ラウンド編。

また、websocketの処理だけ書く。

package trmpj.ws;

import java.util.concurrent.ConcurrentHashMap;

import javax.websocket.CloseReason;
import javax.websocket.OnClose;
import javax.websocket.OnMessage;
import javax.websocket.OnOpen;
import javax.websocket.Session;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;

import trmpj.TPair;
import trmpj.TReq;
import trmpj.TSession;
import trmpj.TWs;
import trmpj.req.TReqEnd;
import trmpj.req.TReqGame;
import trmpj.req.TReqPrepared;
import trmpj.res.TResEnded;
import trmpj.res.TResError;
import trmpj.res.TResGame;
import trmpj.res.TResInvalid;
import trmpj.res.TResNone;
import trmpj.res.TResStart;
import trmpj.res.TResWait;

/**
 * 対戦用処理
 *
 */
@ServerEndpoint(value = "/ws/round/{pairId}")
public class TWsRound extends TWs {

    /**
     * pairId => TPairのマップ
     */
    private static ConcurrentHashMap<String, TPair> pairs = new ConcurrentHashMap<String, TPair>();

    /**
     * pairId => Threadのマップ
     */
    private static ConcurrentHashMap<String, Thread> threads = new ConcurrentHashMap<String, Thread>();

    /**
     * openハンドラ
     * @param pairId
     * @param session
     */
    @OnOpen
    public void onOpen(@PathParam("pairId") String pairId, Session session) {
        // セッションにアクセスIDを格納する
        String accessdId = generateAccessId();
        session.getUserProperties().put(KEY_ACCESS_ID, accessdId);

        // アクセスIDを、onMessage, onCloseで取得するときの関連を付ける
        // ※sessionsのキー側はsessionでもよいのですが、
        // runnableからの参照時に、セッションが破棄されていると困るので、
        // あえて文字列にしている
        TSession ts = new TSession(session);
        sessions.putIfAbsent(accessdId, ts);
    }

    /**
     * messageハンドラ
     * @param pairId
     * @param session
     * @param message
     */
    @OnMessage
    public void onMessage(@PathParam("pairId") String pairId, Session session, String message) {
        // アクセスIDを取得
        String accessdId = (String) session.getUserProperties().get(KEY_ACCESS_ID);
        TSession ts = null;

        try {
            // アクセスIDをもとに、openハンドラで登録したTSessionを呼び出す
            if (!sessions.containsKey(accessdId)) {
                throw new Exception();
            }
            ts = sessions.get(accessdId);

            // リクエストを解析
            TReq req = ts.receiveMessage(message, TReq.class);
            if (req == null) {
                throw new Exception();
            }

            // ペアIDに応じたペア情報をマップから取り出す
            if (!pairs.containsKey(pairId)) {
                pairs.putIfAbsent(pairId, new TPair(pairId));
            }
            TPair pair = pairs.get(pairId);

            // ペア情報に排他制御をかける
            // ※重要
            synchronized (pair) {
                if (req.command.equals(TReq.COMMAND_PREPARED)) {
                    // preparedコマンド
                    TReqPrepared reqPrepared = ts.receiveMessage(message, TReqPrepared.class);
                    if (reqPrepared == null) {
                        throw new Exception();
                    }

                    // 既にペアが揃った状態ならここで終了
                    if (pair.isPrepared()) {
                        throw new Exception();
                    }

                    // 自分の情報をセッションに追加
                    ts.setUser(reqPrepared.user);

                    // ペア情報に自分の情報を追加
                    pair.addTSession(ts);

                    // 同一ユーザIDはエラー
                    if (pair.isDuplicated()) {
                        throw new Exception();
                    }

                    // 2人そろった状態なら、2人にstartを返す
                    if (pair.isPrepared()) {
                        // スレッドを止める
                        threads.remove(pairId);

                        TResStart resStart = new TResStart();
                        resStart.users.addAll(pair.getUsers());
                        pair.sendMessage(resStart);
                        return;
                    }

                    // 揃っていない状態なので、自分にwaitを返す
                    TResWait resWait = new TResWait();
                    ts.sendMessage(resWait);

                    // 別スレッドで、相手がくるまで数秒待機する
                    if (!threads.containsKey(pairId)) {
                        threads.putIfAbsent(pairId, new Thread(new TRunnable(pairId)));
                        threads.get(pair).start();
                    }
                    return;

                } else if (req.command.equals(TReq.COMMAND_GAME)) {
                    // gameコマンド
                    TReqGame reqGame = ts.receiveMessage(message, TReqGame.class);
                    if (reqGame == null) {
                        throw new Exception();
                    }

                    // 同一ユーザIDはエラー
                    if (pair.isDuplicated()) {
                        throw new Exception();
                    }

                    // 揃った状態でない、または終了状態なら何もしない
                    if (!pair.isPrepared() || pair.isEnded()) {
                        throw new Exception();
                    }

                    // 2人に返す
                    TResGame resGame = new TResGame();
                    resGame.useId = ts.getUser().userId;
                    resGame.data = reqGame.data;
                    pair.sendMessage(resGame);
                    return;

                } else if (req.command.equals(TReq.COMMAND_END)) {
                    // endコマンド
                    TReqEnd reqEnd = ts.receiveMessage(message, TReqEnd.class);
                    if (reqEnd == null) {
                        throw new Exception();
                    }

                    // 同一ユーザIDはエラー
                    if (pair.isDuplicated()) {
                        throw new Exception();
                    }

                    // ペアを終了状態にする
                    pair.setEnded(true);

                    // 自分に返す
                    TResEnded resEnded = new TResEnded();
                    ts.sendMessage(resEnded);
                    return;

                }
            }

        } catch (Exception e) {
            e.printStackTrace();
        }

        // エラー処理
        if (ts != null) {
            TResError resError = new TResError();
            ts.sendMessage(resError);
        }
    }

    /**
     * closeハンドラ
     * @param pairId
     * @param session
     * @param reason
     */
    @OnClose
    public void onClose(@PathParam("pairId") String pairId, Session session, CloseReason reason) {
        // アクセスIDを取得
        String accessdId = (String) session.getUserProperties().get(KEY_ACCESS_ID);
        TSession ts = null;

        try {
            // アクセスIDをもとに、openハンドラで登録したTSessionを呼び出す
            if (!sessions.containsKey(accessdId)) {
                throw new Exception();
            }
            ts = sessions.get(accessdId);

            // ペアIDに応じたペア情報をマップから取り出す
            if (!pairs.containsKey(pairId)) {
                throw new Exception();
            }
            TPair pair = pairs.get(pairId);

            // ペア情報に排他制御をかける
            // ※重要
            synchronized (pair) {
                // 自分の情報削除
                pair.removeTSession(ts);

                // 空になっていたら削除
                if (pair.isEmpty()) {
                    pairs.remove(pairId);
                    return;
                }

                // game中で、endでない場合は、残っている方にnoneを送信
                if (pair.isPrepared() && !pair.isEnded()) {
                    TResNone resNone = new TResNone();
                    pair.sendMessage(resNone);
                    return;
                }
            }

        } catch (Exception e) {
            e.printStackTrace();
        }

        // セッション情報も破棄する
        sessions.remove(accessdId);
    }

    /**
     * 2人が揃うまで待つ別スレッド
     * @author furuta
     *
     */
    public class TRunnable implements Runnable {

        /**
         * ペアID
         */
        private String pairId;

        /**
         * コンストラクタ
         * @param pairId
         */
        public TRunnable(String pairId) {
            this.pairId = pairId;
        }

        /**
         * 処理
         */
        public void run() {
            try {
                // 10秒まつ
                Thread.sleep(10000);

                // ペアを取得
                if (!pairs.containsKey(pairId)) {
                    throw new Exception();
                }
                TPair pair = pairs.get(pairId);

                synchronized (pair) {
                    if (!pair.isPrepared()) {
                        // 揃っていない状態であれば、残っている人にinvalidを返す
                        TResInvalid resInvalid = new TResInvalid();
                        pair.sendMessage(resInvalid);
                    }
                }

            } catch (Exception e) {
                e.printStackTrace();
            }

            // スレッドを削除
            threads.remove(pairId);
        }
    }
}

ポイントはsynchronizedでしっかり排他制御をかけること。
これをやらないとだめだぜ。。

ソースコードはちゃんとテストしてないけど、
処理の流れはわかるはず。。

とりあえず、作成したコードをzipで固めておきます。

ダウンロード

これで、websocketは終了。

nodeは書く気力なし。。