ceph/src/jaegertracing/thrift/lib/nodejs/lib/thrift/ws_transport.js

   1 /*
   2  * Licensed to the Apache Software Foundation (ASF) under one
   3  * or more contributor license agreements. See the NOTICE file
   4  * distributed with this work for additional information
   5  * regarding copyright ownership. The ASF licenses this file
   6  * to you under the Apache License, Version 2.0 (the
   7  * "License"); you may not use this file except in compliance
   8  * with the License. You may obtain a copy of the License at
   9  *
  10  *   http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing,
  13  * software distributed under the License is distributed on an
  14  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
  15  * KIND, either express or implied. See the License for the
  16  * specific language governing permissions and limitations
  17  * under the License.
  18  */
  19
  20 var log = require('./log');
  21
  22 module.exports = TWebSocketTransport;
  23
  24 /**
  25  * Constructor Function for the WebSocket transport.
  26  * @constructor
  27  * @param {string} [url] - The URL to connect to.
  28  * @classdesc The Apache Thrift Transport layer performs byte level I/O
  29  * between RPC clients and servers. The JavaScript TWebSocketTransport object
  30  * uses the WebSocket protocol. Target servers must implement WebSocket.
  31  * (see: node.js example server_http.js).
  32  * @example
  33  *   var transport = new Thrift.TWebSocketTransport("http://localhost:8585");
  34  */
  35 function TWebSocketTransport(url) {
  36     this.__reset(url);
  37 };
  38
  39
  40 TWebSocketTransport.prototype.__reset = function(url) {
  41   this.url = url;             //Where to connect
  42   this.socket = null;         //The web socket
  43   this.callbacks = [];        //Pending callbacks
  44   this.send_pending = [];     //Buffers/Callback pairs waiting to be sent
  45   this.send_buf = '';         //Outbound data, immutable until sent
  46   this.recv_buf = '';         //Inbound data
  47   this.rb_wpos = 0;           //Network write position in receive buffer
  48   this.rb_rpos = 0;           //Client read position in receive buffer
  49 };
  50
  51 /**
  52  * Sends the current WS request and registers callback. The async
  53  * parameter is ignored (WS flush is always async) and the callback
  54  * function parameter is required.
  55  * @param {object} async - Ignored.
  56  * @param {object} callback - The client completion callback.
  57  * @returns {undefined|string} Nothing (undefined)
  58  */
  59 TWebSocketTransport.prototype.flush = function(async, callback) {
  60   var self = this;
  61   if (this.isOpen()) {
  62     //Send data and register a callback to invoke the client callback
  63     this.socket.send(this.send_buf);
  64     this.callbacks.push((function() {
  65       var clientCallback = callback;
  66       return function(msg) {
  67         self.setRecvBuffer(msg);
  68         clientCallback();
  69       };
  70     }()));
  71   } else {
  72     //Queue the send to go out __onOpen
  73     this.send_pending.push({
  74       buf: this.send_buf,
  75       cb:  callback
  76     });
  77   }
  78 };
  79
  80 TWebSocketTransport.prototype.__onOpen = function() {
  81    var self = this;
  82    if (this.send_pending.length > 0) {
  83       //If the user made calls before the connection was fully
  84       //open, send them now
  85       this.send_pending.forEach(function(elem) {
  86          this.socket.send(elem.buf);
  87          this.callbacks.push((function() {
  88            var clientCallback = elem.cb;
  89            return function(msg) {
  90               self.setRecvBuffer(msg);
  91               clientCallback();
  92            };
  93          }()));
  94       });
  95       this.send_pending = [];
  96    }
  97 };
  98
  99 TWebSocketTransport.prototype.__onClose = function(evt) {
 100   this.__reset(this.url);
 101 };
 102
 103 TWebSocketTransport.prototype.__onMessage = function(evt) {
 104   if (this.callbacks.length) {
 105     this.callbacks.shift()(evt.data);
 106   }
 107 };
 108
 109 TWebSocketTransport.prototype.__onError = function(evt) {
 110   log.error('websocket: ' + evt.toString());
 111   this.socket.close();
 112 };
 113
 114 /**
 115  * Sets the buffer to use when receiving server responses.
 116  * @param {string} buf - The buffer to receive server responses.
 117  */
 118 TWebSocketTransport.prototype.setRecvBuffer = function(buf) {
 119   this.recv_buf = buf;
 120   this.recv_buf_sz = this.recv_buf.length;
 121   this.wpos = this.recv_buf.length;
 122   this.rpos = 0;
 123 };
 124
 125 /**
 126  * Returns true if the transport is open
 127  * @readonly
 128  * @returns {boolean}
 129  */
 130 TWebSocketTransport.prototype.isOpen = function() {
 131   return this.socket && this.socket.readyState == this.socket.OPEN;
 132 };
 133
 134 /**
 135  * Opens the transport connection
 136  */
 137 TWebSocketTransport.prototype.open = function() {
 138   //If OPEN/CONNECTING/CLOSING ignore additional opens
 139   if (this.socket && this.socket.readyState != this.socket.CLOSED) {
 140     return;
 141   }
 142   //If there is no socket or the socket is closed:
 143   this.socket = new WebSocket(this.url);
 144   this.socket.onopen = this.__onOpen.bind(this);
 145   this.socket.onmessage = this.__onMessage.bind(this);
 146   this.socket.onerror = this.__onError.bind(this);
 147   this.socket.onclose = this.__onClose.bind(this);
 148 };
 149
 150 /**
 151  * Closes the transport connection
 152  */
 153 TWebSocketTransport.prototype.close = function() {
 154   this.socket.close();
 155 };
 156
 157 /**
 158  * Returns the specified number of characters from the response
 159  * buffer.
 160  * @param {number} len - The number of characters to return.
 161  * @returns {string} Characters sent by the server.
 162  */
 163 TWebSocketTransport.prototype.read = function(len) {
 164   var avail = this.wpos - this.rpos;
 165
 166   if (avail === 0) {
 167     return '';
 168   }
 169
 170   var give = len;
 171
 172   if (avail < len) {
 173     give = avail;
 174   }
 175
 176   var ret = this.read_buf.substr(this.rpos, give);
 177   this.rpos += give;
 178
 179   //clear buf when complete?
 180   return ret;
 181 };
 182
 183 /**
 184  * Returns the entire response buffer.
 185  * @returns {string} Characters sent by the server.
 186  */
 187 TWebSocketTransport.prototype.readAll = function() {
 188   return this.recv_buf;
 189 };
 190
 191 /**
 192  * Sets the send buffer to buf.
 193  * @param {string} buf - The buffer to send.
 194  */
 195 TWebSocketTransport.prototype.write = function(buf) {
 196   this.send_buf = buf;
 197 };
 198
 199 /**
 200  * Returns the send buffer.
 201  * @readonly
 202  * @returns {string} The send buffer.
 203  */
 204 TWebSocketTransport.prototype.getSendBuffer = function() {
 205   return this.send_buf;
 206 };