feat: CompressionStream

example/server/createServer.ts

@@ -20,10 +20,19 @@ export function createServer(payload: string | Bun.BufferSource, port: number) {
             return new Response('No body provided', { status: 400 })
           }
 
+          const writer = Bun.file('uploaded_file').writer()
+
           for await (const chunk of req.body) {
-            console.log('Chunk:', chunk)
+            // Write each chunk to file
+            writer.write(chunk)
+            // Debug
+            console.log('Chunk saved:', chunk.length, 'bytes')
           }
 
+          await writer.end()
+
+          console.log('Upload complete, file saved')
+
           return new Response('Upload successful', { status: 200 })
         } catch (error) {
           console.error('Upload error:', error)

example/tests/benchmark.tsx

@@ -9,7 +9,12 @@ import {
   TouchableOpacity,
   View,
 } from 'react-native'
-import { fetch, showOpenFilePicker, WebSocket as FastWS } from 'react-native-fast-io'
+import {
+  CompressionStream,
+  fetch,
+  showOpenFilePicker,
+  WebSocket as FastWS,
+} from 'react-native-fast-io'
 
 import {
   CHAT_PAYLOAD,
@@ -443,21 +448,35 @@ const styles = StyleSheet.create({
   },
 })
 
-// tbd: playground
+// EXAMPLE 1
 setTimeout(async () => {
-  const files = await showOpenFilePicker({
-    startIn: 'desktop',
-    types: [{ description: 'PDF files', accept: { 'application/pdf': ['.pdf'] } }],
+  // File System API - https://developer.mozilla.org/en-US/docs/Web/API/File_System_API
+  const [fileHandle] = await showOpenFilePicker()
+
+  // Get `File` object (file is not loaded into memory)
+  const file = await fileHandle.getFile()
+
+  // File is streamed to the server
+  await fetch('http://localhost:3002/upload', {
+    method: 'POST',
+    body: file,
   })
+}, 2000)
+
+// EXAMPLE 2
+setTimeout(async () => {
+  // File System API - https://developer.mozilla.org/en-US/docs/Web/API/File_System_API
+  const [fileHandle] = await showOpenFilePicker()
 
-  if (!files[0]) {
-    console.log('No file')
-  }
+  // Get `File` object (file is not loaded into memory)
+  const file = await fileHandle.getFile()
 
-  const file = await files[0].getFile()
+  // You can also transform the stream in JavaScript, still no loading, all lazy
+  const compressed = file.stream().pipeThrough(new CompressionStream('gzip'))
 
+  // File is streamed to the server
   await fetch('http://localhost:3002/upload', {
     method: 'POST',
-    body: file.stream(),
+    body: compressed,
   })
 }, 2000)

packages/react-native-fast-io/ios/HybridCompressor.swift

@@ -0,0 +1,175 @@
+//
+//  HybridGzipCompressor.swift
+//  FastIO
+//
+//  Created by Mike Grabowski on 11/11/2024.
+//
+
+import Foundation
+import NitroModules
+import Compression
+
+class HybridCompressor : HybridCompressorSpec {
+  private var stream: compression_stream
+  private var status: compression_status
+
+  private var format: CompressionAlgorithm
+  private var totalSize: UInt32 = 0
+
+  init(algorithm: CompressionAlgorithm) {
+    stream = compression_stream()
+    status = compression_stream_init(&stream, COMPRESSION_STREAM_ENCODE, COMPRESSION_ZLIB)
+    format = algorithm
+  }
+
+  deinit {
+    compression_stream_destroy(&stream)
+  }
+
+  private func compressBuffer(source: UnsafePointer<UInt8>, sourceSize: Int, finalize: Bool = false) throws -> ArrayBufferHolder {
+    let headerSize: Int = if totalSize == 0 {
+      switch format {
+      case .gzip: 10
+      case .deflate: 2
+      case .deflateRaw: 0
+      }
+    } else {
+      0
+    }
+    let footerSize = (format == .gzip && finalize) ? 8 : 0
+
+    let destBufferSize = 64 * 1024 + headerSize + footerSize
+    let destBuffer = UnsafeMutablePointer<UInt8>.allocate(capacity: destBufferSize)
+
+    if headerSize > 0 {
+      switch format {
+      case .gzip:
+        let header = getGzipHeader()
+        destBuffer.update(from: header, count: header.count)
+      case .deflate:
+        let header = getDeflateHeader()
+        destBuffer.update(from: header, count: header.count)
+      default:
+        break
+      }
+    }
+
+    if format == .gzip && sourceSize > 0 {
+      updateCRC32(data: source, size: sourceSize)
+    }
+
+    totalSize = (totalSize &+ UInt32(sourceSize)) & 0xffffffff
+
+    stream.src_ptr = source
+    stream.src_size = sourceSize
+    stream.dst_ptr = destBuffer.advanced(by: headerSize)
+    stream.dst_size = 64 * 1024
+
+    status = compression_stream_process(&stream, Int32(finalize ? COMPRESSION_STREAM_FINALIZE.rawValue : 0))
+
+    guard status != COMPRESSION_STATUS_ERROR else {
+      destBuffer.deallocate()
+      throw RuntimeError.error(withMessage: "Compression error")
+    }
+
+    guard stream.src_size == 0 else {
+      destBuffer.deallocate()
+      throw RuntimeError.error(withMessage: "Unexpected remaining input data.")
+    }
+
+    let currentOffset = headerSize + (64 * 1024 - stream.dst_size)
+
+    if footerSize > 0 {
+      let footer = getGzipFooter()
+      destBuffer.advanced(by: currentOffset).update(from: footer, count: footer.count)
+    }
+
+    let deleteFunc = SwiftClosure {
+      destBuffer.deallocate()
+    }
+
+    return ArrayBufferHolder.makeBuffer(destBuffer, currentOffset + footerSize, deleteFunc)
+  }
+
+  func compress(chunk: ArrayBufferHolder) throws -> ArrayBufferHolder {
+    return try compressBuffer(
+      source: UnsafePointer(chunk.data.assumingMemoryBound(to: UInt8.self)),
+      sourceSize: chunk.size
+    )
+  }
+
+  func finalize() throws -> ArrayBufferHolder {
+    let emptyBuffer = UnsafeMutablePointer<UInt8>.allocate(capacity: 1)
+    defer {
+      emptyBuffer.deallocate()
+    }
+
+    return try compressBuffer(
+      source: UnsafePointer(emptyBuffer),
+      sourceSize: 0,
+      finalize: true
+    )
+  }
+
+  /* Gzip */
+  private var crc32: UInt32 = 0
+
+  private func getGzipHeader() -> [UInt8] {
+    // GZIP header format:
+    // 1F 8B - Magic number
+    // 08    - Deflate compression method
+    // 00    - Flags
+    // 00 00 00 00 - Timestamp
+    // 00    - Extra flags
+    // FF    - OS (unknown)
+    return [0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0xff]
+  }
+
+  private func getGzipFooter() -> [UInt8] {
+    var footer = [UInt8](repeating: 0, count: 8)
+    // CRC32 (4 bytes)
+    footer[0] = UInt8(crc32 & 0xff)
+    footer[1] = UInt8((crc32 >> 8) & 0xff)
+    footer[2] = UInt8((crc32 >> 16) & 0xff)
+    footer[3] = UInt8((crc32 >> 24) & 0xff)
+    // Input size modulo 2^32 (4 bytes)
+    footer[4] = UInt8(totalSize & 0xff)
+    footer[5] = UInt8((totalSize >> 8) & 0xff)
+    footer[6] = UInt8((totalSize >> 16) & 0xff)
+    footer[7] = UInt8((totalSize >> 24) & 0xff)
+    return footer
+  }
+
+  private func updateCRC32(data: UnsafePointer<UInt8>, size: Int) {
+    var crc = ~crc32
+    for i in 0..<size {
+      let byte = data[i]
+      crc = (crc >> 8) ^ crcTable[Int((crc & 0xFF) ^ UInt32(byte))]
+    }
+    crc32 = ~crc
+  }
+
+  /* Deflate */
+  private func getDeflateHeader() -> [UInt8] {
+    // Deflate header (CMF, FLG)
+    return [0x78, 0x9c]
+  }
+
+  var hybridContext = margelo.nitro.HybridContext()
+  var memorySize: Int {
+    return getSizeOf(self)
+  }
+}
+
+// CRC32 lookup table
+private let crcTable: [UInt32] = {
+  var table = [UInt32](repeating: 0, count: 256)
+  for i in 0..<256 {
+    var crc = UInt32(i)
+    for _ in 0..<8 {
+      crc = (crc >> 1) ^ ((crc & 1) == 1 ? 0xEDB88320 : 0)
+    }
+    table[i] = crc
+  }
+  return table
+}()

packages/react-native-fast-io/ios/HybridCompressorFactory.swift

@@ -0,0 +1,19 @@
+//
+//  HybridCompressorFactory.swift
+//  FastIO
+//
+//  Created by Mike Grabowski on 11/11/2024.
+//
+
+import Foundation
+
+class HybridCompressorFactory : HybridCompressorFactorySpec {
+  func create(algorithm: CompressionAlgorithm) throws -> (any HybridCompressorSpec) {
+    return HybridCompressor(algorithm: algorithm)
+  }
+
+  var hybridContext = margelo.nitro.HybridContext()
+  var memorySize: Int {
+    return getSizeOf(self)
+  }
+}

packages/react-native-fast-io/ios/HybridDuplexStream.swift

@@ -15,7 +15,7 @@ class HybridDuplexStream : HybridDuplexStreamSpec {
     var inputStreamRef: InputStream? = InputStream()
     var outputStreamRef: OutputStream? = OutputStream(toMemory: ())
 
-    Stream.getBoundStreams(withBufferSize: 4096, inputStream: &inputStreamRef, outputStream: &outputStreamRef)
+    Stream.getBoundStreams(withBufferSize: 64 * 1024, inputStream: &inputStreamRef, outputStream: &outputStreamRef)
 
     guard let inputStreamRef, let outputStreamRef else {
       fatalError("Could not create streams")

packages/react-native-fast-io/nitro.json

@@ -20,6 +20,9 @@
     },
     "DuplexStream": {
       "swift": "HybridDuplexStream"
+    },
+    "CompressorFactory": {
+      "swift": "HybridCompressorFactory"
     }
   }
 }

packages/react-native-fast-io/nitrogen/generated/android/FastIO+autolinking.cmake

@@ -31,6 +31,8 @@ target_sources(
   ../nitrogen/generated/shared/c++/HybridNetworkSpec.cpp
   ../nitrogen/generated/shared/c++/HybridInputStreamSpec.cpp
   ../nitrogen/generated/shared/c++/HybridOutputStreamSpec.cpp
+  ../nitrogen/generated/shared/c++/HybridCompressorFactorySpec.cpp
+  ../nitrogen/generated/shared/c++/HybridCompressorSpec.cpp
   ../nitrogen/generated/shared/c++/HybridDuplexStreamSpec.cpp
   ../nitrogen/generated/shared/c++/HybridWebSocketSpec.cpp
   ../nitrogen/generated/shared/c++/HybridWebSocketManagerSpec.cpp

packages/react-native-fast-io/nitrogen/generated/ios/FastIO-Swift-Cxx-Bridge.cpp

@@ -9,6 +9,8 @@
 
 // Include C++ implementation defined types
 #include "FastIO-Swift-Cxx-Umbrella.hpp"
+#include "HybridCompressorFactorySpecSwift.hpp"
+#include "HybridCompressorSpecSwift.hpp"
 #include "HybridDuplexStreamSpecSwift.hpp"
 #include "HybridFileSystemSpecSwift.hpp"
 #include "HybridInputStreamSpecSwift.hpp"
@@ -84,6 +86,38 @@ namespace margelo::nitro::fastio::bridge::swift {
     return FastIO::HybridOutputStreamSpecCxxUnsafe::toUnsafe(swiftPart);
   }
 
+  // pragma MARK: std::shared_ptr<margelo::nitro::fastio::HybridCompressorSpec>
+  std::shared_ptr<margelo::nitro::fastio::HybridCompressorSpec> create_std__shared_ptr_margelo__nitro__fastio__HybridCompressorSpec_(void* _Nonnull swiftUnsafePointer) {
+    FastIO::HybridCompressorSpecCxx swiftPart = FastIO::HybridCompressorSpecCxxUnsafe::fromUnsafe(swiftUnsafePointer);
+    return HybridContext::getOrCreate<margelo::nitro::fastio::HybridCompressorSpecSwift>(swiftPart);
+  }
+  void* _Nonnull get_std__shared_ptr_margelo__nitro__fastio__HybridCompressorSpec_(std__shared_ptr_margelo__nitro__fastio__HybridCompressorSpec_ cppType) {
+    std::shared_ptr<margelo::nitro::fastio::HybridCompressorSpecSwift> swiftWrapper = std::dynamic_pointer_cast<margelo::nitro::fastio::HybridCompressorSpecSwift>(cppType);
+  #ifdef NITRO_DEBUG
+    if (swiftWrapper == nullptr) [[unlikely]] {
+      throw std::runtime_error("Class \"HybridCompressorSpec\" is not implemented in Swift!");
+    }
+  #endif
+    FastIO::HybridCompressorSpecCxx swiftPart = swiftWrapper->getSwiftPart();
+    return FastIO::HybridCompressorSpecCxxUnsafe::toUnsafe(swiftPart);
+  }
+
+  // pragma MARK: std::shared_ptr<margelo::nitro::fastio::HybridCompressorFactorySpec>
+  std::shared_ptr<margelo::nitro::fastio::HybridCompressorFactorySpec> create_std__shared_ptr_margelo__nitro__fastio__HybridCompressorFactorySpec_(void* _Nonnull swiftUnsafePointer) {
+    FastIO::HybridCompressorFactorySpecCxx swiftPart = FastIO::HybridCompressorFactorySpecCxxUnsafe::fromUnsafe(swiftUnsafePointer);
+    return HybridContext::getOrCreate<margelo::nitro::fastio::HybridCompressorFactorySpecSwift>(swiftPart);
+  }
+  void* _Nonnull get_std__shared_ptr_margelo__nitro__fastio__HybridCompressorFactorySpec_(std__shared_ptr_margelo__nitro__fastio__HybridCompressorFactorySpec_ cppType) {
+    std::shared_ptr<margelo::nitro::fastio::HybridCompressorFactorySpecSwift> swiftWrapper = std::dynamic_pointer_cast<margelo::nitro::fastio::HybridCompressorFactorySpecSwift>(cppType);
+  #ifdef NITRO_DEBUG
+    if (swiftWrapper == nullptr) [[unlikely]] {
+      throw std::runtime_error("Class \"HybridCompressorFactorySpec\" is not implemented in Swift!");
+    }
+  #endif
+    FastIO::HybridCompressorFactorySpecCxx swiftPart = swiftWrapper->getSwiftPart();
+    return FastIO::HybridCompressorFactorySpecCxxUnsafe::toUnsafe(swiftPart);
+  }
+
   // pragma MARK: std::shared_ptr<margelo::nitro::fastio::HybridDuplexStreamSpec>
   std::shared_ptr<margelo::nitro::fastio::HybridDuplexStreamSpec> create_std__shared_ptr_margelo__nitro__fastio__HybridDuplexStreamSpec_(void* _Nonnull swiftUnsafePointer) {
     FastIO::HybridDuplexStreamSpecCxx swiftPart = FastIO::HybridDuplexStreamSpecCxxUnsafe::fromUnsafe(swiftUnsafePointer);