Add optional read acknowledge

.github/workflows/flutter.yml

@@ -0,0 +1,37 @@
+name: Flutter
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+    branches: [main]
+
+jobs:
+  test:
+    runs-on: ${{ matrix.os }}
+
+    strategy:
+      matrix:
+        os: [ubuntu-latest, windows-latest]
+        sdk: [stable, beta]
+
+    steps:
+      - uses: actions/checkout@v2
+      - uses: subosito/flutter-action@v2
+        with:
+          channdl: ${{ matrix.sdk }}
+
+      - name: Print Flutter SDK version
+        run: flutter --version
+
+      - name: Install dependencies
+        run: flutter pub get
+
+      - name: Verify formatting
+        run: flutter format --output=none --set-exit-if-changed .
+
+      - name: Analyze project source
+        run: flutter analyze
+
+      - name: Run tests
+        run: flutter test integration_test

example/integration_test/flutter_pty_test.dart

@@ -0,0 +1,35 @@
+import 'dart:convert';
+import 'dart:io';
+import 'dart:typed_data';
+
+import 'package:flutter_pty/flutter_pty.dart';
+import 'package:flutter_test/flutter_test.dart';
+
+String get shell {
+  if (Platform.isWindows) {
+    return 'cmd.exe';
+  }
+
+  if (Platform.isLinux || Platform.isMacOS) {
+    return 'bash';
+  }
+
+  return 'sh';
+}
+
+extension StringToUtf8 on String {
+  Uint8List toUtf8() {
+    return Uint8List.fromList(
+      utf8.encode(this),
+    );
+  }
+}
+
+void main() {
+  test('Pty works', () async {
+    final pty = Pty.start(shell);
+    final input = 'random input'.toUtf8();
+    pty.write(input);
+    expect(await pty.output.first, input);
+  });
+}

example/macos/Podfile.lock

@@ -25,4 +25,4 @@ SPEC CHECKSUMS:
 
 PODFILE CHECKSUM: 6eac6b3292e5142cfc23bdeb71848a40ec51c14c
 
-COCOAPODS: 1.11.2
+COCOAPODS: 1.11.3

example/pubspec.lock

@@ -103,7 +103,7 @@ packages:
       path: ".."
       relative: true
     source: path
-    version: "0.0.7"
+    version: "0.1.1"
   flutter_test:
     dependency: "direct dev"
     description: flutter

example/pubspec.yaml

@@ -50,6 +50,9 @@ dev_dependencies:
   flutter_test:
     sdk: flutter
 
+  integration_test:
+    sdk: flutter
+
   # The "flutter_lints" package below contains a set of recommended lints to
   # encourage good coding practices. The lint set provided by the package is
   # activated in the `analysis_options.yaml` file located at the root of your

lib/flutter_pty.dart

