WebRTC信令服务器实现八、第三节 WebRTC信令服务器的实现

客官°小女子只卖身不卖艺 2023-02-25 08:57 8阅读 0赞

今天我们使用信令服务器来实现一个简易的聊天室。

安装

npm install --save socket.io http https express serve-index log4js

socket.js

/**
     * Module dependencies.
     */
    
    var Emitter = require('events').EventEmitter;
    var parser = require('socket.io-parser');
    var url = require('url');
    var debug = require('debug')('socket.io:socket');
    var assign = require('object-assign');
    
    /**
     * Module exports.
     */
    
    module.exports = exports = Socket;
    
    /**
     * Blacklisted events.
     *
     * @api public
     */
    
    exports.events = [
      'error',
      'connect',
      'disconnect',
      'disconnecting',
      'newListener',
      'removeListener'
    ];
    
    /**
     * Flags.
     *
     * @api private
     */
    
    var flags = [
      'json',
      'volatile',
      'broadcast'
    ];
    
    /**
     * `EventEmitter#emit` reference.
     */
    
    var emit = Emitter.prototype.emit;
    
    /**
     * Interface to a `Client` for a given `Namespace`.
     *
     * @param {Namespace} nsp
     * @param {Client} client
     * @api public
     */
    
    function Socket(nsp, client, query){
      this.nsp = nsp;
      this.server = nsp.server;
      this.adapter = this.nsp.adapter;
      this.id = nsp.name !== '/' ? nsp.name + '#' + client.id : client.id;
      this.client = client;
      this.conn = client.conn;
      this.rooms = {};
      this.acks = {};
      this.connected = true;
      this.disconnected = false;
      this.handshake = this.buildHandshake(query);
      this.fns = [];
      this.flags = {};
      this._rooms = [];
    }
    
    /**
     * Inherits from `EventEmitter`.
     */
    
    Socket.prototype.__proto__ = Emitter.prototype;
    
    /**
     * Apply flags from `Socket`.
     */
    
    flags.forEach(function(flag){
      Object.defineProperty(Socket.prototype, flag, {
        get: function() {
          this.flags[flag] = true;
          return this;
        }
      });
    });
    
    /**
     * `request` engine.io shortcut.
     *
     * @api public
     */
    
    Object.defineProperty(Socket.prototype, 'request', {
      get: function() {
        return this.conn.request;
      }
    });
    
    /**
     * Builds the `handshake` BC object
     *
     * @api private
     */
    
    Socket.prototype.buildHandshake = function(query){
      var self = this;
      function buildQuery(){
        var requestQuery = url.parse(self.request.url, true).query;
        //if socket-specific query exist, replace query strings in requestQuery
        return assign({}, query, requestQuery);
      }
      return {
        headers: this.request.headers,
        time: (new Date) + '',
        address: this.conn.remoteAddress,
        xdomain: !!this.request.headers.origin,
        secure: !!this.request.connection.encrypted,
        issued: +(new Date),
        url: this.request.url,
        query: buildQuery()
      };
    };
    
    /**
     * Emits to this client.
     *
     * @return {Socket} self
     * @api public
     */
    
    Socket.prototype.emit = function(ev){
      if (~exports.events.indexOf(ev)) {
        emit.apply(this, arguments);
        return this;
      }
    
      var args = Array.prototype.slice.call(arguments);
      var packet = {
        type: parser.EVENT,
        data: args
      };
    
      // access last argument to see if it's an ACK callback
      if (typeof args[args.length - 1] === 'function') {
        if (this._rooms.length || this.flags.broadcast) {
          throw new Error('Callbacks are not supported when broadcasting');
        }
    
        debug('emitting packet with ack id %d', this.nsp.ids);
        this.acks[this.nsp.ids] = args.pop();
        packet.id = this.nsp.ids++;
      }
    
      var rooms = this._rooms.slice(0);
      var flags = assign({}, this.flags);
    
      // reset flags
      this._rooms = [];
      this.flags = {};
    
      if (rooms.length || flags.broadcast) {
        this.adapter.broadcast(packet, {
          except: [this.id],
          rooms: rooms,
          flags: flags
        });
      } else {
        // dispatch packet
        this.packet(packet, flags);
      }
      return this;
    };
    
    /**
     * Targets a room when broadcasting.
     *
     * @param {String} name
     * @return {Socket} self
     * @api public
     */
    
    Socket.prototype.to =
    Socket.prototype.in = function(name){
      if (!~this._rooms.indexOf(name)) this._rooms.push(name);
      return this;
    };
    
    /**
     * Sends a `message` event.
     *
     * @return {Socket} self
     * @api public
     */
    
    Socket.prototype.send =
    Socket.prototype.write = function(){
      var args = Array.prototype.slice.call(arguments);
      args.unshift('message');
      this.emit.apply(this, args);
      return this;
    };
    
    /**
     * Writes a packet.
     *
     * @param {Object} packet object
     * @param {Object} opts options
     * @api private
     */
    
    Socket.prototype.packet = function(packet, opts){
      packet.nsp = this.nsp.name;
      opts = opts || {};
      opts.compress = false !== opts.compress;
      this.client.packet(packet, opts);
    };
    
    /**
     * Joins a room.
     *
     * @param {String|Array} room or array of rooms
     * @param {Function} fn optional, callback
     * @return {Socket} self
     * @api private
     */
    
    Socket.prototype.join = function(rooms, fn){
      debug('joining room %s', rooms);
      var self = this;
      if (!Array.isArray(rooms)) {
        rooms = [rooms];
      }
      rooms = rooms.filter(function (room) {
        return !self.rooms.hasOwnProperty(room);
      });
      if (!rooms.length) {
        fn && fn(null);
        return this;
      }
      this.adapter.addAll(this.id, rooms, function(err){
        if (err) return fn && fn(err);
        debug('joined room %s', rooms);
        rooms.forEach(function (room) {
          self.rooms[room] = room;
        });
        fn && fn(null);
      });
      return this;
    };
    
    /**
     * Leaves a room.
     *
     * @param {String} room
     * @param {Function} fn optional, callback
     * @return {Socket} self
     * @api private
     */
    
    Socket.prototype.leave = function(room, fn){
      debug('leave room %s', room);
      var self = this;
      this.adapter.del(this.id, room, function(err){
        if (err) return fn && fn(err);
        debug('left room %s', room);
        delete self.rooms[room];
        fn && fn(null);
      });
      return this;
    };
    
    /**
     * Leave all rooms.
     *
     * @api private
     */
    
    Socket.prototype.leaveAll = function(){
      this.adapter.delAll(this.id);
      this.rooms = {};
    };
    
    /**
     * Called by `Namespace` upon successful
     * middleware execution (ie: authorization).
     * Socket is added to namespace array before
     * call to join, so adapters can access it.
     *
     * @api private
     */
    
    Socket.prototype.onconnect = function(){
      debug('socket connected - writing packet');
      this.nsp.connected[this.id] = this;
      this.join(this.id);
      var skip = this.nsp.name === '/' && this.nsp.fns.length === 0;
      if (skip) {
        debug('packet already sent in initial handshake');
      } else {
        this.packet({ type: parser.CONNECT });
      }
    };
    
    /**
     * Called with each packet. Called by `Client`.
     *
     * @param {Object} packet
     * @api private
     */
    
    Socket.prototype.onpacket = function(packet){
      debug('got packet %j', packet);
      switch (packet.type) {
        case parser.EVENT:
          this.onevent(packet);
          break;
    
        case parser.BINARY_EVENT:
          this.onevent(packet);
          break;
    
        case parser.ACK:
          this.onack(packet);
          break;
    
        case parser.BINARY_ACK:
          this.onack(packet);
          break;
    
        case parser.DISCONNECT:
          this.ondisconnect();
          break;
    
        case parser.ERROR:
          this.emit('error', packet.data);
      }
    };
    
    /**
     * Called upon event packet.
     *
     * @param {Object} packet object
     * @api private
     */
    
    Socket.prototype.onevent = function(packet){
      var args = packet.data || [];
      debug('emitting event %j', args);
    
      if (null != packet.id) {
        debug('attaching ack callback to event');
        args.push(this.ack(packet.id));
      }
    
      this.dispatch(args);
    };
    
    /**
     * Produces an ack callback to emit with an event.
     *
     * @param {Number} id packet id
     * @api private
     */
    
    Socket.prototype.ack = function(id){
      var self = this;
      var sent = false;
      return function(){
        // prevent double callbacks
        if (sent) return;
        var args = Array.prototype.slice.call(arguments);
        debug('sending ack %j', args);
    
        self.packet({
          id: id,
          type: parser.ACK,
          data: args
        });
    
        sent = true;
      };
    };
    
    /**
     * Called upon ack packet.
     *
     * @api private
     */
    
    Socket.prototype.onack = function(packet){
      var ack = this.acks[packet.id];
      if ('function' == typeof ack) {
        debug('calling ack %s with %j', packet.id, packet.data);
        ack.apply(this, packet.data);
        delete this.acks[packet.id];
      } else {
        debug('bad ack %s', packet.id);
      }
    };
    
    /**
     * Called upon client disconnect packet.
     *
     * @api private
     */
    
    Socket.prototype.ondisconnect = function(){
      debug('got disconnect packet');
      this.onclose('client namespace disconnect');
    };
    
    /**
     * Handles a client error.
     *
     * @api private
     */
    
    Socket.prototype.onerror = function(err){
      if (this.listeners('error').length) {
        this.emit('error', err);
      } else {
        console.error('Missing error handler on `socket`.');
        console.error(err.stack);
      }
    };
    
    /**
     * Called upon closing. Called by `Client`.
     *
     * @param {String} reason
     * @throw {Error} optional error object
     * @api private
     */
    
    Socket.prototype.onclose = function(reason){
      if (!this.connected) return this;
      debug('closing socket - reason %s', reason);
      this.emit('disconnecting', reason);
      this.leaveAll();
      this.nsp.remove(this);
      this.client.remove(this);
      this.connected = false;
      this.disconnected = true;
      delete this.nsp.connected[this.id];
      this.emit('disconnect', reason);
    };
    
    /**
     * Produces an `error` packet.
     *
     * @param {Object} err error object
     * @api private
     */
    
    Socket.prototype.error = function(err){
      this.packet({ type: parser.ERROR, data: err });
    };
    
    /**
     * Disconnects this client.
     *
     * @param {Boolean} close if `true`, closes the underlying connection
     * @return {Socket} self
     * @api public
     */
    
    Socket.prototype.disconnect = function(close){
      if (!this.connected) return this;
      if (close) {
        this.client.disconnect();
      } else {
        this.packet({ type: parser.DISCONNECT });
        this.onclose('server namespace disconnect');
      }
      return this;
    };
    
    /**
     * Sets the compress flag.
     *
     * @param {Boolean} compress if `true`, compresses the sending data
     * @return {Socket} self
     * @api public
     */
    
    Socket.prototype.compress = function(compress){
      this.flags.compress = compress;
      return this;
    };
    
    /**
     * Dispatch incoming event to socket listeners.
     *
     * @param {Array} event that will get emitted
     * @api private
     */
    
    Socket.prototype.dispatch = function(event){
      debug('dispatching an event %j', event);
      var self = this;
      function dispatchSocket(err) {
        process.nextTick(function(){
          if (err) {
            return self.error(err.data || err.message);
          }
          emit.apply(self, event);
        });
      }
      this.run(event, dispatchSocket);
    };
    
    /**
     * Sets up socket middleware.
     *
     * @param {Function} middleware function (event, next)
     * @return {Socket} self
     * @api public
     */
    
    Socket.prototype.use = function(fn){
      this.fns.push(fn);
      return this;
    };
    
    /**
     * Executes the middleware for an incoming event.
     *
     * @param {Array} event that will get emitted
     * @param {Function} last fn call in the middleware
     * @api private
     */
    Socket.prototype.run = function(event, fn){
      var fns = this.fns.slice(0);
      if (!fns.length) return fn(null);
    
      function run(i){
        fns[i](event, function(err){
          // upon error, short-circuit
          if (err) return fn(err);
    
          // if no middleware left, summon callback
          if (!fns[i + 1]) return fn(null);
    
          // go on to next
          run(i + 1);
        });
      }
    
      run(0);
    };

