Async JS packet handling + native compression.

retrixe · retrixe · commit a66a917cf813 · 2022-05-16T01:54:38.000+05:30
Native compression is actually slower? So It's disabled at the moment.
diff --git a/android/app/build.gradle b/android/app/build.gradle
@@ -1,4 +1,5 @@
 apply plugin: "com.android.application"
+apply plugin: "kotlin-android"
 
 import com.android.build.OutputFile
 import org.apache.tools.ant.taskdefs.condition.Os
@@ -260,6 +261,7 @@ android {
 
 dependencies {
     implementation fileTree(dir: "libs", include: ["*.jar"])
+    implementation "org.jetbrains.kotlin:kotlin-stdlib-jdk7:$kotlinVersion"
 
     //noinspection GradleDynamicVersion
     implementation "com.facebook.react:react-native:+"  // From node_modules
diff --git a/android/app/src/main/java/com/enderchat/CompressionModule.kt b/android/app/src/main/java/com/enderchat/CompressionModule.kt
@@ -0,0 +1,62 @@
+package com.enderchat
+
+import android.util.Base64
+import com.facebook.react.bridge.Promise
+import com.facebook.react.bridge.ReactApplicationContext
+import com.facebook.react.bridge.ReactContextBaseJavaModule
+import com.facebook.react.bridge.ReactMethod
+import java.io.ByteArrayOutputStream
+import java.util.zip.Deflater
+import java.util.zip.Inflater
+
+class CompressionModule(reactContext: ReactApplicationContext)
+    : ReactContextBaseJavaModule(reactContext) {
+
+    override fun getName() = "CompressionModule"
+
+    @ReactMethod fun compressData(data: String, promise: Promise) {
+        val handler = (reactApplicationContext.currentActivity as MainActivity).dataTransformsThreadHandler
+        val bytes = Base64.decode(data, Base64.DEFAULT)
+        handler.post {
+            try {
+                ByteArrayOutputStream(bytes.size).use {
+                    val deflater = Deflater().apply {
+                        setStrategy(Deflater.DEFAULT_STRATEGY)
+                        setLevel(Deflater.DEFAULT_COMPRESSION)
+                        setInput(bytes)
+                        finish()
+                    }
+                    val buffer = ByteArray(1024)
+                    while (!deflater.finished()) {
+                        val count = deflater.deflate(buffer)
+                        it.write(buffer, 0, count)
+                    }
+                    promise.resolve(Base64.encodeToString(it.toByteArray(), Base64.DEFAULT))
+                }
+            } catch (e: Exception) {
+                promise.reject(e)
+            }
+        }
+    }
+
+    @ReactMethod fun decompressData(data: String, promise: Promise) {
+        val handler = (reactApplicationContext.currentActivity as MainActivity).dataTransformsThreadHandler
+        val bytes = Base64.decode(data, Base64.DEFAULT)
+        handler.post {
+            try {
+                ByteArrayOutputStream(bytes.size).use {
+                    val inflater = Inflater().apply { setInput(bytes) }
+                    val buffer = ByteArray(1024)
+                    while (!inflater.finished()) {
+                        val count = inflater.inflate(buffer)
+                        it.write(buffer, 0, count)
+                    }
+                    inflater.end()
+                    promise.resolve(Base64.encodeToString(it.toByteArray(), Base64.DEFAULT))
+                }
+            } catch (e: Exception) {
+                promise.reject(e)
+            }
+        }
+    }
+}
diff --git a/android/app/src/main/java/com/enderchat/CompressionPackage.kt b/android/app/src/main/java/com/enderchat/CompressionPackage.kt
@@ -0,0 +1,19 @@
+package com.enderchat
+
+import android.view.View
+import com.facebook.react.ReactPackage
+import com.facebook.react.bridge.NativeModule
+import com.facebook.react.bridge.ReactApplicationContext
+import com.facebook.react.uimanager.ReactShadowNode
+import com.facebook.react.uimanager.ViewManager
+
+class CompressionPackage : ReactPackage {
+
+    override fun createViewManagers(
+        reactContext: ReactApplicationContext
+    ): MutableList<ViewManager<View, ReactShadowNode<*>>> = mutableListOf()
+
+    override fun createNativeModules(
+        reactContext: ReactApplicationContext
+    ): MutableList<NativeModule> = listOf(CompressionModule(reactContext)).toMutableList()
+}
diff --git a/android/app/src/main/java/com/enderchat/MainActivity.java b/android/app/src/main/java/com/enderchat/MainActivity.java
@@ -1,11 +1,16 @@
 package com.enderchat;
 
 import android.os.Bundle;
+import android.os.Handler;
+import android.os.HandlerThread;
+import android.os.Process;
 import com.facebook.react.ReactActivity;
 import com.facebook.react.ReactActivityDelegate;
 import com.facebook.react.ReactRootView;
 
 public class MainActivity extends ReactActivity {
+  private HandlerThread dataHandlerThread;
+  private Handler dataTransformsThreadHandler;
 
   /**
    * Returns the name of the main component registered from JavaScript. This is used to schedule
@@ -42,5 +47,19 @@ protected ReactRootView createRootView() {
   @Override
   protected void onCreate(Bundle savedInstanceState) {
     super.onCreate(null);
+    dataHandlerThread = new HandlerThread("enderchat-data-handler-thread",
+          Process.THREAD_PRIORITY_MORE_FAVORABLE);
+    dataHandlerThread.start();
+    dataTransformsThreadHandler = new Handler(dataHandlerThread.getLooper());
+  }
+
+  @Override
+  protected void onDestroy() {
+    super.onDestroy();
+    dataHandlerThread.quitSafely();
+  }
+
+  public Handler getDataTransformsThreadHandler() {
+    return dataTransformsThreadHandler;
   }
 }
diff --git a/android/app/src/main/java/com/enderchat/MainApplication.java b/android/app/src/main/java/com/enderchat/MainApplication.java
@@ -29,6 +29,7 @@ protected List<ReactPackage> getPackages() {
           // Packages that cannot be autolinked yet can be added manually here, for example:
           // packages.add(new MyReactNativePackage());
           packages.add(new NavBarColorPackage());
+          packages.add(new CompressionPackage());
           return packages;
         }
 
diff --git a/android/build.gradle b/android/build.gradle
@@ -4,6 +4,7 @@ import org.apache.tools.ant.taskdefs.condition.Os
 
 buildscript {
     ext {
+        kotlinVersion = '1.6.20'
         buildToolsVersion = "31.0.0"
         minSdkVersion = 21
         compileSdkVersion = 31
@@ -27,6 +28,7 @@ buildscript {
         classpath("de.undercouch:gradle-download-task:4.1.2")
         // NOTE: Do not place your application dependencies here; they belong
         // in the individual module build.gradle files
+        classpath("org.jetbrains.kotlin:kotlin-gradle-plugin:${project.ext.kotlinVersion}")
     }
 }
 
diff --git a/package-lock.json b/package-lock.json
diff --git a/package.json b/package.json
@@ -34,6 +34,7 @@
     "react-native-tcp": "github:aprock/react-native-tcp",
     "react-native-vector-icons": "^7.1.0",
     "react-native-webview": "^11.17.1",
+    "semaphore-async-await": "^1.5.1",
     "stream-browserify": "^3.0.0"
   },
   "devDependencies": {
diff --git a/src/minecraft/connection.ts b/src/minecraft/connection.ts
@@ -1,3 +1,7 @@
+import { InteractionManager } from 'react-native'
+import Semaphore from 'semaphore-async-await'
+import net from 'react-native-tcp'
+import events from 'events'
 import {
   Cipher,
   createCipheriv,
@@ -6,9 +10,6 @@ import {
   Decipher,
   publicEncrypt
 } from 'react-native-crypto'
-import { InteractionManager } from 'react-native'
-import net from 'react-native-tcp'
-import events from 'events'
 import {
   concatPacketData,
   makeBaseCompressedPacket,
@@ -67,8 +68,9 @@ export class ServerConnection extends events.EventEmitter {
     data: Buffer,
     cb?: ((err?: Error | undefined) => void) | undefined
   ): Promise<boolean> {
+    const compressionThreshold = this.compressionThreshold
     const packet = this.compressionEnabled
-      ? makeBaseCompressedPacket(this.compressionThreshold, packetId, data)
+      ? await makeBaseCompressedPacket(compressionThreshold, packetId, data)
       : makeBasePacket(packetId, data)
     const toWrite = this.aesCipher ? this.aesCipher.update(packet) : packet
     return this.socket.write(toWrite, cb)
@@ -126,23 +128,24 @@ const initiateConnection = async (opts: ConnectionOptions) => {
       if (!resolved) reject(err)
       else conn.emit('error', err)
     })
+    const lock = new Semaphore(1)
     socket.on('data', newData => {
       // Handle timeout after 20 seconds of no data.
       if (conn.disconnectTimer) clearTimeout(conn.disconnectTimer)
       conn.disconnectTimer = setTimeout(() => conn.close(), 20000)
       // Run after interactions to improve user experience.
-      InteractionManager.runAfterInteractions(() => {
+      InteractionManager.runAfterInteractions(async () => {
+        await lock.acquire()
         try {
           // Note: the entire packet is encrypted, including the length fields and the packet's data.
           // https://github.com/PrismarineJS/node-minecraft-protocol/blob/master/src/transforms/encryption.js
           let finalData = newData
           if (conn.aesDecipher) finalData = conn.aesDecipher.update(newData)
           // Buffer data for read.
           conn.bufferedData = Buffer.concat([conn.bufferedData, finalData])
-          // ;(async () => { This would need a mutex.
           while (true) {
             const packet = conn.compressionEnabled
-              ? parseCompressedPacket(conn.bufferedData)
+              ? await parseCompressedPacket(conn.bufferedData)
               : parsePacket(conn.bufferedData)
             if (packet) {
               // Remove packet from buffered data.
@@ -233,6 +236,7 @@ const initiateConnection = async (opts: ConnectionOptions) => {
         } catch (err) {
           conn.emit('error', err)
         }
+        lock.release()
       }).then(() => {}, console.error)
     })
   })
diff --git a/src/minecraft/native/compression.ts b/src/minecraft/native/compression.ts
@@ -0,0 +1,22 @@
+import { NativeModules } from 'react-native'
+import zlib from 'zlib'
+
+const { CompressionModule } = NativeModules
+const isCompressionModuleAvailable = false
+// compressionModule?.compressData && CompressionModule.decompressData
+
+export const compressData = async (data: Buffer): Promise<Buffer> => {
+  if (isCompressionModuleAvailable) {
+    return CompressionModule.compressData(data.toString('base64')).then(
+      (res: string) => Buffer.from(res, 'base64')
+    )
+  } else return zlib.deflateSync(data)
+}
+
+export const decompressData = async (data: Buffer): Promise<Buffer> => {
+  if (isCompressionModuleAvailable) {
+    return CompressionModule.decompressData(data.toString('base64')).then(
+      (res: string) => Buffer.from(res, 'base64')
+    )
+  } else return zlib.unzipSync(data, { finishFlush: 2 })
+}
diff --git a/src/minecraft/packet.ts b/src/minecraft/packet.ts
@@ -1,6 +1,6 @@
-import zlib from 'zlib'
 import { Buffer } from 'buffer'
 import { toggleEndian, encodeString, readVarInt, writeVarInt } from './utils'
+import { compressData, decompressData } from './native/compression'
 
 export const makeBasePacket = (packetId: number, data: Buffer) => {
   const finalData = Buffer.concat([writeVarInt(packetId), data])
@@ -9,7 +9,7 @@ export const makeBasePacket = (packetId: number, data: Buffer) => {
   return Buffer.concat([finalDataLength, finalData])
 }
 
-export const makeBaseCompressedPacket = (
+export const makeBaseCompressedPacket = async (
   threshold: number,
   packetId: number,
   data: Buffer
@@ -24,7 +24,7 @@ export const makeBaseCompressedPacket = (
   /* : await new Promise((resolve, reject) => {
         zlib.deflate(finalData, (err, res) => err ? reject(err) : resolve(res))
       }) */
-  const dataToSend = toCompress ? zlib.deflateSync(finalData) : finalData
+  const dataToSend = toCompress ? await compressData(finalData) : finalData
   return makeBasePacket(dataLength, dataToSend)
 }
 
@@ -73,7 +73,9 @@ export const parsePacket = (packet: Buffer): Packet | undefined => {
   } catch (e) {} // If the packet is incomplete, readVarInt could error, so no packet parsed.
 }
 
-export const parseCompressedPacket = (packet: Buffer): Packet | undefined => {
+export const parseCompressedPacket = async (
+  packet: Buffer
+): Promise<Packet | undefined> => {
   const dissect = parsePacket(packet)
   if (!dissect) return
   else if (dissect.id === 0) {
@@ -92,7 +94,7 @@ export const parseCompressedPacket = (packet: Buffer): Packet | undefined => {
   const dataLength = dissect.id
   let dataWithId: Buffer
   try {
-    dataWithId = zlib.unzipSync(dissect.data, { finishFlush: 2 })
+    dataWithId = await decompressData(dissect.data)
   } catch (e) {
     console.error(`problem inflating chunk
 uncompressed length: ${dataLength}
diff --git a/yarn.lock b/yarn.lock
@@ -7568,6 +7568,11 @@
     "loose-envify" "^1.1.0"
     "object-assign" "^4.1.1"
 
+"semaphore-async-await@^1.5.1":
+  "integrity" "sha1-hXvvXjZEYBykuVcLh+nfXKEpdPo="
+  "resolved" "https://registry.npmjs.org/semaphore-async-await/-/semaphore-async-await-1.5.1.tgz"
+  "version" "1.5.1"
+
 "semver@^5.5.0":
   "integrity" "sha512-sauaDf/PZdVgrLTNYHRtpXa1iRiKcaebiKQ1BJdpQlWH2lCvexQdX55snPFyK7QzpudqbCI0qXFfOasHdyNDGQ=="
   "resolved" "https://registry.npmjs.org/semver/-/semver-5.7.1.tgz"

Original file line number	Diff line number	Diff line change
`@@ -29,6 +29,7 @@ protected List<ReactPackage> getPackages() {`
`29`	`29`	`// Packages that cannot be autolinked yet can be added manually here, for example:`
`30`	`30`	`// packages.add(new MyReactNativePackage());`
`31`	`31`	`packages.add(new NavBarColorPackage());`
	`32`	`+ packages.add(new CompressionPackage());`
`32`	`33`	`return packages;`
`33`	`34`	`}`
`34`	`35`
Original file line number	Diff line number	Diff line change
`@@ -4,6 +4,7 @@ import org.apache.tools.ant.taskdefs.condition.Os`
`4`	`4`
`5`	`5`	`buildscript {`
`6`	`6`	`ext {`
	`7`	`+ kotlinVersion = '1.6.20'`
`7`	`8`	`buildToolsVersion = "31.0.0"`
`8`	`9`	`minSdkVersion = 21`
`9`	`10`	`compileSdkVersion = 31`
`@@ -27,6 +28,7 @@ buildscript {`
`27`	`28`	`classpath("de.undercouch:gradle-download-task:4.1.2")`
`28`	`29`	`// NOTE: Do not place your application dependencies here; they belong`
`29`	`30`	`// in the individual module build.gradle files`
	`31`	`+ classpath("org.jetbrains.kotlin:kotlin-gradle-plugin:${project.ext.kotlinVersion}")`
`30`	`32`	`}`
`31`	`33`	`}`
`32`	`34`