Switch to sync zlib with 512k chunks, adjustable compression level (#174

)

* Switch to sync zlib with 512k chunks, adjustable compression level

* update serverPlayer

docs/API.md

@@ -20,8 +20,8 @@ Returns a `Client` instance and connects to the server.
 | skipPing | *optional* | Whether pinging the server to check its version should be skipped. |
 | conLog | *optional* | Where to log connection information (server join, kick messages to). Defaults to console.log, set to `null` to not log anywhere. |
 | useNativeRaknet | *optional* | Whether to use the C++ version of RakNet. Set to false to use JS. |
-| authTitle | *optional* | The client ID to sign in as, defaults to Minecraft for Nintendo Switch. Set false to sign in through Azure. See prismarine-auth |
-| deviceType | *optional* | The device type to sign in as, defaults to "Nintendo". See prismarine-auth |
+| compressionLevel | *optional* | What zlib compression level to use, default to **7** |
+| batchingInterval | *optional* | How frequently, in milliseconds to flush and write the packet queue (default: 20ms) |
 
 The following events are emitted by the client:
 * 'status' - When the client's login sequence status has changed
@@ -30,6 +30,8 @@ The following events are emitted by the client:
 * 'kick' - The server has kicked the client
 * 'close' - The server has closed the connection
 * 'error' - An recoverable exception has happened. Not catching will throw an exception
+* 'connect_allowed' - Emitted after the client has pinged the server and gets version information.
+* 'heartbeat' - Emitted after two successful tick_sync (keepalive) packets have been sent bidirectionally
 
 ## be.createServer(options) : Server
 

index.d.ts

