route DaemonClient through a DaemonTransport seam (T-331)
The UI's backend client connected straight to the workspace unix socket, hard-coding the local shape. It now talks JSON-lines through a DaemonTransport (new lib/src/ipc/transport.dart, Flutter-free), with LocalSocketTransport reproducing today's connect byte-for-byte — zero behavior change, proven by the untouched client test suite plus new seam tests driving the client over an in-memory transport. This is the slot the SSH-remote backend (T-329/Q-23) plugs into: request correlation, reconnect/backoff, and event forwarding live above the seam and won't change when the endpoint is remote. main.dart's swapIpcServer becomes swapBackend per the same plan. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -56,13 +56,50 @@ Future<String> _tmpSocket() async {
|
||||
}
|
||||
|
||||
DaemonClient _build(String socketPath, DaemonBus bus) {
|
||||
return DaemonClient(
|
||||
return DaemonClient.unixSocket(
|
||||
socketPath: socketPath,
|
||||
log: Logger(minLevel: LogLevel.error, sinks: const []),
|
||||
events: bus,
|
||||
);
|
||||
}
|
||||
|
||||
/// In-memory transport (T-331): proves DaemonClient runs unmodified over
|
||||
/// any [DaemonTransport], not just the unix socket — the seam the remote
|
||||
/// backend (T-329) slots into.
|
||||
class _MemoryTransport implements DaemonTransport {
|
||||
final toClient = StreamController<String>.broadcast();
|
||||
final fromClient = StreamController<String>.broadcast();
|
||||
int opens = 0;
|
||||
|
||||
@override
|
||||
String get endpoint => 'memory://test';
|
||||
|
||||
@override
|
||||
Future<DaemonConnection> open() async {
|
||||
opens++;
|
||||
return _MemoryConnection(this);
|
||||
}
|
||||
|
||||
Future<void> close() async {
|
||||
await toClient.close();
|
||||
await fromClient.close();
|
||||
}
|
||||
}
|
||||
|
||||
class _MemoryConnection implements DaemonConnection {
|
||||
_MemoryConnection(this._t);
|
||||
final _MemoryTransport _t;
|
||||
|
||||
@override
|
||||
Stream<String> get lines => _t.toClient.stream;
|
||||
|
||||
@override
|
||||
void writeLine(String line) => _t.fromClient.add(line);
|
||||
|
||||
@override
|
||||
Future<void> close() async {}
|
||||
}
|
||||
|
||||
void main() {
|
||||
group('DaemonClient — happy path', () {
|
||||
test('connect → request → matching response completes', () async {
|
||||
@@ -341,4 +378,63 @@ void main() {
|
||||
expect(resp.ok, isTrue);
|
||||
});
|
||||
});
|
||||
|
||||
group('DaemonClient — transport seam (T-331)', () {
|
||||
test('request/response round-trips over a non-socket transport', () async {
|
||||
final transport = _MemoryTransport();
|
||||
addTearDown(transport.close);
|
||||
final bus = DaemonBus();
|
||||
addTearDown(bus.dispose);
|
||||
final client = DaemonClient(
|
||||
transport: transport,
|
||||
log: Logger(minLevel: LogLevel.error, sinks: const []),
|
||||
events: bus,
|
||||
);
|
||||
addTearDown(client.dispose);
|
||||
|
||||
await client.start();
|
||||
expect(transport.opens, 1);
|
||||
expect(client.isConnected, isTrue);
|
||||
expect(client.socketPath, 'memory://test');
|
||||
|
||||
final lineFuture = transport.fromClient.stream.first;
|
||||
final respFuture = client.request('ping', args: {'n': 1});
|
||||
final line = await lineFuture;
|
||||
final req = IpcMessage.decode(line) as IpcRequest;
|
||||
expect(req.cmd, 'ping');
|
||||
transport.toClient.add(IpcResponse.ok(id: req.id, data: const {'pong': true}).encode());
|
||||
final resp = await respFuture.timeout(const Duration(seconds: 2));
|
||||
expect(resp.ok, isTrue);
|
||||
expect(resp.data['pong'], isTrue);
|
||||
});
|
||||
|
||||
test('reconnectWith swaps from a socket transport to another transport', () async {
|
||||
final path = await _tmpSocket();
|
||||
final daemon = _TestDaemon(path);
|
||||
await daemon.start();
|
||||
addTearDown(daemon.close);
|
||||
|
||||
final bus = DaemonBus();
|
||||
addTearDown(bus.dispose);
|
||||
final client = _build(path, bus);
|
||||
addTearDown(client.dispose);
|
||||
await client.start();
|
||||
await daemon.waitForClient();
|
||||
await Future<void>.delayed(const Duration(milliseconds: 50));
|
||||
expect(client.isConnected, isTrue);
|
||||
|
||||
final transport = _MemoryTransport();
|
||||
addTearDown(transport.close);
|
||||
await client.reconnectWith(transport);
|
||||
expect(client.isConnected, isTrue);
|
||||
expect(client.socketPath, 'memory://test');
|
||||
|
||||
// Requests now flow over the new transport, not the old socket.
|
||||
final lineFuture = transport.fromClient.stream.first;
|
||||
final respFuture = client.request('over-memory');
|
||||
final req = IpcMessage.decode(await lineFuture) as IpcRequest;
|
||||
transport.toClient.add(IpcResponse.ok(id: req.id, data: const {}).encode());
|
||||
expect((await respFuture).ok, isTrue);
|
||||
});
|
||||
});
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user