@@ -43,13 +43,15 @@ class Pty {
 
   /// Spawns a process in a pseudo-terminal. The arguments have the same meaning
   /// as in [Process.start].
+  /// [ackRead] indicates if the pty should wait for a call to [Pty.ackRead] before sending the next data.
   Pty.start(
     this.executable, {
     this.arguments = const [],
     String? workingDirectory,
     Map<String, String>? environment,
     int rows = 25,
     int columns = 80,
+    bool ackRead = false,
   }) {
     _ensureInitialized();
 
@@ -104,6 +106,7 @@ class Pty {
     options.ref.environment = envp.cast();
     options.ref.stdout_port = _stdoutPort.sendPort.nativePort;
     options.ref.exit_port = _exitPort.sendPort.nativePort;
+    options.ref.ackRead = ackRead ? 1 : 0;
 
     if (workingDirectory != null) {
       options.ref.working_directory = workingDirectory.toNativeUtf8().cast();
@@ -180,6 +183,13 @@ class Pty {
   bool kill([ProcessSignal signal = ProcessSignal.sigterm]) {
     return Process.killPid(pid, signal);
   }
+
+  /// indicates that a data chunk has been processed.
+  /// This is needed when ackRead is set to true as the pty will wait for this signal to happen
+  /// before any additional data is sent.
+  void ackRead() {
+    _bindings.pty_ack_read(_handle);
+  }
 }
 
 String? _getPtyError() {

lib/flutter_pty_bindings_generated.dart

@@ -405,6 +405,20 @@ class FlutterPtyBindings {
   late final _pty_write = _pty_writePtr.asFunction<
       void Function(ffi.Pointer<PtyHandle>, ffi.Pointer<ffi.Int8>, int)>();
 
+  void pty_ack_read(
+    ffi.Pointer<PtyHandle> handle,
+  ) {
+    return _pty_ack_read(
+      handle,
+    );
+  }
+
+  late final _pty_ack_readPtr =
+      _lookup<ffi.NativeFunction<ffi.Void Function(ffi.Pointer<PtyHandle>)>>(
+          'pty_ack_read');
+  late final _pty_ack_read =
+      _pty_ack_readPtr.asFunction<void Function(ffi.Pointer<PtyHandle>)>();
+
   int pty_resize(
     ffi.Pointer<PtyHandle> handle,
     int rows,
@@ -734,6 +748,9 @@ class PtyOptions extends ffi.Struct {
 
   @Dart_Port()
   external int exit_port;
+
+  @ffi.Uint8()
+  external int ackRead;
 }
 
 class PtyHandle extends ffi.Opaque {}

pubspec.yaml

@@ -14,9 +14,11 @@ dependencies:
   ffi: ^1.2.0-dev.0
 
 dev_dependencies:
-  ffigen: ^4.1.2
   flutter_test:
     sdk: flutter
+
+  ffigen: ^4.1.2
+
   flutter_lints: ^1.0.0
 
 # For information on the generic Dart part of this file, see the

src/flutter_pty.h

@@ -31,6 +31,8 @@ typedef struct PtyOptions
 
     Dart_Port exit_port;
 
+    bool ackRead;
+
 } PtyOptions;
 
 typedef struct PtyHandle PtyHandle;
@@ -39,6 +41,8 @@ FFI_PLUGIN_EXPORT PtyHandle *pty_create(PtyOptions *options);
 
 FFI_PLUGIN_EXPORT void pty_write(PtyHandle *handle, char *buffer, int length);
 
+FFI_PLUGIN_EXPORT void pty_ack_read(PtyHandle *handle);
+
 FFI_PLUGIN_EXPORT int pty_resize(PtyHandle *handle, int rows, int cols);
 
 FFI_PLUGIN_EXPORT int pty_getpid(PtyHandle *handle);

src/flutter_pty_unix.c

@@ -21,14 +21,22 @@ typedef struct PtyHandle
 
     int pid;
 
+    pthread_mutex_t mutex;
+
+    bool ackRead;
+
 } PtyHandle;
 
 typedef struct ReadLoopOptions
 {
     int fd;
 
+    pthread_mutex_t* mutex;
+
     Dart_Port port;
 
+    bool waitForReadAck;
+
 } ReadLoopOptions;
 
 static void *read_loop(void *arg)
@@ -39,6 +47,12 @@ static void *read_loop(void *arg)
 
     while (1)
     {
+        if(options->waitForReadAck)
+        {
+            // if we are in ack mode then we get a mutex here that is
+            // freed again once the chunk of data has been processed
+            pthread_mutex_lock(options->mutex);
+        }
         ssize_t n = read(options->fd, buffer, sizeof(buffer));
 
         if (n < 0)
@@ -64,13 +78,17 @@ static void *read_loop(void *arg)
     return NULL;
 }
 
-static void start_read_thread(int fd, Dart_Port port)
+static void start_read_thread(int fd, Dart_Port port, pthread_mutex_t* mutex, bool waitForReadAck)
 {
     ReadLoopOptions *options = malloc(sizeof(ReadLoopOptions));
 
     options->fd = fd;
 
     options->port = port;
+
+    options->mutex = mutex;
+
+    options->waitForReadAck = waitForReadAck;
 
     pthread_t _thread;
 
@@ -167,8 +185,10 @@ FFI_PLUGIN_EXPORT PtyHandle *pty_create(PtyOptions *options)
 
     handle->ptm = ptm;
     handle->pid = pid;
+    pthread_mutex_init( &handle->mutex, NULL );
+    handle->ackRead = options->ackRead;
 
-    start_read_thread(ptm, options->stdout_port);
+    start_read_thread(ptm, options->stdout_port, &handle->mutex, options->ackRead);
 
     start_wait_exit_thread(pid, options->exit_port);
 
@@ -180,6 +200,15 @@ FFI_PLUGIN_EXPORT void pty_write(PtyHandle *handle, char *buffer, int length)
     write(handle->ptm, buffer, length);
 }
 
+FFI_PLUGIN_EXPORT void pty_ack_read(PtyHandle *handle)
+{
+    if(handle->ackRead)
+    {
+        // frees the mutex so that the next chunk of data can be read
+        pthread_mutex_unlock( &handle->mutex );
+    }
+}
+
 FFI_PLUGIN_EXPORT int pty_resize(PtyHandle *handle, int rows, int cols)
 {
     struct winsize ws;

src/flutter_pty_win.c

@@ -156,6 +156,10 @@ typedef struct ReadLoopOptions
 
     Dart_Port port;
 
+    HANDLE hMutex;
+
+    BOOL ackRead;
+
 } ReadLoopOptions;
 
 static DWORD WINAPI read_loop(LPVOID arg)
@@ -168,6 +172,11 @@ static DWORD WINAPI read_loop(LPVOID arg)
     {
         DWORD readlen = 0;
 
+        if (options->ackRead)
+        {
+            WaitForSingleObject(options->hMutex, INFINITE);
+        }
+
         BOOL ok = ReadFile(options->fd, buffer, sizeof(buffer), &readlen, NULL);
 
         if (!ok)
@@ -192,12 +201,14 @@ static DWORD WINAPI read_loop(LPVOID arg)
     return 0;
 }
 
-static void start_read_thread(HANDLE fd, Dart_Port port)
+static void start_read_thread(HANDLE fd, Dart_Port port, HANDLE mutex, BOOL ackRead)
 {
     ReadLoopOptions *options = malloc(sizeof(ReadLoopOptions));
 
     options->fd = fd;
     options->port = port;
+    options->hMutex = mutex;
+    options->ackRead = ackRead;
 
     DWORD thread_id;
 
@@ -215,6 +226,7 @@ typedef struct WaitExitOptions
 
     Dart_Port port;
 
+    HANDLE hMutex;
 } WaitExitOptions;
 
 static DWORD WINAPI wait_exit_thread(LPVOID arg)
@@ -228,18 +240,20 @@ static DWORD WINAPI wait_exit_thread(LPVOID arg)
     GetExitCodeProcess(options->pid, &exit_code);
 
     CloseHandle(options->pid);
+    CloseHandle(options->hMutex);
 
     Dart_PostInteger_DL(options->port, exit_code);
 
     return 0;
 }
 
-static void start_wait_exit_thread(HANDLE pid, Dart_Port port)
+static void start_wait_exit_thread(HANDLE pid, Dart_Port port, HANDLE mutex)
 {
     WaitExitOptions *options = malloc(sizeof(WaitExitOptions));
 
     options->pid = pid;
     options->port = port;
+    options->hMutex = mutex;
 
     DWORD thread_id;
 
@@ -261,6 +275,10 @@ typedef struct PtyHandle
 
     HANDLE hProcess;
 
+    BOOL ackRead;
+
+    HANDLE hMutex;
+
 } PtyHandle;
 
 char *error_message = NULL;
@@ -386,9 +404,11 @@ FFI_PLUGIN_EXPORT PtyHandle *pty_create(PtyOptions *options)
 
     // CloseHandle(processInfo.hThread);
 
-    start_read_thread(outputReadSide, options->stdout_port);
+    HANDLE mutex = CreateMutex(NULL, FALSE, NULL);
 
-    start_wait_exit_thread(processInfo.hProcess, options->exit_port);
+    start_read_thread(outputReadSide, options->stdout_port, mutex, options->ackRead);
+
+    start_wait_exit_thread(processInfo.hProcess, options->exit_port, mutex);
 
     PtyHandle *pty = malloc(sizeof(PtyHandle));
 
@@ -402,6 +422,8 @@ FFI_PLUGIN_EXPORT PtyHandle *pty_create(PtyOptions *options)
     pty->outputReadSide = outputReadSide;
     pty->hPty = hPty;
     pty->hProcess = processInfo.hProcess;
+    pty->ackRead = options->ackRead;
+    pty->hMutex = mutex;
 
     return pty;
 }
@@ -417,6 +439,14 @@ FFI_PLUGIN_EXPORT void pty_write(PtyHandle *handle, char *buffer, int length)
     return;
 }
 
+FFI_PLUGIN_EXPORT void pty_ack_read(PtyHandle *handle)
+{
+    if (handle->ackRead)
+    {
+        ReleaseMutex(handle->hMutex);
+    }
+}
+
 FFI_PLUGIN_EXPORT int pty_resize(PtyHandle *handle, int rows, int cols)
 {
     COORD size;