diff --git a/CHANGELOG.md b/CHANGELOG.md index 9e6a6ba..3f68c6e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,21 @@ +## 1.55.0 + +### Changed + +- **Transport now rides on [omnyhub](https://pub.dev/packages/omnyhub) 1.1.0.** + The hand-written `dart:io` WebSocket adapter (`WebSocketConnection`) is replaced + by `FrameConnection`, which wraps an omnyhub `Connection` in a + `TypedConnection`: OmnyShell's `FrameCodec` is adapted to + omnyhub's `ConnectionCodec`, mapping control↔text and data↔binary frames. + Node dials and the Hub listener both hand off `OmnyShellConnection`s produced by + omnyhub's transport, so the two packages share one connection/TLS stack. The + protocol core (frames, control messages, channels, broker, runtimes, tunnels) + and the browser transport (`WsChannelConnection`) are unchanged; the full + existing suite — including the wss/TLS-dir/e2e/TestCluster integration tests — + passes as the behavioral gate. +- Bumped `omnydrive` to `^1.12.0` (which is itself now hosted on omnyhub), so all + three packages share the same omnyhub transport/HTTP stack. + ## 1.54.0 ### Changed diff --git a/lib/src/application/node/node_runtime.dart b/lib/src/application/node/node_runtime.dart index 096f3d1..15e1a04 100644 --- a/lib/src/application/node/node_runtime.dart +++ b/lib/src/application/node/node_runtime.dart @@ -21,10 +21,11 @@ import '../../infrastructure/identity/machine_id.dart'; import '../../infrastructure/identity/uid_computer.dart'; import '../../infrastructure/backend/process_inspector.dart'; import '../../infrastructure/identity/uid_store.dart'; -import '../../infrastructure/transport/web_socket_connection.dart'; +import '../../infrastructure/transport/frame_connection.dart'; import '../../protocol/channel.dart'; import '../../protocol/channel_multiplexer.dart'; import '../../protocol/control_message.dart'; +import '../../protocol/omnyshell_connection.dart'; import '../../protocol/omnyshell_frame.dart'; import '../../version.dart'; import '../../protocol/protocol_version.dart'; @@ -182,7 +183,7 @@ class NodeRuntime { final NodeConfig config; NodeState _state = NodeState.disconnected; - WebSocketConnection? _connection; + OmnyShellConnection? _connection; ChannelMultiplexer? _mux; StreamSubscription? _controlSub; Timer? _heartbeatTimer; @@ -262,9 +263,9 @@ class NodeRuntime { if (_stopped) return; await _ensureUid(); _setState(NodeState.connecting); - final WebSocketConnection connection; + final OmnyShellConnection connection; try { - connection = await WebSocketConnection.connect( + connection = await FrameConnection.connect( config.hubUri, securityContext: config.securityContext, onBadCertificate: config.onBadCertificate, diff --git a/lib/src/infrastructure/transport/frame_connection.dart b/lib/src/infrastructure/transport/frame_connection.dart new file mode 100644 index 0000000..8dffa07 --- /dev/null +++ b/lib/src/infrastructure/transport/frame_connection.dart @@ -0,0 +1,102 @@ +import 'dart:io'; + +import 'package:omnyhub/omnyhub.dart' as omnyhub; +import 'package:web_socket_channel/web_socket_channel.dart'; + +import '../../protocol/frame_codec.dart'; +import '../../protocol/omnyshell_connection.dart'; +import '../../protocol/omnyshell_frame.dart'; + +/// Bridges OmnyShell's [FrameCodec] to omnyhub's raw [omnyhub.Message] model, so +/// OmnyShell frames ride on omnyhub's transport. +class _FrameConnectionCodec implements omnyhub.ConnectionCodec { + final FrameCodec codec; + + _FrameConnectionCodec(this.codec); + + @override + omnyhub.Message encode(OmnyShellFrame frame) { + final encoded = codec.encode(frame); // String (control) or Uint8List (data) + return encoded is String + ? omnyhub.TextMessage(encoded) + : omnyhub.BinaryMessage(encoded as List); + } + + @override + OmnyShellFrame decode(omnyhub.Message message) => switch (message) { + omnyhub.TextMessage(:final data) => codec.decode(data), + omnyhub.BinaryMessage(:final data) => codec.decode(data), + }; +} + +/// An [OmnyShellConnection] backed by an omnyhub [omnyhub.Connection] (WebSocket +/// on the VM), applying [FrameCodec] at the boundary via omnyhub's +/// [omnyhub.TypedConnection]. +/// +/// This replaces OmnyShell's hand-written `dart:io` WebSocket adapter: the +/// listener wraps each accepted omnyhub connection with [FrameConnection.fromChannel], +/// and clients dial with [FrameConnection.connect]. Undecodable frames are +/// dropped (never tearing the connection down), and text↔control / +/// binary↔data mapping is preserved. +class FrameConnection implements OmnyShellConnection { + final omnyhub.TypedConnection _typed; + + FrameConnection._(this._typed); + + /// Wraps an omnyhub [connection], applying [codec] (defaults to the standard + /// registry). + factory FrameConnection.wrap( + omnyhub.Connection connection, { + FrameCodec? codec, + }) => FrameConnection._( + omnyhub.TypedConnection( + connection, + _FrameConnectionCodec(codec ?? FrameCodec.standard()), + ), + ); + + /// Wraps an already-upgraded WebSocket [channel] (Hub server side). + factory FrameConnection.fromChannel( + WebSocketChannel channel, { + FrameCodec? codec, + }) => FrameConnection.wrap( + omnyhub.WebSocketConnection.fromChannel(channel), + codec: codec, + ); + + /// Dials [uri] (a `wss://`/`ws://` URL) and returns a connected + /// [FrameConnection] (client / node side). + static Future connect( + Uri uri, { + Map? headers, + FrameCodec? codec, + SecurityContext? securityContext, + Duration? pingInterval, + bool Function(X509Certificate cert, String host, int port)? + onBadCertificate, + }) async { + final connection = await omnyhub.WebSocketConnection.connect( + uri, + headers: headers, + securityContext: securityContext, + pingInterval: pingInterval, + onBadCertificate: onBadCertificate, + ); + return FrameConnection.wrap(connection, codec: codec); + } + + @override + Stream get incoming => _typed.incoming; + + @override + bool get isOpen => _typed.isOpen; + + @override + Future get done => _typed.done; + + @override + void send(OmnyShellFrame frame) => _typed.send(frame); + + @override + Future close([int? code, String? reason]) => _typed.close(code, reason); +} diff --git a/lib/src/infrastructure/transport/io_connection_factory.dart b/lib/src/infrastructure/transport/io_connection_factory.dart index e6710a8..6f9ec6b 100644 --- a/lib/src/infrastructure/transport/io_connection_factory.dart +++ b/lib/src/infrastructure/transport/io_connection_factory.dart @@ -1,7 +1,7 @@ import 'dart:io'; import 'connection_factory.dart'; -import 'web_socket_connection.dart'; +import 'frame_connection.dart'; /// Builds a `dart:io` [ConnectionFactory] with explicit TLS trust overrides. /// @@ -14,7 +14,7 @@ ConnectionFactory ioConnectionFactory({ bool Function(X509Certificate cert, String host, int port)? onBadCertificate, Duration? pingInterval, }) => - (Uri uri, {Map? headers}) => WebSocketConnection.connect( + (Uri uri, {Map? headers}) => FrameConnection.connect( uri, headers: headers, securityContext: securityContext, diff --git a/lib/src/infrastructure/transport/transport_factory_io.dart b/lib/src/infrastructure/transport/transport_factory_io.dart index 25dfa7f..0377e31 100644 --- a/lib/src/infrastructure/transport/transport_factory_io.dart +++ b/lib/src/infrastructure/transport/transport_factory_io.dart @@ -1,5 +1,5 @@ import '../../protocol/omnyshell_connection.dart'; -import 'web_socket_connection.dart'; +import 'frame_connection.dart'; /// The native default: dials a `wss://` WebSocket over `dart:io` with standard /// TLS verification. TLS trust overrides (self-signed test certs, pinning) are @@ -7,4 +7,4 @@ import 'web_socket_connection.dart'; Future defaultConnectionFactory( Uri uri, { Map? headers, -}) => WebSocketConnection.connect(uri, headers: headers); +}) => FrameConnection.connect(uri, headers: headers); diff --git a/lib/src/infrastructure/transport/web_socket_connection.dart b/lib/src/infrastructure/transport/web_socket_connection.dart deleted file mode 100644 index 5a8d17e..0000000 --- a/lib/src/infrastructure/transport/web_socket_connection.dart +++ /dev/null @@ -1,110 +0,0 @@ -import 'dart:async'; -import 'dart:io'; - -import 'package:web_socket_channel/io.dart'; -import 'package:web_socket_channel/web_socket_channel.dart'; - -import '../../protocol/frame_codec.dart'; -import '../../protocol/omnyshell_connection.dart'; -import '../../protocol/omnyshell_frame.dart'; - -/// An [OmnyShellConnection] over a WebSocket (carried on TLS for `wss://`). -/// -/// Wraps a [WebSocketChannel], translating raw text/binary events to and from -/// [OmnyShellFrame]s with a [FrameCodec]. Undecodable frames are dropped rather -/// than crashing the peer; protocol-level validation and fail-closed behaviour -/// live in the runtimes. -class WebSocketConnection implements OmnyShellConnection { - final WebSocketChannel _channel; - - /// The codec used to (de)serialise frames. - final FrameCodec codec; - - final StreamController _incoming = - StreamController(); - final Completer _done = Completer(); - bool _open = true; - - WebSocketConnection._(this._channel, this.codec) { - _channel.stream.listen( - (event) { - try { - _incoming.add(codec.decode(event as Object)); - } on Object { - // Drop undecodable frames; do not tear the connection down here. - } - }, - onDone: _handleClosed, - onError: (Object _) => _handleClosed(), - cancelOnError: false, - ); - } - - /// Wraps an already-upgraded WebSocket channel (Hub server side). - factory WebSocketConnection.fromChannel( - WebSocketChannel channel, { - FrameCodec? codec, - }) => WebSocketConnection._(channel, codec ?? FrameCodec.standard()); - - /// Dials [uri] (a `wss://` URL) and returns a connected - /// [WebSocketConnection]. - /// - /// [headers] are sent on the upgrade request. [securityContext] supplies the - /// trust roots for TLS; for tests against a self-signed certificate, provide - /// a context that trusts it. [onBadCertificate] is an escape hatch for - /// certificate pinning or self-signed test certificates — when `null` (the - /// default) standard verification applies and untrusted certificates are - /// rejected. The connection is established before returning. - static Future connect( - Uri uri, { - Map? headers, - FrameCodec? codec, - SecurityContext? securityContext, - Duration? pingInterval, - bool Function(X509Certificate cert, String host, int port)? - onBadCertificate, - }) async { - final httpClient = HttpClient(context: securityContext); - if (onBadCertificate != null) { - httpClient.badCertificateCallback = onBadCertificate; - } - final channel = IOWebSocketChannel.connect( - uri, - headers: headers, - pingInterval: pingInterval, - customClient: httpClient, - ); - await channel.ready; - return WebSocketConnection._(channel, codec ?? FrameCodec.standard()); - } - - @override - Stream get incoming => _incoming.stream; - - @override - bool get isOpen => _open; - - @override - Future get done => _done.future; - - @override - void send(OmnyShellFrame frame) { - if (!_open) return; - _channel.sink.add(codec.encode(frame)); - } - - @override - Future close([int? code, String? reason]) async { - if (!_open) return; - _open = false; - await _channel.sink.close(code, reason); - _handleClosed(); - } - - void _handleClosed() { - if (!_open && _done.isCompleted) return; - _open = false; - if (!_incoming.isClosed) _incoming.close(); - if (!_done.isCompleted) _done.complete(); - } -} diff --git a/lib/src/infrastructure/transport/ws_channel_connection.dart b/lib/src/infrastructure/transport/ws_channel_connection.dart index d19100c..b6da3e0 100644 --- a/lib/src/infrastructure/transport/ws_channel_connection.dart +++ b/lib/src/infrastructure/transport/ws_channel_connection.dart @@ -7,11 +7,12 @@ import '../../protocol/omnyshell_connection.dart'; import '../../protocol/omnyshell_frame.dart'; /// An [OmnyShellConnection] over any [WebSocketChannel], with **no** `dart:io` -/// dependency — the browser-safe counterpart of `WebSocketConnection`. +/// dependency — the browser-safe counterpart of the native `FrameConnection`. /// -/// The native `WebSocketConnection` carries `dart:io` at file scope (it builds -/// an `HttpClient` for TLS pinning), which poisons the dart2js build, so the -/// web transport wraps the cross-platform [WebSocketChannel] here instead. +/// The native `FrameConnection` rides on omnyhub's `dart:io` WebSocket +/// transport (which builds an `HttpClient` for TLS pinning), poisoning the +/// dart2js build, so the web transport wraps the cross-platform +/// [WebSocketChannel] here instead. /// Both translate raw text/binary events to and from [OmnyShellFrame]s with a /// [FrameCodec]; undecodable frames are dropped rather than tearing the /// connection down. diff --git a/lib/src/infrastructure/transport/ws_server_endpoint.dart b/lib/src/infrastructure/transport/ws_server_endpoint.dart index 20c8e94..5b8b61f 100644 --- a/lib/src/infrastructure/transport/ws_server_endpoint.dart +++ b/lib/src/infrastructure/transport/ws_server_endpoint.dart @@ -6,17 +6,18 @@ import 'package:shelf/shelf_io.dart' as shelf_io; import 'package:shelf_web_socket/shelf_web_socket.dart'; import '../../protocol/frame_codec.dart'; -import 'web_socket_connection.dart'; +import '../../protocol/omnyshell_connection.dart'; +import 'frame_connection.dart'; /// Called for every accepted WebSocket connection. -typedef OnConnection = void Function(WebSocketConnection connection); +typedef OnConnection = void Function(OmnyShellConnection connection); /// A Hub-side TLS WebSocket listener. /// /// Binds an HTTPS server with the provided [SecurityContext] and upgrades -/// incoming WebSocket requests, handing each accepted [WebSocketConnection] to -/// the supplied [OnConnection] callback. There is no plaintext mode: a -/// [SecurityContext] is mandatory. +/// incoming WebSocket requests, handing each accepted [OmnyShellConnection] +/// (an omnyhub-backed [FrameConnection]) to the supplied [OnConnection] +/// callback. There is no plaintext mode: a [SecurityContext] is mandatory. class WsServerEndpoint { final HttpServer _server; @@ -49,7 +50,7 @@ class WsServerEndpoint { }) async { final handler = webSocketHandler((channel, _) { onConnection( - WebSocketConnection.fromChannel( + FrameConnection.fromChannel( channel, codec: codecFactory?.call() ?? FrameCodec.standard(), ), diff --git a/lib/src/version.dart b/lib/src/version.dart index 9f4ddbf..ef249f4 100644 --- a/lib/src/version.dart +++ b/lib/src/version.dart @@ -3,4 +3,4 @@ /// This is the single source of truth for "what build is this": it is rendered /// in the CLI banner and is the default a node reports as its /// [NodeConfig.agentVersion] / [PlatformInfo.agentVersion]. -const String omnyShellVersion = '1.54.0'; +const String omnyShellVersion = '1.55.0'; diff --git a/pubspec.yaml b/pubspec.yaml index bd4c649..7f4c8f6 100644 --- a/pubspec.yaml +++ b/pubspec.yaml @@ -4,7 +4,7 @@ description: >- connect to a Hub by node identity (not host:port); the Hub authenticates, authorizes and brokers encrypted sessions to Nodes over WebSocket-on-TLS. Ships Hub, Node, Client and CLI implementations behind first-class Dart APIs. -version: 1.54.0 +version: 1.55.0 repository: https://github.com/OmnyGrid/omnyshell environment: @@ -22,7 +22,8 @@ dependencies: ffi: ^2.1.0 http: ^1.2.0 meta: ^1.18.3 - omnydrive: ^1.11.0 + omnydrive: ^1.12.0 + omnyhub: ^1.1.0 path: ^1.9.0 pointycastle: ^4.0.0 #portable_pty: ^0.0.5 # waiting crash fix. diff --git a/test/integration/frame_connection_wire_test.dart b/test/integration/frame_connection_wire_test.dart new file mode 100644 index 0000000..73ba274 --- /dev/null +++ b/test/integration/frame_connection_wire_test.dart @@ -0,0 +1,165 @@ +import 'dart:async'; +import 'dart:typed_data'; + +import 'package:omnyshell/omnyshell.dart'; +import 'package:omnyshell/src/infrastructure/transport/frame_connection.dart'; +import 'package:omnyshell/src/infrastructure/transport/ws_server_endpoint.dart'; +import 'package:test/test.dart'; + +import '../support/harness.dart'; + +/// End-to-end wire-fidelity for the omnyhub-backed [FrameConnection] over a real +/// `wss` socket. +/// +/// The unit test drives a fake connection, so it never proves that control +/// frames actually cross the wire as WebSocket **text** frames and data frames +/// as **binary** frames — the one thing that would silently corrupt every +/// session if omnyhub's transport got it wrong. This binds a real TLS +/// [WsServerEndpoint], dials it with [FrameConnection.connect], and asserts the +/// frames survive the round trip byte-for-byte in both directions. +void main() { + late WsServerEndpoint server; + late FrameConnection client; + late OmnyShellConnection accepted; + + setUp(() async { + final acceptedC = Completer(); + server = await WsServerEndpoint.bind( + host: '127.0.0.1', + port: 0, + securityContext: hubSecurityContext(), + onConnection: (c) { + if (!acceptedC.isCompleted) acceptedC.complete(c); + }, + ); + client = await FrameConnection.connect( + Uri.parse('wss://127.0.0.1:${server.port}'), + securityContext: trustContext(), + onBadCertificate: (cert, host, port) => true, + ); + accepted = await acceptedC.future; + }); + + tearDown(() async { + await client.close(); + await server.close(force: true); + }); + + test('a control frame crosses the wire as a text frame', () async { + final received = accepted.incoming.first; + client.send( + const ControlFrame( + Hello(role: 'node', protocolVersion: 1, minVersion: 1, nonce: 'wire'), + ), + ); + + final frame = await received.timeout(const Duration(seconds: 5)); + expect(frame, isA()); + expect(((frame as ControlFrame).message as Hello).nonce, 'wire'); + }); + + test('a data frame with non-UTF-8 bytes survives byte-for-byte', () async { + // Bytes that are NOT valid UTF-8: if omnyhub ever sent this as a text frame + // it would be mangled by UTF-8 transcoding. This is the core regression. + final payload = Uint8List.fromList([0xFF, 0x00, 0xFE, 0x80, 0x01, 0xC0]); + final received = accepted.incoming.first; + client.send( + DataFrame(opcode: DataOpcode.stdout, channel: 5, payload: payload), + ); + + final frame = await received.timeout(const Duration(seconds: 5)); + expect(frame, isA()); + final data = frame as DataFrame; + expect(data.opcode, DataOpcode.stdout); + expect(data.channel, 5); + expect(data.payload, orderedEquals(payload)); + }); + + test('the server can push a data frame back to the client', () async { + final payload = Uint8List.fromList([1, 2, 3, 4, 5]); + final received = client.incoming.first; + accepted.send( + DataFrame(opcode: DataOpcode.stderr, channel: 9, payload: payload), + ); + + final frame = await received.timeout(const Duration(seconds: 5)); + expect(frame, isA()); + expect((frame as DataFrame).opcode, DataOpcode.stderr); + expect(frame.payload, orderedEquals(payload)); + }); + + test('a large binary payload round-trips intact', () async { + // Just under the protocol's 64 KiB per-frame cap: a real multi-KB binary + // frame across the socket, covering every byte value. + final payload = Uint8List(60 * 1024); + for (var i = 0; i < payload.length; i++) { + payload[i] = (i * 31 + 7) & 0xFF; + } + final received = accepted.incoming.first; + client.send( + DataFrame(opcode: DataOpcode.stdout, channel: 1, payload: payload), + ); + + final frame = await received.timeout(const Duration(seconds: 10)); + expect((frame as DataFrame).payload, orderedEquals(payload)); + }); + + test( + 'control and data frames interleave without cross-contamination', + () async { + final frames = []; + final sub = accepted.incoming.listen(frames.add); + + client.send( + const ControlFrame( + Hello(role: 'node', protocolVersion: 1, minVersion: 1, nonce: 'a'), + ), + ); + client.send( + DataFrame( + opcode: DataOpcode.stdout, + channel: 2, + payload: Uint8List.fromList([9, 9, 9]), + ), + ); + client.send( + const ControlFrame( + Hello(role: 'node', protocolVersion: 1, minVersion: 1, nonce: 'b'), + ), + ); + + await _until(() => frames.length >= 3); + await sub.cancel(); + + expect(frames[0], isA()); + expect(((frames[0] as ControlFrame).message as Hello).nonce, 'a'); + expect(frames[1], isA()); + expect((frames[1] as DataFrame).payload, orderedEquals([9, 9, 9])); + expect(frames[2], isA()); + expect(((frames[2] as ControlFrame).message as Hello).nonce, 'b'); + }, + ); + + test('a custom WebSocket close code propagates to the peer', () async { + // omnyshell closes with app-specific codes (e.g. 4401 unauthorized). Verify + // omnyhub's transport forwards them: the server side observes `done`. + final serverDone = accepted.done; + await client.close(4401, 'unauthorized'); + await serverDone.timeout(const Duration(seconds: 5)); + expect(accepted.isOpen, isFalse); + }); +} + +/// Polls [condition] until it holds, yielding to the event loop between checks. +Future _until( + bool Function() condition, { + Duration timeout = const Duration(seconds: 5), +}) async { + final deadline = DateTime.now().add(timeout); + while (!condition()) { + if (DateTime.now().isAfter(deadline)) { + throw StateError('condition not met within $timeout'); + } + await Future.delayed(const Duration(milliseconds: 5)); + } +} diff --git a/test/unit/transport/frame_connection_test.dart b/test/unit/transport/frame_connection_test.dart new file mode 100644 index 0000000..8e7f7b1 --- /dev/null +++ b/test/unit/transport/frame_connection_test.dart @@ -0,0 +1,159 @@ +import 'dart:async'; +import 'dart:convert'; +import 'dart:typed_data'; + +import 'package:omnyhub/omnyhub.dart' as omnyhub; +import 'package:omnyshell/omnyshell.dart'; +import 'package:omnyshell/src/infrastructure/transport/frame_connection.dart'; +import 'package:test/test.dart'; + +/// An in-memory omnyhub [omnyhub.Connection] the test drives directly: whatever +/// is [send]-ed is captured in [sent], and [deliver] pushes an inbound +/// [omnyhub.Message] as if it arrived from the peer. +class _FakeConnection implements omnyhub.Connection { + final _incoming = StreamController(); + final _done = Completer(); + final List sent = []; + bool _open = true; + + void deliver(omnyhub.Message message) => _incoming.add(message); + + @override + Stream get incoming => _incoming.stream; + + @override + bool get isOpen => _open; + + @override + Future get done => _done.future; + + @override + String? get remoteAddress => '203.0.113.9'; + + @override + void send(omnyhub.Message message) => sent.add(message); + + @override + Future close([int? code, String? reason]) async { + _open = false; + // Don't await: closing a single-subscription controller that was never + // listened returns a future that only completes once drained. + unawaited(_incoming.close()); + if (!_done.isCompleted) _done.complete(); + } +} + +void main() { + group('FrameConnection', () { + late _FakeConnection raw; + late FrameConnection conn; + + setUp(() { + raw = _FakeConnection(); + conn = FrameConnection.wrap(raw); + }); + + test('encodes a control frame as an omnyhub text message', () { + conn.send( + const ControlFrame( + Hello(role: 'hub', protocolVersion: 1, minVersion: 1, nonce: 'abc'), + ), + ); + + expect(raw.sent, hasLength(1)); + final message = raw.sent.single; + expect(message, isA()); + expect(jsonDecode((message as omnyhub.TextMessage).data)['t'], 'hello'); + }); + + test('encodes a data frame as an omnyhub binary message', () { + conn.send( + DataFrame( + opcode: DataOpcode.stdout, + channel: 3, + payload: Uint8List.fromList([1, 2, 3]), + ), + ); + + expect(raw.sent.single, isA()); + }); + + test('decodes an inbound text message into a ControlFrame', () async { + final frames = []; + conn.incoming.listen(frames.add); + + final encoded = FrameCodec.standard().encodeControl( + const Hello( + role: 'node', + protocolVersion: 1, + minVersion: 1, + nonce: 'z', + ), + ); + raw.deliver(omnyhub.TextMessage(encoded)); + await pumpEventQueue(); + + expect(frames, hasLength(1)); + final frame = frames.single as ControlFrame; + expect((frame.message as Hello).nonce, 'z'); + }); + + test('round-trips a data frame through encode + decode', () async { + final frames = []; + conn.incoming.listen(frames.add); + + final codec = FrameCodec.standard(); + final encoded = codec.encode( + DataFrame( + opcode: DataOpcode.stderr, + channel: 9, + payload: Uint8List.fromList([9, 8, 7]), + ), + ); + raw.deliver(omnyhub.BinaryMessage(encoded as List)); + await pumpEventQueue(); + + final frame = frames.single as DataFrame; + expect(frame.opcode, DataOpcode.stderr); + expect(frame.channel, 9); + expect(frame.payload, [9, 8, 7]); + }); + + test('drops an undecodable inbound frame without tearing down', () async { + final frames = []; + final errors = []; + conn.incoming.listen(frames.add, onError: errors.add); + + raw.deliver(const omnyhub.TextMessage('not valid json {{{')); + // A valid frame after the bad one still arrives. + raw.deliver( + omnyhub.TextMessage( + FrameCodec.standard().encodeControl( + const Hello( + role: 'node', + protocolVersion: 1, + minVersion: 1, + nonce: 'after-bad', + ), + ), + ), + ); + await pumpEventQueue(); + + expect(errors, isEmpty); + expect(frames, hasLength(1)); + expect((frames.single as ControlFrame).message, isA()); + expect( + ((frames.single as ControlFrame).message as Hello).nonce, + 'after-bad', + ); + }); + + test('close() closes the underlying connection', () async { + expect(conn.isOpen, isTrue); + await conn.close(); + expect(raw.isOpen, isFalse); + await conn.done; + }); + }); +}