feat: add fast header support

Author: ronag

src/event/event-handler.js

@@ -1,9 +1,8 @@
-const messageBuilder = require('../message/message-builder')
-const messageParser = require('../message/message-parser')
 const C = require('../constants/constants')
 const MulticastListener = require('../utils/multicast-listener')
 const UnicastListener = require('../utils/unicast-listener')
 const EventEmitter = require('component-emitter2')
+const Message = require('../message/message')
 const rxjs = require('rxjs')
 
 const EventHandler = function (options, connection, client) {
@@ -105,7 +104,7 @@ EventHandler.prototype.emit = function (name, data) {
     throw new Error('invalid argument name')
   }
 
-  this._connection.sendMsg(C.TOPIC.EVENT, C.ACTIONS.EVENT, [name, messageBuilder.typed(data)])
+  this._connection.sendMsg(C.TOPIC.EVENT, C.ACTIONS.EVENT, [name, Message.encodeTyped(data)])
   this._emitter.emit(name, data)
   this._stats.emitted += 1
 }
@@ -143,7 +142,7 @@ EventHandler.prototype._$handle = function (message) {
 
   if (message.action === C.ACTIONS.EVENT) {
     if (message.data && message.data.length === 2) {
-      this._emitter.emit(name, messageParser.convertTyped(data, this._client))
+      this._emitter.emit(name, Message.decodeTyped(data, this._client))
     } else {
       this._emitter.emit(name)
     }

src/message/connection.js

@@ -1,8 +1,7 @@
 const BrowserWebSocket = globalThis.WebSocket || globalThis.MozWebSocket
 const utils = require('../utils/utils')
 const NodeWebSocket = utils.isNode ? require('ws') : null
-const messageParser = require('./message-parser')
-const messageBuilder = require('./message-builder')
+const Message = require('./message')
 const C = require('../constants/constants')
 const pkg = require('../../package.json')
 const xxhash = require('xxhash-wasm')
@@ -79,15 +78,15 @@ Connection.prototype.authenticate = function (authParams, callback) {
 }
 
 Connection.prototype.sendMsg = function (topic, action, data) {
-  return this.send(messageBuilder.getMsg(topic, action, data))
+  return this.send(Message.encode(topic, action, data))
 }
 
 Connection.prototype.sendMsg1 = function (topic, action, p0) {
-  return this.send(messageBuilder.getMsg1(topic, action, p0))
+  return this.send(Message.encode(topic, action, [p0]))
 }
 
 Connection.prototype.sendMsg2 = function (topic, action, p0, p1) {
-  return this.send(messageBuilder.getMsg2(topic, action, p0, p1))
+  return this.send(Message.encode(topic, action, [p0, p1]))
 }
 
 Connection.prototype.close = function () {
@@ -101,19 +100,22 @@ Connection.prototype.close = function () {
 }
 
 Connection.prototype._createEndpoint = function () {
-  this._endpoint = NodeWebSocket
-    ? new NodeWebSocket(this._url, {
-        generateMask() {},
-      })
-    : new BrowserWebSocket(this._url)
+  if (NodeWebSocket) {
+    this._endpoint = new NodeWebSocket(this._url, {
+      generateMask() {},
+    })
+  } else {
+    this._endpoint = new BrowserWebSocket(this._url)
+    this._endpoint.binaryType = 'arraybuffer'
+  }
   this._corked = false
 
   this._endpoint.onopen = this._onOpen.bind(this)
   this._endpoint.onerror = this._onError.bind(this)
   this._endpoint.onclose = this._onClose.bind(this)
   this._endpoint.onmessage = BrowserWebSocket
-    ? ({ data }) => this._onMessage(typeof data === 'string' ? data : Buffer.from(data).toString())
-    : ({ data }) => this._onMessage(typeof data === 'string' ? data : data.toString())
+    ? ({ data }) => this._onMessage(Buffer.from(data))
+    : ({ data }) => this._onMessage(data)
 }
 
 Connection.prototype.send = function (message) {
@@ -172,14 +174,15 @@ Connection.prototype._submit = function (message) {
 
 Connection.prototype._sendAuthParams = function () {
   this._setState(C.CONNECTION_STATE.AUTHENTICATING)
-  const authMessage = messageBuilder.getMsg(C.TOPIC.AUTH, C.ACTIONS.REQUEST, [
-    this._authParams,
-    pkg.version,
-    utils.isNode
-      ? `Node/${process.version}`
-      : globalThis.navigator && globalThis.navigator.userAgent,
-  ])
-  this._submit(authMessage)
+  this._submit(
+    Message.encode(C.TOPIC.AUTH, C.ACTIONS.REQUEST, [
+      this._authParams,
+      pkg.version,
+      utils.isNode
+        ? `Node/${process.version}`
+        : globalThis.navigator && globalThis.navigator.userAgent,
+    ])
+  )
 }
 
 Connection.prototype._onOpen = function () {
@@ -220,11 +223,6 @@ Connection.prototype._onClose = function () {
 }
 
 Connection.prototype._onMessage = function (data) {
-  // Remove MESSAGE_SEPERATOR if exists.
-  if (data.charCodeAt(data.length - 1) === 30) {
-    data = data.slice(0, -1)
-  }
-
   this._recvQueue.push(data)
   if (!this._processingRecv) {
     this._processingRecv = true
@@ -239,30 +237,26 @@ Connection.prototype._recvMessages = function (deadline) {
     deadline ? deadline.didTimeout || deadline.timeRemaining() : n < this._batchSize;
     ++n
   ) {
-    const message = this._recvQueue.shift()
-    if (!message) {
+    const raw = this._recvQueue.shift()
+    if (!raw) {
       this._processingRecv = false
       return
     }
 
-    if (message.length <= 2) {
+    if (raw.length <= 2) {
       continue
     }
 
-    if (this._logger) {
-      this._logger.trace(message, 'receive')
-    }
-
-    messageParser.parseMessage(message, this._client, this._message)
+    const message = Message.decode(raw)
 
-    this.emit('recv', this._message)
+    this.emit('recv', message)
 
-    if (this._message.topic === C.TOPIC.CONNECTION) {
-      this._handleConnectionResponse(this._message)
-    } else if (this._message.topic === C.TOPIC.AUTH) {
-      this._handleAuthResponse(this._message)
+    if (message.topic === C.TOPIC.CONNECTION) {
+      this._handleConnectionResponse(message)
+    } else if (message.topic === C.TOPIC.AUTH) {
+      this._handleAuthResponse(message)
     } else {
-      this._client._$onMessage(this._message)
+      this._client._$onMessage(message)
     }
   }
 
@@ -271,17 +265,15 @@ Connection.prototype._recvMessages = function (deadline) {
 
 Connection.prototype._handleConnectionResponse = function (message) {
   if (message.action === C.ACTIONS.PING) {
-    this._submit(messageBuilder.getMsg(C.TOPIC.CONNECTION, C.ACTIONS.PONG))
+    this._submit(Message.encode(C.TOPIC.CONNECTION, C.ACTIONS.PONG))
   } else if (message.action === C.ACTIONS.ACK) {
     this._setState(C.CONNECTION_STATE.AWAITING_AUTHENTICATION)
     if (this._authParams) {
       this._sendAuthParams()
     }
   } else if (message.action === C.ACTIONS.CHALLENGE) {
     this._setState(C.CONNECTION_STATE.CHALLENGING)
-    this._submit(
-      messageBuilder.getMsg(C.TOPIC.CONNECTION, C.ACTIONS.CHALLENGE_RESPONSE, [this._url])
-    )
+    this._submit(Message.encode(C.TOPIC.CONNECTION, C.ACTIONS.CHALLENGE_RESPONSE, [this._url]))
   } else if (message.action === C.ACTIONS.REJECTION) {
     this._challengeDenied = true
     this.close()
@@ -319,7 +311,7 @@ Connection.prototype._getAuthData = function (data) {
   if (data === undefined) {
     return null
   } else {
-    return messageParser.convertTyped(data, this._client)
+    return Message.decodeTyped(data, this._client)
   }
 }