Unsubscribe

lib/src/client/disconnect_code.dart

@@ -27,7 +27,10 @@ enum DisconnectCode {
   messageSizeLimit(3, 'message size limit exceeded'),
 
   /// Timeout
-  timeout(4, 'timeout exceeded');
+  timeout(4, 'timeout exceeded'),
+
+  /// Unsubscribe error
+  unsubscribeError(5, 'unsubscribe error');
 
   /// {@nodoc}
   const DisconnectCode(this.code, this.reason);

lib/src/subscription/client_subscription_impl.dart

@@ -1,6 +1,7 @@
 import 'dart:async';
 
 import 'package:centrifuge_dart/centrifuge.dart';
+import 'package:centrifuge_dart/src/client/disconnect_code.dart';
 import 'package:centrifuge_dart/src/subscription/subscription_states_stream.dart';
 import 'package:centrifuge_dart/src/transport/transport_interface.dart';
 import 'package:centrifuge_dart/src/util/event_queue.dart';
@@ -244,7 +245,23 @@ base mixin CentrifugeClientSubscriptionSubscribeMixin
       reason: reason,
       since: state.since,
     ));
-    // TODO(plugfox): implement
+    try {
+      await _transport.unsubscribe(channel, _config);
+    } on Object catch (error, stackTrace) {
+      final centrifugeException = CentrifugeSubscriptionException(
+        message: 'Error while unsubscribing',
+        channel: channel,
+        error: error,
+      );
+      _emitError(centrifugeException, stackTrace);
+      _transport
+          .disconnect(
+            DisconnectCode.unsubscribeError.code,
+            DisconnectCode.unsubscribeError.reason,
+          )
+          .ignore();
+      Error.throwWithStackTrace(centrifugeException, stackTrace);
+    }
   }
 
   @override

lib/src/transport/transport_interface.dart

@@ -23,15 +23,24 @@ abstract interface class ICentrifugeTransport {
   /// Send asynchronous message to a server. This method makes sense
   /// only when using Centrifuge library for Go on a server side. In Centrifuge
   /// asynchronous message handler does not exist.
+  /// {@nodoc}
   Future<void> sendAsyncMessage(List<int> data);
 
   /// Subscribe on channel with optional [since] position.
+  /// {@nodoc}
   Future<SubcibedOnChannel> subscribe(
     String channel,
     CentrifugeSubscriptionConfig config,
     CentrifugeStreamPosition? since,
   );
 
+  /// Unsubscribe from channel.
+  /// {@nodoc}
+  Future<void> unsubscribe(
+    String channel,
+    CentrifugeSubscriptionConfig config,
+  );
+
   /// Disconnect from the server.
   /// e.g. code: 0, reason: 'disconnect called'
   /// {@nodoc}

lib/src/transport/ws_protobuf_transport.dart

@@ -603,6 +603,15 @@ base mixin CentrifugeWSPBSubscription
       data: result.hasData() ? result.data : null,
     );
   }
+
+  @override
+  Future<void> unsubscribe(
+    String channel,
+    CentrifugeSubscriptionConfig config,
+  ) async {
+    final request = pb.UnsubscribeRequest()..channel = channel;
+    await _sendMessage(request, pb.UnsubscribeResult()).timeout(config.timeout);
+  }
 }
 
 /// To maintain connection alive and detect broken connections