Update

lib/src/client/centrifuge.dart

@@ -114,6 +114,8 @@ base mixin CentrifugeStateMixin on CentrifugeBase {
 
   @protected
   @nonVirtual
+  @pragma('vm:prefer-inline')
+  @pragma('dart2js:tryInline')
   void _onStateChange(CentrifugeState newState) {
     logger.info('State changed: ${_state.type} -> ${state.type}');
     switch (newState) {

lib/src/client/disconnect_code.dart

@@ -24,7 +24,10 @@ enum DisconnectCode {
   badProtocol(2, 'bad protocol'),
 
   /// Client message write error
-  messageSizeLimit(3, 'message size limit exceeded');
+  messageSizeLimit(3, 'message size limit exceeded'),
+
+  /// Timeout
+  timeout(4, 'timeout exceeded');
 
   /// {@nodoc}
   const DisconnectCode(this.code, this.reason);

lib/src/subscription/client_subscription_impl.dart

@@ -188,9 +188,11 @@ base mixin CentrifugeClientSubscriptionSubscribeMixin
         ttl: subscribed.ttl,
       ));
     } on CentrifugeException catch (error, stackTrace) {
+      unsubscribe(0, 'error while subscribing').ignore();
       _emitError(error, stackTrace);
       rethrow;
     } on Object catch (error, stackTrace) {
+      unsubscribe(0, 'error while subscribing').ignore();
       final centrifugeException = CentrifugeSubscriptionException(
         message: 'Error while subscribing',
         channel: channel,

lib/src/subscription/client_subscription_manager.dart

@@ -94,36 +94,30 @@ final class ClientSubscriptionManager {
   /// {@nodoc}
   void subscribeAll() {
     for (final entry in _channelSubscriptions.values) {
-      try {
-        // TODO(plugfox): moveToSubscribing if subscribed now
-      } on Object {
-        /* ignore */
-      }
+      entry.subscribe().ignore();
     }
   }
 
   /// Disconnect all subscriptions for the specific client
   /// from internal registry.
   /// {@nodoc}
-  void unsubscribeAll() {
+  void unsubscribeAll([
+    int code = 0,
+    String reason = 'connection closed',
+  ]) {
     for (final entry in _channelSubscriptions.values) {
-      try {
-        // TODO(plugfox): moveToSubscribing if subscribed now
-      } on Object {
-        /* ignore */
-      }
+      entry.unsubscribe(code, reason).ignore();
     }
   }
 
   /// Remove all subscriptions for the specific client from internal registry.
   /// {@nodoc}
-  void removeAll() {
+  void removeAll([
+    int code = 0,
+    String reason = 'client closed',
+  ]) {
     for (final entry in _channelSubscriptions.values) {
-      try {
-        // TODO(plugfox): moveToSubscribing if subscribed now
-      } on Object {
-        /* ignore */
-      }
+      entry.unsubscribe(code, reason).ignore();
     }
     _channelSubscriptions.clear();
   }

lib/src/transport/ws_protobuf_transport.dart

@@ -378,9 +378,7 @@ base mixin CentrifugeWSPBStateHandlerMixin
   /// {@nodoc}
   @protected
   @nonVirtual
-  void _setState(CentrifugeState state) {
-    states.notify(state);
-  }
+  void _setState(CentrifugeState state) => states.notify(state);
 
   @protected
   @nonVirtual
@@ -556,6 +554,12 @@ base mixin CentrifugeWSPBSubscription
     try {
       result = await _sendMessage(request, pb.SubscribeResult())
           .timeout(config.timeout);
+    } on TimeoutException {
+      disconnect(
+        DisconnectCode.timeout.code,
+        'Timeout while subscribing to channel $channel',
+      ).ignore();
+      rethrow;
     } on Object catch (error, stackTrace) {
       Error.throwWithStackTrace(
         CentrifugeSubscriptionException(