@@ -21,6 +21,10 @@ declare module "bedrock-protocol" {
     useNativeRaknet?: boolean,
     // If using JS implementation of RakNet, should we use workers? (This only affects the client)
     useRaknetWorker?: boolean
+    // Compression level for zlib, default to 7
+    compressionLevel?: number
+    // How frequently the packet queue should be flushed in milliseconds, defaults to 20ms
+    batchingInterval?: number
   }
 
   export interface ClientOptions extends Options {

src/client.js

@@ -68,6 +68,7 @@ class Client extends Connection {
     if (this.options.protocolVersion < Options.MIN_VERSION) {
       throw new Error(`Protocol version < ${Options.MIN_VERSION} : ${this.options.protocolVersion}, too old`)
     }
+    this.compressionLevel = this.options.compressionLevel || 7
   }
 
   get entityId () {
@@ -140,7 +141,12 @@ class Client extends Connection {
     if (this.status === ClientStatus.Initializing && this.options.autoInitPlayer === true) {
       if (statusPacket.status === 'player_spawn') {
         this.status = ClientStatus.Initialized
-        this.write('set_local_player_as_initialized', { runtime_entity_id: this.entityId })
+        if (!this.entityId) {
+          // We need to wait for start_game in the rare event we get a player_spawn before start_game race condition
+          this.on('start_game', () => this.write('set_local_player_as_initialized', { runtime_entity_id: this.entityId }))
+        } else {
+          this.write('set_local_player_as_initialized', { runtime_entity_id: this.entityId })
+        }
         this.emit('spawn')
       }
     }

src/connection.js

@@ -60,7 +60,7 @@ class Connection extends EventEmitter {
   write (name, params) {
     this.outLog?.(name, params)
     if (name === 'start_game') this.updateItemPalette(params.itemstates)
-    const batch = new Framer()
+    const batch = new Framer(this.compressionLevel)
     const packet = this.serializer.createPacketBuffer({ name, params })
     batch.addEncodedPacket(packet)
 
@@ -84,29 +84,33 @@ class Connection extends EventEmitter {
     this.sendIds.push(name)
   }
 
+  _tick () {
+    if (this.sendQ.length) {
+      const batch = new Framer(this.compressionLevel)
+      batch.addEncodedPackets(this.sendQ)
+      this.sendQ = []
+      this.sendIds = []
+      if (this.encryptionEnabled) {
+        this.sendEncryptedBatch(batch)
+      } else {
+        this.sendDecryptedBatch(batch)
+      }
+    }
+  }
+
+  onTick = this._tick.bind(this)
+
   startQueue () {
     this.sendQ = []
-    this.loop = setInterval(() => {
-      if (this.sendQ.length) {
-        const batch = new Framer()
-        batch.addEncodedPackets(this.sendQ)
-        this.sendQ = []
-        this.sendIds = []
-        if (this.encryptionEnabled) {
-          this.sendEncryptedBatch(batch)
-        } else {
-          this.sendDecryptedBatch(batch)
-        }
-      }
-    }, 20)
+    this.loop = setInterval(this.onTick, this.options.batchingInterval || 20)
   }
 
   /**
    * Sends a MCPE packet buffer
    */
   sendBuffer (buffer, immediate = false) {
     if (immediate) {
-      const batch = new Framer()
+      const batch = new Framer(this.compressionLevel)
       batch.addEncodedPacket(buffer)
       if (this.encryptionEnabled) {
         this.sendEncryptedBatch(batch)
@@ -121,7 +125,7 @@ class Connection extends EventEmitter {
 
   sendDecryptedBatch (batch) {
     // send to raknet
-    batch.encode(buf => this.sendMCPE(buf, true))
+    this.sendMCPE(batch.encode(), true)
   }
 
   sendEncryptedBatch (batch) {
@@ -158,11 +162,10 @@ class Connection extends EventEmitter {
       if (this.encryptionEnabled) {
         this.decrypt(buffer.slice(1))
       } else {
-        Framer.decode(buffer, packets => {
-          for (const packet of packets) {
-            this.readPacket(packet)
-          }
-        })
+        const packets = Framer.decode(buffer)
+        for (const packet of packets) {
+          this.readPacket(packet)
+        }
       }
     }
   }

src/createClient.js

@@ -17,6 +17,7 @@ function createClient (options) {
       const ad = advertisement.fromServerName(data)
       client.options.version = options.version ?? (Versions[ad.version] ? ad.version : CURRENT_VERSION)
       if (client.conLog) client.conLog(`Connecting to server ${ad.motd} (${ad.name}), version ${ad.version}`, client.options.version !== ad.version ? ` (as ${client.options.version})` : '')
+      client.emit('connect_allowed')
       connect(client)
     }, client)
   }
@@ -46,15 +47,17 @@ function connect (client) {
     sleep(500).then(() => client.queue('request_chunk_radius', { chunk_radius: client.viewDistance || 10 }))
   })
 
-  const KEEPALIVE_INTERVAL = 10 // Send tick sync packets every 10 ticks
+  // Send tick sync packets every 10 ticks
+  const keepAliveInterval = 10
+  const keepAliveIntervalBig = BigInt(keepAliveInterval)
   let keepalive
   client.tick = 0n
   client.once('spawn', () => {
     keepalive = setInterval(() => {
       // Client fills out the request_time and the server does response_time in its reply.
       client.queue('tick_sync', { request_time: client.tick, response_time: 0n })
-      client.tick += BigInt(KEEPALIVE_INTERVAL)
-    }, 50 * KEEPALIVE_INTERVAL)
+      client.tick += keepAliveIntervalBig
+    }, 50 * keepAliveInterval)
 
     client.on('tick_sync', async packet => {
       client.emit('heartbeat', packet.response_time)

src/server.js

@@ -33,6 +33,7 @@ class Server extends EventEmitter {
     if (this.options.protocolVersion < Options.MIN_VERSION) {
       throw new Error(`Protocol version < ${Options.MIN_VERSION} : ${this.options.protocolVersion}, too old`)
     }
+    this.compressionLevel = this.options.compressionLevel || 7
   }
 
   onOpenConnection = (conn) => {

src/serverPlayer.js

@@ -15,6 +15,7 @@ class Player extends Connection {
     this.deserializer = server.deserializer
     this.connection = connection
     this.options = server.options
+    this.compressionLevel = server.compressionLevel
 
     KeyExchange(this, server, server.options)
     Login(this, server, server.options)

src/transforms/encryption.js

@@ -36,12 +36,10 @@ function createEncryptor (client, iv) {
   // The send counter is represented as a little-endian 64-bit long and incremented after each packet.
 
   function process (chunk) {
-    Zlib.deflateRaw(chunk, { level: 7 }, (err, buffer) => {
-      if (err) throw err
-      const packet = Buffer.concat([buffer, computeCheckSum(buffer, client.sendCounter, client.secretKeyBytes)])
-      client.sendCounter++
-      client.cipher.write(packet)
-    })
+    const buffer = Zlib.deflateRawSync(chunk, { level: client.compressionLevel })
+    const packet = Buffer.concat([buffer, computeCheckSum(buffer, client.sendCounter, client.secretKeyBytes)])
+    client.sendCounter++
+    client.cipher.write(packet)
   }
 
   client.cipher.on('data', client.onEncryptedPacket)
@@ -72,10 +70,8 @@ function createDecryptor (client, iv) {
       return
     }
 
-    Zlib.inflateRaw(chunk, { chunkSize: 1024 * 1024 * 2 }, (err, buffer) => {
-      if (err) throw err
-      client.onDecryptedPacket(buffer)
-    })
+    const buffer = Zlib.inflateRawSync(chunk, { chunkSize: 512000 })
+    client.onDecryptedPacket(buffer)
   }
 
   client.decipher.on('data', verify)

src/transforms/framer.js

@@ -3,34 +3,30 @@ const zlib = require('zlib')
 
 // Concatenates packets into one batch packet, and adds length prefixs.
 class Framer {
-  constructor () {
+  constructor (compressionLevel) {
     // Encoding
     this.packets = []
-    this.compressionLevel = 7
+    this.compressionLevel = compressionLevel
   }
 
-  static decode (buf, cb) {
+  static decode (buf) {
     // Read header
     if (buf[0] !== 0xfe) throw Error('bad batch packet header ' + buf[0])
     const buffer = buf.slice(1)
 
-    // Decode the payload
-    zlib.inflateRaw(buffer, { chunkSize: 1024 * 1024 * 2 }, (err, inflated) => {
-      if (err) { // Try to decode without compression
-        Framer.getPackets(buffer)
-        return
-      }
-      cb(Framer.getPackets(inflated))
-    })
+    // Decode the payload with 512kb buffer
+    try {
+      const inflated = zlib.inflateRawSync(buffer, { chunkSize: 512000 })
+      return Framer.getPackets(inflated)
+    } catch (e) { // Try to decode without compression
+      return Framer.getPackets(buffer)
+    }
   }
 
-  encode (cb) {
+  encode () {
     const buf = Buffer.concat(this.packets)
-    zlib.deflateRaw(buf, { level: this.compressionLevel }, (err, def) => {
-      if (err) throw err
-      const ret = Buffer.concat([Buffer.from([0xfe]), def])
-      cb(ret)
-    })
+    const def = zlib.deflateRawSync(buf, { level: this.compressionLevel })
+    return Buffer.concat([Buffer.from([0xfe]), def])
   }
 
   addEncodedPacket (chunk) {