websocket support init

package.json

@@ -19,7 +19,8 @@
     "inotify": "*",
     "request": "*",
     "mongoose": "*",
-        "bunyan": "*"
+    "bunyan": "*",
+    "websocket": "~1.0.8"
   },
   "license": "GPLv3",
   "main": "ygopro-server.js",

room.coffee

@@ -54,7 +54,9 @@ class Room
     @status = 'starting'
     @established = false
     @watcher_buffers = []
+    @watcher_stanzas = []
     @watchers = []
+    @ws_watchers = []
     Room.all.push this
 
     @hostinfo =
@@ -218,5 +220,4 @@ class Room
         @process.kill()
         this.delete()
 
-
 module.exports = Room
\ No newline at end of file

room.js

@@ -74,7 +74,9 @@
       this.status = 'starting';
       this.established = false;
       this.watcher_buffers = [];
+      this.watcher_stanzas = [];
       this.watchers = [];
+      this.ws_watchers = [];
       Room.all.push(this);
       this.hostinfo = {
         lflist: 0,

ygopro-server.coffee

@@ -12,6 +12,7 @@ _.str = require 'underscore.string'
 _.mixin(_.str.exports());
 
 Inotify = require('inotify').Inotify
+WebSocketServer = require('websocket').server
 request = require 'request'
 
 bunyan = require 'bunyan'
@@ -150,6 +151,7 @@ net.createServer (client) ->
       else
         if stoc_buffer.length >= 2 + stoc_message_length
           #console.log "STOC", ygopro.constants.STOC[stoc_proto]
+          stanzas = stoc_proto
           if ygopro.stoc_follows[stoc_proto]
             b = stoc_buffer.slice(3, stoc_message_length - 1 + 3)
             if struct = ygopro.structs[ygopro.proto_structs.STOC[ygopro.constants.STOC[stoc_proto]]]
@@ -255,14 +257,42 @@ ygopro.stoc_follow 'JOIN_GAME', false, (buffer, info, client, server)->
       }
       ygopro.ctos_send watcher, 'HS_TOOBSERVER'
 
+    watcher.ws_buffer = new Buffer(0)
+    watcher.ws_message_length = 0
+    client.room.watcher_stanzas = []
+
     watcher.on 'data', (data)->
       client.room.watcher_buffers.push data
       for w in client.room.watchers
         w.write data if w #a WTF fix
 
+      watcher.ws_buffer = Buffer.concat([watcher.ws_buffer, data], watcher.ws_buffer.length + data.length) #buffer的错误使用方式，好孩子不要学
+
+      while true
+        if watcher.ws_message_length == 0
+          if watcher.ws_buffer.length >= 2
+            watcher.ws_message_length = watcher.ws_buffer.readUInt16LE(0)
+          else
+            break
+        else
+          if watcher.ws_buffer.length >= 2 + watcher.ws_message_length
+            stanza = watcher.ws_buffer.slice(2, watcher.ws_message_length + 2)
+            for w in client.room.ws_watchers
+              w.sendBytes stanza if w #a WTF fix
+            client.room.watcher_stanzas.push stanza
+
+            watcher.ws_buffer = watcher.ws_buffer.slice(2 + watcher.ws_message_length)
+            watcher.ws_message_length = 0
+          else
+            break
+
     watcher.on 'error', (error)->
       log.error "watcher error", error
 
+    watcher.on 'close', (had_error)->
+      for w in client.room.ws_watchers
+        w.close()
+
 #登场台词
 if settings.modules.dialogues
   dialogues = {}
@@ -494,14 +524,14 @@ if settings.modules.http
     waiting.push []
 
   log.info 'level_points loaded', level_points
-  http.createServer (request, response)->
+  http_server = http.createServer (request, response)->
     #http://122.0.65.70:7922/?operation=getroomjson
-      url = url.parse(request.url)
-      #log.info url
-      if url.pathname == '/count.json'
+      u = url.parse(request.url)
+      #log.info u
+      if u.pathname == '/count.json'
         response.writeHead(200);
         response.end(Room.all.length.toString())
