Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 19 additions & 2 deletions lib/client/client.dart
Original file line number Diff line number Diff line change
Expand Up @@ -25,11 +25,16 @@ class DeviceListReceivedEvent extends ButtplugClientEvent {
DeviceListReceivedEvent();
}

class DisconnectEvent extends ButtplugClientEvent {
DisconnectEvent();
}

class ButtplugClient {
final String name;
String? _serverName;
ButtplugClientCommunicator? _communicator;
final Map<int, ButtplugClientDevice> _devices = {};
bool _isConnected = false;

ButtplugClient(this.name);

Expand Down Expand Up @@ -59,6 +64,8 @@ class ButtplugClient {
}
_communicator!.eventStreamController.add(DeviceListReceivedEvent());
}
}, onDone: () {
_handleDisconnect();
});

await _communicator!.connect();
Expand All @@ -84,14 +91,24 @@ class ButtplugClient {
for (var device in deviceList.devices.values) {
_devices[device.deviceIndex] = ButtplugClientDevice(device, _communicator!);
}
_isConnected = true;
}

bool connected() {
return true;
return _isConnected;
}

void _handleDisconnect() {
if (!_isConnected) return;
_isConnected = false;
_devices.clear();
_communicator?.eventStreamController.add(DisconnectEvent());
}

Future<void> disconnect() async {
await _communicator!.disconnect();
if (!_isConnected) return;
await _communicator?.disconnect();
_handleDisconnect();
}

Future<void> startScanning() async {
Expand Down
47 changes: 33 additions & 14 deletions lib/connectors/websocket_connector.dart
Original file line number Diff line number Diff line change
Expand Up @@ -14,31 +14,50 @@ class ButtplugWebsocketClientConnector implements ButtplugClientConnector {

@override
Future<void> connect() async {
_wsChannel = WebSocketChannel.connect(Uri.parse('ws://127.0.0.1:12345'));
_wsChannel = WebSocketChannel.connect(Uri.parse(address));
await _wsChannel?.ready;
_wsChannel!.stream.forEach((element) async {
try {
logInfo(element);
List<dynamic> msgs = jsonDecode(element);
for (var msg in msgs) {
_serverMessageStream.add(ButtplugServerMessage.fromJson(msg));
_wsChannel!.stream.listen(
(element) {
try {
logInfo(element);
List<dynamic> msgs = jsonDecode(element);
for (var msg in msgs) {
_serverMessageStream.add(ButtplugServerMessage.fromJson(msg));
}
} catch (e, s) {
logError("Error adding message to stream: $e");
logError(s);
disconnect();
}
} catch (e, s) {
logError("Error adding message to stream: $e");
logError(s);
await disconnect();
}
});
},
onError: (error) {
logError("WebSocket error: $error");
disconnect();
},
onDone: () {
logInfo("WebSocket connection closed");
disconnect();
},
cancelOnError: true,
);
}

@override
Future<void> disconnect() async {
_wsChannel!.sink.close();
if (_wsChannel == null) return;
var ws = _wsChannel;
_wsChannel = null;
try {
await ws?.sink.close();
} catch (_) {}
if (!_serverMessageStream.isClosed) {
await _serverMessageStream.close();
}
}

@override
void send(List<ButtplugClientMessageUnion> messages) {
if (_wsChannel == null) return;
String msg = jsonEncode(messages);
_wsChannel!.sink.add(msg);
}
Expand Down
94 changes: 93 additions & 1 deletion test/buttplug_test.dart
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
import 'dart:async';
import 'dart:convert';

import 'package:buttplug/buttplug.dart';
import 'package:buttplug/messages/messages.dart';
import 'package:test/test.dart';

void main() {
Expand Down Expand Up @@ -41,4 +41,96 @@ void main() {
});
*/
});

group('Client Disconnection and Events', () {
test('Connector stores the correct address', () {
final connector = ButtplugWebsocketClientConnector('ws://127.0.0.1:54321');
expect(connector.address, equals('ws://127.0.0.1:54321'));
});

test('Client triggers DisconnectEvent on stream close', () async {
final client = ButtplugClient('Test Client');
final connector = MockConnector();

// Start connection in background
final connectFuture = client.connect(connector);

// Wait a microtask for the client to send RequestServerInfo
await Future.delayed(Duration(milliseconds: 10));

// Simulate handshake server messages:
// 1. ServerInfo (matching the request ID)
final serverInfo = ButtplugServerMessage()
..serverInfo = (ServerInfo()
..id = connector.lastMessageId
..serverName = 'Test Server'
..protocolVersionMajor = 4
..protocolVersionMinor = 0);
connector.simulateServerMessage(serverInfo);

// Wait a microtask for the client to send RequestDeviceList
await Future.delayed(Duration(milliseconds: 10));

// 2. DeviceList (matching the request ID)
final deviceList = ButtplugServerMessage()
..deviceList = (DeviceList()
..id = connector.lastMessageId
..devices = {});
connector.simulateServerMessage(deviceList);

await connectFuture;

expect(client.connected(), isTrue);

// Listen for DisconnectEvent
ButtplugClientEvent? receivedEvent;
client.eventStream.listen((event) {
receivedEvent = event;
});

// Simulate connection drop
connector.simulateDisconnect();

// Wait a microtask to let the stream events propagate
await Future.delayed(Duration(milliseconds: 10));

expect(client.connected(), isFalse);
expect(receivedEvent, isA<DisconnectEvent>());
});
});
}

class MockConnector implements ButtplugClientConnector {
final StreamController<ButtplugServerMessage> _messageStreamController = StreamController.broadcast();
int lastMessageId = 0;
bool isConnected = false;

@override
Future<void> connect() async {
isConnected = true;
}

@override
Future<void> disconnect() async {
isConnected = false;
await _messageStreamController.close();
}

@override
void send(List<ButtplugClientMessageUnion> messages) {
if (messages.isNotEmpty) {
lastMessageId = messages[0].id;
}
}

@override
Stream<ButtplugServerMessage> get messageStream => _messageStreamController.stream;

void simulateServerMessage(ButtplugServerMessage message) {
_messageStreamController.add(message);
}

void simulateDisconnect() {
_messageStreamController.close();
}
}