From d8df20ad629fa17dde6f663ac595005f51fa34b3 Mon Sep 17 00:00:00 2001 From: "Victor C." <34163765+vicajilau@users.noreply.github.com> Date: Tue, 14 Jul 2026 05:33:25 +0200 Subject: [PATCH 1/5] feat(socket): add flush() to SSHSocket, SSHClient and SSHChannel --- lib/src/socket/ssh_socket.dart | 3 +++ lib/src/socket/ssh_socket_io.dart | 5 +++++ lib/src/ssh_channel.dart | 10 ++++++++++ lib/src/ssh_client.dart | 6 ++++++ lib/src/ssh_forward.dart | 7 +++++++ lib/src/ssh_transport.dart | 5 +++++ test/src/http/http_client_test.dart | 3 +++ test/src/ssh_auth_abort_error_test.dart | 3 +++ test/src/ssh_client_forward_dynamic_test.dart | 3 +++ test/src/ssh_client_ident_test.dart | 3 +++ test/src/ssh_client_run_with_result_test.dart | 3 +++ test/src/ssh_client_timeout_test.dart | 3 +++ test/src/ssh_transport_aead_test.dart | 3 +++ test/src/ssh_transport_version_test.dart | 3 +++ 14 files changed, 60 insertions(+) diff --git a/lib/src/socket/ssh_socket.dart b/lib/src/socket/ssh_socket.dart index 13a1d07..0c369ad 100644 --- a/lib/src/socket/ssh_socket.dart +++ b/lib/src/socket/ssh_socket.dart @@ -28,4 +28,7 @@ abstract class SSHSocket { Future close(); void destroy(); + + /// Force flush any buffered outgoing data. + Future flush() async {} } diff --git a/lib/src/socket/ssh_socket_io.dart b/lib/src/socket/ssh_socket_io.dart index a8f155b..7815054 100644 --- a/lib/src/socket/ssh_socket_io.dart +++ b/lib/src/socket/ssh_socket_io.dart @@ -37,6 +37,11 @@ class _SSHNativeSocket implements SSHSocket { _socket.destroy(); } + @override + Future flush() async { + await _socket.flush(); + } + @override String toString() { final address = '${_socket.remoteAddress.host}:${_socket.remotePort}'; diff --git a/lib/src/ssh_channel.dart b/lib/src/ssh_channel.dart index 8a3387f..708145a 100644 --- a/lib/src/ssh_channel.dart +++ b/lib/src/ssh_channel.dart @@ -31,6 +31,8 @@ class SSHChannelController { SSHChannel get channel => SSHChannel(this); + final Future Function()? onFlush; + SSHChannelController({ required this.localId, required this.localMaximumPacketSize, @@ -39,6 +41,7 @@ class SSHChannelController { required this.remoteInitialWindowSize, required this.remoteMaximumPacketSize, required this.sendMessage, + this.onFlush, this.printDebug, }) { if (remoteInitialWindowSize > 0) { @@ -404,6 +407,10 @@ class SSHChannelController { _remoteWindow -= data.bytes.length; } }); + + Future flush() async { + await onFlush?.call(); + } } class SSHChannel { @@ -434,6 +441,9 @@ class SSHChannel { sink.add(SSHChannelData(data, type: type)); } + /// Force flush any buffered outgoing data on this channel to the socket. + Future flush() => _controller.flush(); + void setRequestHandler(SSHChannelRequestHandler handler) { _controller._requestHandler = handler; } diff --git a/lib/src/ssh_client.dart b/lib/src/ssh_client.dart index fcc1e44..630388a 100644 --- a/lib/src/ssh_client.dart +++ b/lib/src/ssh_client.dart @@ -701,6 +701,11 @@ class SSHClient { _transport.close(); } + /// Force flush any buffered outgoing data to the socket. + Future flush() async { + await _transport.flush(); + } + /// Close all channels that are currently open. void _closeChannels() { for (final channel in _channels.values) { @@ -1376,6 +1381,7 @@ class SSHClient { remoteInitialWindowSize: remoteInitialWindowSize, remoteMaximumPacketSize: remoteMaximumPacketSize, sendMessage: _sendMessage, + onFlush: flush, printDebug: printDebug, ); diff --git a/lib/src/ssh_forward.dart b/lib/src/ssh_forward.dart index 2a03f2c..c9d9936 100644 --- a/lib/src/ssh_forward.dart +++ b/lib/src/ssh_forward.dart @@ -76,6 +76,13 @@ class SSHForwardChannel implements SSHSocket { void destroy() { _channel.destroy(); } + + /// Force flush any buffered outgoing data. + @override + Future flush() async { + await Future.microtask(() {}); + await _channel.flush(); + } } class SSHX11Channel extends SSHForwardChannel { diff --git a/lib/src/ssh_transport.dart b/lib/src/ssh_transport.dart index d0e234f..ff46a42 100644 --- a/lib/src/ssh_transport.dart +++ b/lib/src/ssh_transport.dart @@ -464,6 +464,11 @@ class SSHTransport { socket.destroy(); } + /// Force flush any buffered outgoing data to the socket. + Future flush() async { + await socket.flush(); + } + /// Subscribes to the underlying socket stream to handle incoming data and status events. void _initSocket() { _socketSubscription = socket.stream.listen( diff --git a/test/src/http/http_client_test.dart b/test/src/http/http_client_test.dart index e683dfd..8e97b5d 100644 --- a/test/src/http/http_client_test.dart +++ b/test/src/http/http_client_test.dart @@ -300,4 +300,7 @@ class _FakeSocket implements SSHSocket { } unawaited(_sinkController.close()); } + + @override + Future flush() async {} } diff --git a/test/src/ssh_auth_abort_error_test.dart b/test/src/ssh_auth_abort_error_test.dart index e400574..fd0b774 100644 --- a/test/src/ssh_auth_abort_error_test.dart +++ b/test/src/ssh_auth_abort_error_test.dart @@ -75,6 +75,9 @@ class _FakeSSHSocket implements SSHSocket { } unawaited(_inputController.close()); } + + @override + Future flush() async {} } class _RecordingSink implements StreamSink> { diff --git a/test/src/ssh_client_forward_dynamic_test.dart b/test/src/ssh_client_forward_dynamic_test.dart index e1e3b89..e973e66 100644 --- a/test/src/ssh_client_forward_dynamic_test.dart +++ b/test/src/ssh_client_forward_dynamic_test.dart @@ -64,6 +64,9 @@ class _FakeSSHSocket implements SSHSocket { } unawaited(_inputController.close()); } + + @override + Future flush() async {} } class _NoopSink implements StreamSink> { diff --git a/test/src/ssh_client_ident_test.dart b/test/src/ssh_client_ident_test.dart index ea6c57b..2b792de 100644 --- a/test/src/ssh_client_ident_test.dart +++ b/test/src/ssh_client_ident_test.dart @@ -100,6 +100,9 @@ class _FakeSSHSocket implements SSHSocket { } unawaited(_inputController.close()); } + + @override + Future flush() async {} } class _RecordingSink implements StreamSink> { diff --git a/test/src/ssh_client_run_with_result_test.dart b/test/src/ssh_client_run_with_result_test.dart index 006d092..66953fc 100644 --- a/test/src/ssh_client_run_with_result_test.dart +++ b/test/src/ssh_client_run_with_result_test.dart @@ -284,6 +284,9 @@ class _FakeSSHSocket implements SSHSocket { } unawaited(_inputController.close()); } + + @override + Future flush() async {} } class _NoopSink implements StreamSink> { diff --git a/test/src/ssh_client_timeout_test.dart b/test/src/ssh_client_timeout_test.dart index 0cf7fd3..41ba404 100644 --- a/test/src/ssh_client_timeout_test.dart +++ b/test/src/ssh_client_timeout_test.dart @@ -77,6 +77,9 @@ class _FakeSSHSocket implements SSHSocket { } unawaited(_inputController.close()); } + + @override + Future flush() async {} } class _RecordingSink implements StreamSink> { diff --git a/test/src/ssh_transport_aead_test.dart b/test/src/ssh_transport_aead_test.dart index 36b4f7f..09317f8 100644 --- a/test/src/ssh_transport_aead_test.dart +++ b/test/src/ssh_transport_aead_test.dart @@ -548,6 +548,9 @@ class _CaptureSSHSocket implements SSHSocket { } unawaited(_inputController.close()); } + + @override + Future flush() async {} } class _CaptureSink implements StreamSink> { diff --git a/test/src/ssh_transport_version_test.dart b/test/src/ssh_transport_version_test.dart index 848c18d..8627b15 100644 --- a/test/src/ssh_transport_version_test.dart +++ b/test/src/ssh_transport_version_test.dart @@ -133,6 +133,9 @@ class _FakeSSHSocket implements SSHSocket { } unawaited(_inputController.close()); } + + @override + Future flush() async {} } class _RecordingSink implements StreamSink> { From 7cac210f9bf2b83a50e2f342c22982c18ee25f68 Mon Sep 17 00:00:00 2001 From: "Victor C." <34163765+vicajilau@users.noreply.github.com> Date: Tue, 14 Jul 2026 05:36:41 +0200 Subject: [PATCH 2/5] docs(changelog): add entry for 2.22.2 with flush() changes --- CHANGELOG.md | 3 +++ 1 file changed, 3 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index 6e073a4..6c6f577 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,6 @@ +## [2.22.2] - 2026-07-14 +- Added `flush()` to `SSHSocket`, `SSHClient`, and `SSHChannel` to allow force flushing of buffered outgoing data [#183]. Thanks [@vicajilau]. + ## [2.22.1] - 2026-07-13 - Fixed a keepalive issue where overlapping pings could occur and caught errors during ping execution. Thanks [@vicajilau]. From f2e811822b0600bf68ab65f64119bfd378c96213 Mon Sep 17 00:00:00 2001 From: "Victor C." <34163765+vicajilau@users.noreply.github.com> Date: Tue, 14 Jul 2026 05:40:42 +0200 Subject: [PATCH 3/5] chore: update release date for version 2.22.0 in CHANGELOG.md --- CHANGELOG.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 40250d2..6c6f577 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,7 +4,7 @@ ## [2.22.1] - 2026-07-13 - Fixed a keepalive issue where overlapping pings could occur and caught errors during ping execution. Thanks [@vicajilau]. -## [2.22.0] - 2026-07-13 +## [2.22.0] - 2026-07-03 - Added optional `handshakeTimeout` and `authTimeout` to `SSHClient` to limit connection negotiation and user authentication times [#182]. Thanks [@GT-610]. ## [2.21.1] - 2026-07-02 From 1047270a2c773f1f529ae1ba98da8fde7cd673fd Mon Sep 17 00:00:00 2001 From: "Victor C." <34163765+vicajilau@users.noreply.github.com> Date: Tue, 14 Jul 2026 05:51:09 +0200 Subject: [PATCH 4/5] test: add unit tests for flush() to cover new API --- lib/src/ssh_session.dart | 9 +++++++++ test/src/ssh_client_test.dart | 29 +++++++++++++++++++++++++++++ 2 files changed, 38 insertions(+) diff --git a/lib/src/ssh_session.dart b/lib/src/ssh_session.dart index fe1514e..126fa1b 100644 --- a/lib/src/ssh_session.dart +++ b/lib/src/ssh_session.dart @@ -32,6 +32,9 @@ class SSHSession { /// be available on the [stdout] and [stderr] streams at this time. Future get done => _channel.done; + /// The underlying SSH channel. + SSHChannel get channel => _channel; + SSHSession(this._channel) { _channel.setRequestHandler(_handleRequest); @@ -79,6 +82,12 @@ class SSHSession { stdin.add(data); } + /// Force flush any buffered stdin data to the remote process. + Future flush() async { + await Future.microtask(() {}); + await _channel.flush(); + } + /// Inform remote process of the current window size. void resizeTerminal( int width, diff --git a/test/src/ssh_client_test.dart b/test/src/ssh_client_test.dart index 12a6f2e..2901233 100644 --- a/test/src/ssh_client_test.dart +++ b/test/src/ssh_client_test.dart @@ -4,6 +4,7 @@ library; import 'dart:convert'; import 'package:dartssh2/dartssh2.dart'; +import 'package:dartssh2/src/ssh_channel.dart'; import 'package:test/test.dart'; import '../test_utils.dart'; @@ -264,4 +265,32 @@ void main() { client.close(); }); }); + + group('SSHClient.flush', () { + test('can flush client, session, channel, and forward channel', () async { + final client = await getTestClient(); + await client.authenticated; + + await client.flush(); + + final session = await client.execute('echo flush_test'); + await session.flush(); + + final controller = SSHChannelController( + localId: 1, + localMaximumPacketSize: 1024, + localInitialWindowSize: 1024, + remoteId: 2, + remoteMaximumPacketSize: 1024, + remoteInitialWindowSize: 1024, + sendMessage: (msg) {}, + onFlush: () async {}, + ); + final forwardChannel = SSHForwardChannel(controller.channel); + await forwardChannel.flush(); + + await session.done; + client.close(); + }); + }); } From f83aae1f6ed11812268f1aa11093b91e0e2abd28 Mon Sep 17 00:00:00 2001 From: "Victor C." <34163765+vicajilau@users.noreply.github.com> Date: Tue, 14 Jul 2026 08:07:58 +0200 Subject: [PATCH 5/5] test: add unit tests for flush functionality across sockets, transport, channels, and sessions --- test/src/ssh_flush_test.dart | 180 +++++++++++++++++++++++++++++++++++ 1 file changed, 180 insertions(+) create mode 100644 test/src/ssh_flush_test.dart diff --git a/test/src/ssh_flush_test.dart b/test/src/ssh_flush_test.dart new file mode 100644 index 0000000..9c53f5b --- /dev/null +++ b/test/src/ssh_flush_test.dart @@ -0,0 +1,180 @@ +import 'dart:async'; +import 'dart:io'; +import 'dart:mirrors'; +import 'dart:typed_data'; + +import 'package:dartssh2/dartssh2.dart'; +import 'package:dartssh2/src/ssh_channel.dart'; +import 'package:test/test.dart'; + +void main() { + group('SSHSocket base class', () { + test('default flush implementation completes normally', () async { + final socket = _TestSSHSocket(); + await expectLater(socket.flush(), completes); + }); + }); + + group('SSHNativeSocket', () { + test('native socket flush works', () async { + final server = await ServerSocket.bind('127.0.0.1', 0); + final socket = await SSHSocket.connect('127.0.0.1', server.port); + await socket.flush(); + await socket.close(); + await server.close(); + }); + }); + + group('SSHTransport.flush', () { + test('delegates to socket.flush', () async { + final socket = _FakeSSHSocket(); + final transport = SSHTransport(socket); + await transport.flush(); + expect(socket.flushCount, 1); + transport.close(); + }); + }); + + group('SSHClient.flush and channel callback delegation', () { + test('delegates to transport and sets up onFlush', () async { + final socket = _FakeSSHSocket(); + final client = SSHClient(socket, username: 'demo'); + await client.flush(); + expect(socket.flushCount, 1); + + final clientLibrary = reflectClass(SSHClient).owner as LibraryMirror; + Symbol privateSymbol(String name) => + MirrorSystem.getSymbol(name, clientLibrary); + + // Invoke _acceptChannel using reflection to verify onFlush setup. + final channelController = reflect(client).invoke( + privateSymbol('_acceptChannel'), + [], + { + #localChannelId: 1, + #remoteChannelId: 2, + #remoteInitialWindowSize: 1024, + #remoteMaximumPacketSize: 1024, + }, + ).reflectee as SSHChannelController; + + expect(channelController.onFlush, isNotNull); + await channelController.flush(); + expect(socket.flushCount, 2); + + client.close(); + }); + }); + + group('SSHChannel.flush', () { + test('delegates to controller.flush', () async { + var flushed = false; + final controller = SSHChannelController( + localId: 1, + localMaximumPacketSize: 1024, + localInitialWindowSize: 1024, + remoteId: 2, + remoteMaximumPacketSize: 1024, + remoteInitialWindowSize: 1024, + sendMessage: (msg) {}, + onFlush: () async { + flushed = true; + }, + ); + final channel = controller.channel; + await channel.flush(); + expect(flushed, isTrue); + }); + }); + + group('SSHSession.flush', () { + test('exposes channel and delegates flush', () async { + var flushed = false; + final controller = SSHChannelController( + localId: 1, + localMaximumPacketSize: 1024, + localInitialWindowSize: 1024, + remoteId: 2, + remoteMaximumPacketSize: 1024, + remoteInitialWindowSize: 1024, + sendMessage: (msg) {}, + onFlush: () async { + flushed = true; + }, + ); + final channel = controller.channel; + final session = SSHSession(channel); + expect(session.channel, channel); + + await session.flush(); + expect(flushed, isTrue); + }); + }); + + group('SSHForwardChannel.flush', () { + test('delegates to channel.flush', () async { + var flushed = false; + final controller = SSHChannelController( + localId: 1, + localMaximumPacketSize: 1024, + localInitialWindowSize: 1024, + remoteId: 2, + remoteMaximumPacketSize: 1024, + remoteInitialWindowSize: 1024, + sendMessage: (msg) {}, + onFlush: () async { + flushed = true; + }, + ); + final forwardChannel = SSHForwardChannel(controller.channel); + await forwardChannel.flush(); + expect(flushed, isTrue); + }); + }); +} + +class _TestSSHSocket extends SSHSocket { + @override + Stream get stream => throw UnimplementedError(); + + @override + StreamSink> get sink => throw UnimplementedError(); + + @override + Future get done => throw UnimplementedError(); + + @override + Future close() => throw UnimplementedError(); + + @override + void destroy() => throw UnimplementedError(); +} + +class _FakeSSHSocket implements SSHSocket { + int flushCount = 0; + final _streamController = StreamController(); + final _sinkController = StreamController>(); + + @override + Stream get stream => _streamController.stream; + + @override + StreamSink> get sink => _sinkController.sink; + + @override + Future get done => _streamController.done; + + @override + Future close() async { + await _streamController.close(); + await _sinkController.close(); + } + + @override + void destroy() {} + + @override + Future flush() async { + flushCount++; + } +}