-      else if url.pathname == '/match'
+      else if u.pathname == '/match'
         if request.headers['authorization']
           [name, password] = new Buffer(request.headers['authorization'].split(/\s+/).pop() ? '','base64').toString().split(':')
           User.findOne { name: name }, (err, user)->
@@ -529,10 +559,10 @@ if settings.modules.http
             index = waiting[level].indexOf(response)
             waiting[level].splice(index, 1) unless index == -1
 
-      else if url.pathname == '/rooms.json'
+      else if u.pathname == '/rooms.json'
         response.writeHead(404);
         response.end();
-      else if url.query == 'operation=getroomjson'
+      else if u.query == 'operation=getroomjson'
         response.writeHead(200);
         response.end JSON.stringify rooms: (for room in Room.all when room.established
           roomid: room.port.toString(),
@@ -548,7 +578,7 @@ if settings.modules.http
       else
         response.writeHead(404);
         response.end();
-  .listen settings.modules.http.port
+  http_server.listen settings.modules.http.port
 
   setInterval ()->
     for level in [level_points.length..0]
@@ -595,6 +625,51 @@ if settings.modules.http
 
   , 2000
 
+  originIsAllowed = (origin) ->
+    # allow all origin, for debug
+    true
+  wsServer = new WebSocketServer(
+    httpServer: http_server
+    autoAcceptConnections: false
+  )
+  wsServer.on "request", (request) ->
+    unless originIsAllowed(request.origin)
+      # Make sure we only accept requests from an allowed origin
+      request.reject()
+      console.log (new Date()) + " Connection from origin " + request.origin + " rejected."
+      return
+
+    room_name = decodeURIComponent(request.resource.slice(1))
+    if room_name == 'started'
+      room = _.find Room.all, (room)->
+        room.started
+    else
+      room = Room.find_by_name room_name
+    unless room
+      request.reject()
+      console.log (new Date()) + " Connection from origin " + request.origin + " rejected. #{room_name}"
+      return
+
+    connection = request.accept(null, request.origin)
+    console.log (new Date()) + " Connection accepted. #{room.name}"
+    room.ws_watchers.push connection
+
+    for stanza in room.watcher_stanzas
+      connection.sendBytes stanza
+
+    ###
+    connection.on "message", (message) ->
+      if message.type is "utf8"
+        console.log "Received Message: " + message.utf8Data
+        connection.sendUTF message.utf8Data
+      else if message.type is "binary"
+        console.log "Received Binary Message of " + message.binaryData.length + " bytes"
+        connection.sendBytes message.binaryData
+    ###
+    connection.on "close", (reasonCode, description) ->
+      index = _.indexOf(room.ws_watchers, connection)
+      room.ws_watchers.splice(index, 1) unless index == -1
+      console.log (new Date()) + " Peer " + connection.remoteAddress + " disconnected."
 
 #清理90s没活动的房间
 inotify = new Inotify()
@@ -614,7 +689,7 @@ inotify.addWatch
           room.alive = true
     else
       log.error "event without filename"
-
+###
 setInterval ()->
   for room in Room.all
     if room.alive
@@ -626,3 +701,4 @@ setInterval ()->
         ygopro.stoc_send_chat(player, "由于长时间没有活动被关闭") unless player.closed
       room.process.kill()
 , 900000
+###
\ No newline at end of file

ygopro-server.js

 // Generated by CoffeeScript 1.6.3
 (function() {
-  var Deck, Inotify, Room, User, bunyan, debug, dialogues, execFile, fs, http, i, inotify, level_points, log, mycard, net, path, request, settings, tips, url, victories, waiting, ygopro, _,
+  var Deck, Inotify, Room, User, WebSocketServer, bunyan, debug, dialogues, execFile, fs, http, http_server, i, inotify, level_points, log, mycard, net, originIsAllowed, path, request, settings, tips, url, victories, waiting, wsServer, ygopro, _,
     __indexOf = [].indexOf || function(item) { for (var i = 0, l = this.length; i < l; i++) { if (i in this && this[i] === item) return i; } return -1; };
 
   net = require('net');
@@ -23,6 +23,8 @@
 
   Inotify = require('inotify').Inotify;
 
+  WebSocketServer = require('websocket').server;
+
   request = require('request');
 
   bunyan = require('bunyan');
@@ -172,7 +174,7 @@
     stoc_message_length = 0;
     stoc_proto = 0;
     return server.on('data', function(data) {
-      var b, struct, _results;
+      var b, stanzas, struct, _results;
       stoc_buffer = Buffer.concat([stoc_buffer, data], stoc_buffer.length + data.length);
       client.write(data);
       _results = [];
@@ -191,6 +193,7 @@
           }
         } else {
           if (stoc_buffer.length >= 2 + stoc_message_length) {
+            stanzas = stoc_proto;
             if (ygopro.stoc_follows[stoc_proto]) {
               b = stoc_buffer.slice(3, stoc_message_length - 1 + 3);
               if (struct = ygopro.structs[ygopro.proto_structs.STOC[ygopro.constants.STOC[stoc_proto]]]) {
@@ -316,24 +319,61 @@
         });
         return ygopro.ctos_send(watcher, 'HS_TOOBSERVER');
       });
+      watcher.ws_buffer = new Buffer(0);
+      watcher.ws_message_length = 0;
+      client.room.watcher_stanzas = [];
       watcher.on('data', function(data) {
-        var w, _i, _len, _ref, _results;
+        var stanza, w, _i, _j, _len, _len1, _ref, _ref1, _results;
         client.room.watcher_buffers.push(data);
         _ref = client.room.watchers;
-        _results = [];
         for (_i = 0, _len = _ref.length; _i < _len; _i++) {
           w = _ref[_i];
           if (w) {
-            _results.push(w.write(data));
+            w.write(data);
+          }
+        }
+        watcher.ws_buffer = Buffer.concat([watcher.ws_buffer, data], watcher.ws_buffer.length + data.length);
+        _results = [];
+        while (true) {
+          if (watcher.ws_message_length === 0) {
+            if (watcher.ws_buffer.length >= 2) {
+              _results.push(watcher.ws_message_length = watcher.ws_buffer.readUInt16LE(0));
             } else {
-            _results.push(void 0);
+              break;
+            }
+          } else {
+            if (watcher.ws_buffer.length >= 2 + watcher.ws_message_length) {
+              stanza = watcher.ws_buffer.slice(2, watcher.ws_message_length + 2);
+              _ref1 = client.room.ws_watchers;
+              for (_j = 0, _len1 = _ref1.length; _j < _len1; _j++) {
+                w = _ref1[_j];
+                if (w) {
+                  w.sendBytes(stanza);
+                }
+              }
+              client.room.watcher_stanzas.push(stanza);
+              watcher.ws_buffer = watcher.ws_buffer.slice(2 + watcher.ws_message_length);
+              _results.push(watcher.ws_message_length = 0);
+            } else {
+              break;
+            }
           }
         }
         return _results;
       });
-      return watcher.on('error', function(error) {
+      watcher.on('error', function(error) {
         return log.error("watcher error", error);
       });
+      return watcher.on('close', function(had_error) {
+        var w, _i, _len, _ref, _results;
+        _ref = client.room.ws_watchers;
+        _results = [];
+        for (_i = 0, _len = _ref.length; _i < _len; _i++) {
+          w = _ref[_i];
+          _results.push(w.close());
+        }
+        return _results;
+      });
     }
   });
 
@@ -642,13 +682,13 @@
       waiting.push([]);
     }
     log.info('level_points loaded', level_points);
-    http.createServer(function(request, response) {
-      var level, name, password, player, room, _ref, _ref1;
-      url = url.parse(request.url);
-      if (url.pathname === '/count.json') {
+    http_server = http.createServer(function(request, response) {
+      var level, name, password, player, room, u, _ref, _ref1;
+      u = url.parse(request.url);
+      if (u.pathname === '/count.json') {
         response.writeHead(200);
         return response.end(Room.all.length.toString());
-      } else if (url.pathname === '/match') {
+      } else if (u.pathname === '/match') {
         if (request.headers['authorization']) {
           _ref1 = new Buffer((_ref = request.headers['authorization'].split(/\s+/).pop()) != null ? _ref : '', 'base64').toString().split(':'), name = _ref1[0], password = _ref1[1];
           return User.findOne({
@@ -692,10 +732,10 @@
             }
           });
         }
-      } else if (url.pathname === '/rooms.json') {
+      } else if (u.pathname === '/rooms.json') {
         response.writeHead(404);
         return response.end();
-      } else if (url.query === 'operation=getroomjson') {
+      } else if (u.query === 'operation=getroomjson') {
         response.writeHead(200);
         return response.end(JSON.stringify({
           rooms: (function() {
@@ -736,7 +776,8 @@
         response.writeHead(404);
         return response.end();
       }
-    }).listen(settings.modules.http.port);
+    });
+    http_server.listen(settings.modules.http.port);
     setInterval(function() {
       var displacement, headers, index, level, opponent, opponent_level, player, room, _i, _ref, _results;
       _results = [];
@@ -800,6 +841,60 @@
       }
       return _results;
     }, 2000);
+    originIsAllowed = function(origin) {
+      return true;
+    };
+    wsServer = new WebSocketServer({
+      httpServer: http_server,
+      autoAcceptConnections: false
+    });
+    wsServer.on("request", function(request) {
+      var connection, room, room_name, stanza, _i, _len, _ref;
+      if (!originIsAllowed(request.origin)) {
+        request.reject();
+        console.log((new Date()) + " Connection from origin " + request.origin + " rejected.");
+        return;
+      }
+      room_name = decodeURIComponent(request.resource.slice(1));
+      if (room_name === 'started') {
+        room = _.find(Room.all, function(room) {
+          return room.started;
+        });
+      } else {
+        room = Room.find_by_name(room_name);
+      }
+      if (!room) {
+        request.reject();
+        console.log((new Date()) + " Connection from origin " + request.origin + (" rejected. " + room_name));
+        return;
+      }
+      connection = request.accept(null, request.origin);
+      console.log((new Date()) + (" Connection accepted. " + room.name));
+      room.ws_watchers.push(connection);
+      _ref = room.watcher_stanzas;
+      for (_i = 0, _len = _ref.length; _i < _len; _i++) {
+        stanza = _ref[_i];
+        connection.sendBytes(stanza);
+      }
+      /*
+      connection.on "message", (message) ->
+        if message.type is "utf8"
+          console.log "Received Message: " + message.utf8Data
+          connection.sendUTF message.utf8Data
+        else if message.type is "binary"
+          console.log "Received Binary Message of " + message.binaryData.length + " bytes"
+          connection.sendBytes message.binaryData
+      */
+
+      return connection.on("close", function(reasonCode, description) {
+        var index;
+        index = _.indexOf(room.ws_watchers, connection);
+        if (index !== -1) {
+          room.ws_watchers.splice(index, 1);
+        }
+        return console.log((new Date()) + " Peer " + connection.remoteAddress + " disconnected.");
+      });
+    });
   }
 
   inotify = new Inotify();
@@ -828,28 +923,20 @@
     }
   });
 
-  setInterval(function() {
-    var player, room, _i, _j, _len, _len1, _ref, _ref1, _results;
-    _ref = Room.all;
-    _results = [];
-    for (_i = 0, _len = _ref.length; _i < _len; _i++) {
-      room = _ref[_i];
-      if (room.alive) {
-        _results.push(room.alive = false);
-      } else {
-        log.info("kill room", room.port);
-        _ref1 = room.players;
-        for (_j = 0, _len1 = _ref1.length; _j < _len1; _j++) {
-          player = _ref1[_j];
-          if (!player.closed) {
-            ygopro.stoc_send_chat(player, "由于长时间没有活动被关闭");
-          }
-        }
-        _results.push(room.process.kill());
-      }
-    }
-    return _results;
-  }, 900000);
+  /*
+  setInterval ()->
+    for room in Room.all
+      if room.alive
+        room.alive = false
+      else
+        log.info "kill room", room.port
+  
+        for player in room.players
+          ygopro.stoc_send_chat(player, "由于长时间没有活动被关闭") unless player.closed
+        room.process.kill()
+  , 900000
+  */
+
 
 }).call(this);