JavaClient stream response body using byte arrays

pkgs/java_http/jnigen.yaml

@@ -12,6 +12,7 @@ class_path:
   - 'classes.jar'
 
 classes:
+  - 'java.io.BufferedInputStream'
   - 'java.io.InputStream'
   - 'java.io.OutputStream'
   - 'java.lang.System'

pkgs/java_http/lib/src/java_client.dart

@@ -12,6 +12,7 @@ import 'package:http/http.dart';
 import 'package:jni/jni.dart';
 import 'package:path/path.dart';
 
+import 'third_party/java/io/BufferedInputStream.dart';
 import 'third_party/java/lang/System.dart';
 import 'third_party/java/net/HttpURLConnection.dart';
 import 'third_party/java/net/URL.dart';
@@ -96,15 +97,15 @@ class JavaClient extends BaseClient {
   }
 
   // TODO: Rename _isolateMethod to something more descriptive.
-  void _isolateMethod(
+  Future<void> _isolateMethod(
     ({
       SendPort sendPort,
       Uint8List body,
       Map<String, String> headers,
       String method,
       Uri url,
     }) request,
-  ) {
+  ) async {
     final httpUrlConnection = URL
         .ctor3(request.url.toString().toJString())
         .openConnection()
@@ -140,7 +141,7 @@ class JavaClient extends BaseClient {
       responseHeaders,
     );
 
-    _responseBody(
+    await _responseBody(
       request.url,
       httpUrlConnection,
       request.sendPort,
@@ -230,35 +231,61 @@ class JavaClient extends BaseClient {
     return contentLength;
   }
 
-  void _responseBody(
+  Future<void> _responseBody(
     Uri requestUrl,
     HttpURLConnection httpUrlConnection,
     SendPort sendPort,
     int? expectedBodyLength,
-  ) {
+  ) async {
     final responseCode = httpUrlConnection.getResponseCode();
 
-    final inputStream = (responseCode >= 200 && responseCode <= 299)
-        ? httpUrlConnection.getInputStream()
-        : httpUrlConnection.getErrorStream();
+    final responseBodyStream = (responseCode >= 200 && responseCode <= 299)
+        ? BufferedInputStream(httpUrlConnection.getInputStream())
+        : BufferedInputStream(httpUrlConnection.getErrorStream());
 
-    int byte;
     var actualBodyLength = 0;
-    // TODO: inputStream.read() could throw IOException.
-    while ((byte = inputStream.read()) != -1) {
-      sendPort.send([byte]);
-      actualBodyLength++;
+    final bytesBuffer = JArray(jbyte.type, 4096);
+
+    while (true) {
+      // TODO: read1() could throw IOException.
+      final bytesCount =
+          responseBodyStream.read1(bytesBuffer, 0, bytesBuffer.length);
+
+      if (bytesCount == -1) {
+        break;
+      }
+
+      if (bytesCount == 0) {
+        // No more data is available without blocking so give other Isolates an
+        // opportunity to run.
+        await Future<void>.delayed(Duration.zero);
+        continue;
+      }
+
+      sendPort.send(bytesBuffer.toUint8List(length: bytesCount));
+      actualBodyLength += bytesCount;
     }
 
     if (expectedBodyLength != null && actualBodyLength < expectedBodyLength) {
       sendPort.send(ClientException('Unexpected end of body', requestUrl));
     }
 
-    inputStream.close();
+    responseBodyStream.close();
   }
 }
 
 extension on Uint8List {
   JArray<jbyte> toJArray() =>
       JArray(jbyte.type, length)..setRange(0, length, this);
 }
+
+extension on JArray<jbyte> {
+  Uint8List toUint8List({int? length}) {
+    length ??= this.length;
+    final list = Uint8List(length);
+    for (var i = 0; i < length; i++) {
+      list[i] = this[i];
+    }
+    return list;
+  }
+}

pkgs/java_http/lib/src/third_party/java/io/BufferedInputStream.dart

@@ -0,0 +1,248 @@
+// Autogenerated by jnigen. DO NOT EDIT!
+
+// ignore_for_file: annotate_overrides
+// ignore_for_file: camel_case_extensions
+// ignore_for_file: camel_case_types
+// ignore_for_file: constant_identifier_names
+// ignore_for_file: file_names
+// ignore_for_file: no_leading_underscores_for_local_identifiers
+// ignore_for_file: non_constant_identifier_names
+// ignore_for_file: overridden_fields
+// ignore_for_file: unnecessary_cast
+// ignore_for_file: unused_element
+// ignore_for_file: unused_field
+// ignore_for_file: unused_import
+// ignore_for_file: unused_shown_name
+
+import "dart:isolate" show ReceivePort;
+import "dart:ffi" as ffi;
+import "package:jni/internal_helpers_for_jnigen.dart";
+import "package:jni/jni.dart" as jni;
+
+import "InputStream.dart" as inputstream_;
+
+/// from: java.io.BufferedInputStream
+class BufferedInputStream extends jni.JObject {
+  @override
+  late final jni.JObjType<BufferedInputStream> $type = type;
+
+  BufferedInputStream.fromRef(
+    jni.JObjectPtr ref,
+  ) : super.fromRef(ref);
+
+  static final _class = jni.Jni.findJClass(r"java/io/BufferedInputStream");
+
+  /// The type which includes information such as the signature of this class.
+  static const type = $BufferedInputStreamType();
+  static final _id_buf = jni.Jni.accessors.getFieldIDOf(
+    _class.reference,
+    r"buf",
+    r"[B",
+  );
+
+  /// from: protected byte[] buf
+  /// The returned object must be deleted after use, by calling the `delete` method.
+  jni.JArray<jni.jbyte> get buf =>
+      const jni.JArrayType(jni.jbyteType()).fromRef(jni.Jni.accessors
+          .getField(reference, _id_buf, jni.JniCallType.objectType)
+          .object);
+
+  /// from: protected byte[] buf
+  /// The returned object must be deleted after use, by calling the `delete` method.
+  set buf(jni.JArray<jni.jbyte> value) =>
+      jni.Jni.env.SetObjectField(reference, _id_buf, value.reference);
+
+  static final _id_count = jni.Jni.accessors.getFieldIDOf(
+    _class.reference,
+    r"count",
+    r"I",
+  );
+
+  /// from: protected int count
+  int get count => jni.Jni.accessors
+      .getField(reference, _id_count, jni.JniCallType.intType)
+      .integer;
+
+  /// from: protected int count
+  set count(int value) => jni.Jni.env.SetIntField(reference, _id_count, value);
+
+  static final _id_pos = jni.Jni.accessors.getFieldIDOf(
+    _class.reference,
+    r"pos",
+    r"I",
+  );
+
+  /// from: protected int pos
+  int get pos => jni.Jni.accessors
+      .getField(reference, _id_pos, jni.JniCallType.intType)
+      .integer;
+
+  /// from: protected int pos
+  set pos(int value) => jni.Jni.env.SetIntField(reference, _id_pos, value);
+
+  static final _id_markpos = jni.Jni.accessors.getFieldIDOf(
+    _class.reference,
+    r"markpos",
+    r"I",
+  );
+
+  /// from: protected int markpos
+  int get markpos => jni.Jni.accessors
+      .getField(reference, _id_markpos, jni.JniCallType.intType)
+      .integer;
+
+  /// from: protected int markpos
+  set markpos(int value) =>
+      jni.Jni.env.SetIntField(reference, _id_markpos, value);
+
+  static final _id_marklimit = jni.Jni.accessors.getFieldIDOf(
+    _class.reference,
+    r"marklimit",
+    r"I",
+  );
+
+  /// from: protected int marklimit
+  int get marklimit => jni.Jni.accessors
+      .getField(reference, _id_marklimit, jni.JniCallType.intType)
+      .integer;
+
+  /// from: protected int marklimit
+  set marklimit(int value) =>
+      jni.Jni.env.SetIntField(reference, _id_marklimit, value);
+
+  static final _id_ctor = jni.Jni.accessors
+      .getMethodIDOf(_class.reference, r"<init>", r"(Ljava/io/InputStream;)V");
+
+  /// from: public void <init>(java.io.InputStream inputStream)
+  /// The returned object must be deleted after use, by calling the `delete` method.
+  factory BufferedInputStream(
+    inputstream_.InputStream inputStream,
+  ) {
+    return BufferedInputStream.fromRef(jni.Jni.accessors.newObjectWithArgs(
+        _class.reference, _id_ctor, [inputStream.reference]).object);
+  }
+
+  static final _id_ctor1 = jni.Jni.accessors
+      .getMethodIDOf(_class.reference, r"<init>", r"(Ljava/io/InputStream;I)V");
+
+  /// from: public void <init>(java.io.InputStream inputStream, int i)
+  /// The returned object must be deleted after use, by calling the `delete` method.
+  factory BufferedInputStream.ctor1(
+    inputstream_.InputStream inputStream,
+    int i,
+  ) {
+    return BufferedInputStream.fromRef(jni.Jni.accessors.newObjectWithArgs(
+        _class.reference,
+        _id_ctor1,
+        [inputStream.reference, jni.JValueInt(i)]).object);
+  }
+
+  static final _id_read =
+      jni.Jni.accessors.getMethodIDOf(_class.reference, r"read", r"()I");
+
+  /// from: public int read()
+  int read() {
+    return jni.Jni.accessors.callMethodWithArgs(
+        reference, _id_read, jni.JniCallType.intType, []).integer;
+  }
+
+  static final _id_read1 =
+      jni.Jni.accessors.getMethodIDOf(_class.reference, r"read", r"([BII)I");
+
+  /// from: public int read(byte[] bs, int i, int i1)
+  int read1(
+    jni.JArray<jni.jbyte> bs,
+    int i,
+    int i1,
+  ) {
+    return jni.Jni.accessors.callMethodWithArgs(
+        reference,
+        _id_read1,
+        jni.JniCallType.intType,
+        [bs.reference, jni.JValueInt(i), jni.JValueInt(i1)]).integer;
+  }
+
+  static final _id_skip =
+      jni.Jni.accessors.getMethodIDOf(_class.reference, r"skip", r"(J)J");
+
+  /// from: public long skip(long j)
+  int skip(
+    int j,
+  ) {
+    return jni.Jni.accessors.callMethodWithArgs(
+        reference, _id_skip, jni.JniCallType.longType, [j]).long;
+  }
+
+  static final _id_available =
+      jni.Jni.accessors.getMethodIDOf(_class.reference, r"available", r"()I");
+
+  /// from: public int available()
+  int available() {
+    return jni.Jni.accessors.callMethodWithArgs(
+        reference, _id_available, jni.JniCallType.intType, []).integer;
+  }
+
+  static final _id_mark =
+      jni.Jni.accessors.getMethodIDOf(_class.reference, r"mark", r"(I)V");
+
+  /// from: public void mark(int i)
+  void mark(
+    int i,
+  ) {
+    return jni.Jni.accessors.callMethodWithArgs(reference, _id_mark,
+        jni.JniCallType.voidType, [jni.JValueInt(i)]).check();
+  }
+
+  static final _id_reset =
+      jni.Jni.accessors.getMethodIDOf(_class.reference, r"reset", r"()V");
+
+  /// from: public void reset()
+  void reset() {
+    return jni.Jni.accessors.callMethodWithArgs(
+        reference, _id_reset, jni.JniCallType.voidType, []).check();
+  }
+
+  static final _id_markSupported = jni.Jni.accessors
+      .getMethodIDOf(_class.reference, r"markSupported", r"()Z");
+
+  /// from: public boolean markSupported()
+  bool markSupported() {
+    return jni.Jni.accessors.callMethodWithArgs(
+        reference, _id_markSupported, jni.JniCallType.booleanType, []).boolean;
+  }
+
+  static final _id_close =
+      jni.Jni.accessors.getMethodIDOf(_class.reference, r"close", r"()V");
+
+  /// from: public void close()
+  void close() {
+    return jni.Jni.accessors.callMethodWithArgs(
+        reference, _id_close, jni.JniCallType.voidType, []).check();
+  }
+}
+
+class $BufferedInputStreamType extends jni.JObjType<BufferedInputStream> {
+  const $BufferedInputStreamType();
+
+  @override
+  String get signature => r"Ljava/io/BufferedInputStream;";
+
+  @override
+  BufferedInputStream fromRef(jni.JObjectPtr ref) =>
+      BufferedInputStream.fromRef(ref);
+
+  @override
+  jni.JObjType get superType => const jni.JObjectType();
+
+  @override
+  final superCount = 1;
+
+  @override
+  int get hashCode => ($BufferedInputStreamType).hashCode;
+
+  @override
+  bool operator ==(Object other) {
+    return other.runtimeType == ($BufferedInputStreamType) &&
+        other is $BufferedInputStreamType;
+  }
+}

pkgs/java_http/lib/src/third_party/java/io/_package.dart

@@ -1,2 +1,3 @@
+export "BufferedInputStream.dart";
 export "InputStream.dart";
 export "OutputStream.dart";