diff --git a/lib/client/client.dart b/lib/client/client.dart
index 7f835fa..31ffea0 100644
--- a/lib/client/client.dart
+++ b/lib/client/client.dart
@@ -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);
 
@@ -59,6 +64,8 @@ class ButtplugClient {
         }
         _communicator!.eventStreamController.add(DeviceListReceivedEvent());
       }
+    }, onDone: () {
+      _handleDisconnect();
     });
 
     await _communicator!.connect();
@@ -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 {
diff --git a/lib/connectors/websocket_connector.dart b/lib/connectors/websocket_connector.dart
index d58a50c..551ba69 100644
--- a/lib/connectors/websocket_connector.dart
+++ b/lib/connectors/websocket_connector.dart
@@ -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);
   }
diff --git a/test/buttplug_test.dart b/test/buttplug_test.dart
index e69c420..1d60e77 100644
--- a/test/buttplug_test.dart
+++ b/test/buttplug_test.dart
@@ -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() {
@@ -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();
+  }
 }