client.js

'use strict'
    
    // 1、拿到HTML的元素
    var userName = document.querySelector('input#username');
    var inputRoom = document.querySelector('input#room');
    var btnConnect = document.querySelector('button#connect');
    var btnLeave = document.querySelector('button#leave');
    var outputArea = document.querySelector('textarea#output');
    var inputArea = document.querySelector('textarea#input');
    var btnSend = document.querySelector('button#send');
    
    var socket; // 2.1 前 定义一个socket全局对象
    var room;
    
    // 2.0、建立连接
    btnConnect.onclick = ()=>{
    
    	//connect 2.1 连接服务器
    	socket = io.connect(); 
    	
    	//recieve message 2.2 处理消息 加入房间 这时收到返回room和id分别是所在房间和用户id
    	socket.on('joined', (room, id) => {
    		btnConnect.disabled = true;
    		btnLeave.disabled = false;
    		inputArea.disabled = false;
    		btnSend.disabled = false;
    	});	
    	
    	// 2.4  离开房间 离开房间后的一些处理
    	socket.on('leaved', (room, id) => {
    		btnConnect.disabled = false;
    		btnLeave.disabled = true;
    		inputArea.disabled = true;
    		btnSend.disabled = true;
    
    		socket.disconnect();
    	});	
    
    	// 2.5 消息转发 room, id, data 分别表示房间号、用户id和数据
    	// socket.on('message', (room, id, data) => {
    		socket.on('message', (room, data) => {
    		outputArea.scrollTop = outputArea.scrollHeight; //窗口总是显示最后的内容
    		console.log(data)
    		outputArea.value = outputArea.value + data + '\r';
    	});	
    
    	socket.on('disconnect', (socket)=>{
    		btnConnect.disabled = false;
    		btnLeave.disabled = true;
    		inputArea.disabled = true;
    		btnSend.disabled = true;
    	});
    
    	//send message 2.3 发送消息
    	room = inputRoom.value;
    	socket.emit('join', room); // 加入房间
    }
    
    // 3.0 发送消息
    btnSend.onclick = ()=>{
    	console.log(inputArea)
    	var data = inputArea.value;
    	data = userName.value + ':' + data;
    	outputArea.value = outputArea.value + data + '\r';
    	socket.emit('message', room, data); // 发送消息
    	inputArea.value = '';
    }
    
    btnLeave.onclick = ()=>{
    	room = inputRoom.value;
    	socket.emit('leave', room);
    }
    
    inputArea.onkeypress = (event)=> {
        //event = event || window.event;
        if (event.keyCode == 13) { //回车发送消息
    	var data = inputArea.value;
    	data = userName.value + ':' + data;
    	socket.emit('message', room, data);
    	inputArea.value = '';
    	event.preventDefault();//阻止默认行为
        }
    }

运行 node server.js

打开 [https://0.0.0.0/chatroom/index.html][https_0.0.0.0_chatroom_index.html]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h5cGhm_size_16_color_FFFFFF_t_70][]

[https_0.0.0.0_chatroom_index.html]: https://0.0.0.0/chatroom/index.html
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3h5cGhm_size_16_color_FFFFFF_t_70]: /images/20230209/05a39e1d58474fc89007b250836a3f6d.png