State management improvements (#9)

* version is empty by default
* avoid base64
* properly handle temporary errors
* handle errors during unsubscribe
* bump to 0.0.7

Author: FZambia

src/Centrifugal.Centrifuge/Centrifugal.Centrifuge.csproj

@@ -15,7 +15,7 @@
 
     <!-- Package Information -->
     <PackageId>Centrifugal.Centrifuge</PackageId>
-    <Version>0.0.6</Version>
+    <Version>0.0.7</Version>
     <Authors>Centrifugal Labs</Authors>
     <Company>Centrifugal Labs</Company>
     <Description>

src/Centrifugal.Centrifuge/Client.cs

@@ -110,7 +110,7 @@ public class CentrifugeClient : IDisposable, IAsyncDisposable
         public event EventHandler<CentrifugeDisconnectedEventArgs>? Disconnected;
 
         /// <summary>
-        /// Event raised when an error occurs.
+        /// Event raised when an error occurs. Mostly for the logging purposes.
         /// </summary>
         public event EventHandler<CentrifugeErrorEventArgs>? Error;
 
@@ -814,7 +814,16 @@ private async Task StartConnectingAsync(int code, string reason)
 
         private async Task CreateTransportAsync()
         {
-            _transport?.Dispose();
+            // Unsubscribe from old transport events before disposing to prevent race conditions
+            if (_transport != null)
+            {
+                _transport.Opened -= OnTransportOpened;
+                _transport.MessageReceived -= OnTransportMessage;
+                _transport.Closed -= OnTransportClosed;
+                _transport.Error -= OnTransportError;
+                _transport.Dispose();
+                _transport = null;
+            }
 
             ITransport transport;
 
@@ -1489,7 +1498,7 @@ private async Task HandleTransportClosedAsync(TransportClosedEventArgs e)
             string reason = e.Reason;
             _logger?.LogDebug($"HandleTransportClosedAsync - processing with code: {code}, reason: '{reason}'");
 
-            // Check for non-reconnectable disconnect codes (matching centrifuge-js behavior)
+            // Check for non-reconnectable disconnect codes
             // Codes 3500-3999 and 4500-4999 mean permanent disconnect
             // Also BadProtocol, Unauthorized, and MessageSizeLimit are permanent
             if (e.Code.HasValue)
@@ -1537,7 +1546,6 @@ private async Task HandleTransportClosedAsync(TransportClosedEventArgs e)
             SetState(CentrifugeClientState.Connecting);
 
             // Move all subscribed subscriptions to subscribing state BEFORE emitting connecting event
-            // (matching centrifuge-js behavior - see _disconnect method)
             foreach (var sub in _subscriptions.Values)
             {
                 sub.MoveToSubscribing(CentrifugeSubscribingCodes.TransportClosed, "transport closed");
@@ -1564,7 +1572,7 @@ private void OnTransportError(object? sender, Exception e)
 
         internal async Task HandleSubscribeTimeoutAsync()
         {
-            // Subscribe timeout triggers client disconnect with reconnect, matching centrifuge-js behavior
+            // Subscribe timeout triggers client disconnect with reconnect
             if (_state == CentrifugeClientState.Disconnected)
             {
                 return;
@@ -1578,6 +1586,22 @@ internal async Task HandleSubscribeTimeoutAsync()
             await ScheduleReconnectAsync().ConfigureAwait(false);
         }
 
+        internal async Task HandleUnsubscribeErrorAsync()
+        {
+            // Unsubscribe error triggers client disconnect with reconnect (matching centrifuge-js behavior)
+            if (_state == CentrifugeClientState.Disconnected)
+            {
+                return;
+            }
+
+            await StartConnectingAsync(CentrifugeConnectingCodes.UnsubscribeError, "unsubscribe error").ConfigureAwait(false);
+
+            // Properly clean up transport before reconnecting
+            await CleanupTransportAsync().ConfigureAwait(false);
+
+            await ScheduleReconnectAsync().ConfigureAwait(false);
+        }
+
         /// <summary>
         /// Cleans up the current transport (unsubscribes events, closes, disposes).
         /// This should be called before reconnecting to prevent duplicate connections.

src/Centrifugal.Centrifuge/ClientOptions.cs

@@ -11,7 +11,8 @@ namespace Centrifugal.Centrifuge
     public class CentrifugeClientOptions
     {
         /// <summary>
-        /// Gets or sets the initial connection token (JWT).
+        /// Gets or sets the initial connection token (JWT). For proper JWT expiration handling
+        /// GetToken must be also set.
         /// </summary>
         public string? Token { get; set; }
 
@@ -35,9 +36,9 @@ public class CentrifugeClientOptions
 
         /// <summary>
         /// Gets or sets the client version.
-        /// Default is the assembly version.
+        /// By default not set.
         /// </summary>
-        public string Version { get; set; } = typeof(CentrifugeClient).Assembly.GetName().Version?.ToString() ?? "1.0.0";
+        public string Version { get; set; } = "";
 
         /// <summary>
         /// Gets or sets the minimum delay between reconnect attempts.

src/Centrifugal.Centrifuge/Subscription.cs

@@ -488,6 +488,24 @@ internal async Task SendSubscribeIfNeededAsync()
             {
                 await SetUnsubscribedAsync(CentrifugeUnsubscribedCodes.Unauthorized, "unauthorized").ConfigureAwait(false);
             }
+            catch (CentrifugeException ex)
+            {
+                OnError("subscribe", ex);
+                // Temporary errors or token expired (109) - schedule resubscribe
+                // Permanent errors - unsubscribe (matching centrifuge-js behavior)
+                if (ex.Code < 100 || ex.Code == 109 || ex.Temporary)
+                {
+                    if (ex.Code == 109)
+                    {
+                        _refreshRequired = true;
+                    }
+                    await ScheduleResubscribeAsync().ConfigureAwait(false);
+                }
+                else
+                {
+                    await SetUnsubscribedAsync(ex.Code, ex.Message).ConfigureAwait(false);
+                }
+            }
             catch (Exception ex)
             {
                 OnError("subscribe", ex);
@@ -528,6 +546,24 @@ private async Task StartSubscribingAsync(int code, string reason)
             {
                 await SetUnsubscribedAsync(CentrifugeUnsubscribedCodes.Unauthorized, "unauthorized").ConfigureAwait(false);
             }
+            catch (CentrifugeException ex)
+            {
+                OnError("subscribe", ex);
+                // Temporary errors or token expired (109) - schedule resubscribe
+                // Permanent errors - unsubscribe (matching centrifuge-js behavior)
+                if (ex.Code < 100 || ex.Code == 109 || ex.Temporary)
+                {
+                    if (ex.Code == 109)
+                    {
+                        _refreshRequired = true;
+                    }
+                    await ScheduleResubscribeAsync().ConfigureAwait(false);
+                }
+                else
+                {
+                    await SetUnsubscribedAsync(ex.Code, ex.Message).ConfigureAwait(false);
+                }
+            }
             catch (Exception ex)
             {
                 OnError("subscribe", ex);
@@ -831,7 +867,8 @@ internal async Task SetUnsubscribedAsync(int code, string reason)
                     }
                     catch
                     {
-                        // Ignore errors during unsubscribe
+                        // Unsubscribe error triggers client disconnect with reconnect (matching centrifuge-js behavior)
+                        await _client.HandleUnsubscribeErrorAsync().ConfigureAwait(false);
                     }
                 }
             }

src/Centrifugal.Centrifuge/Transports/BrowserHttpStreamTransport.cs

@@ -233,19 +233,17 @@ public void OnOpen()
         /// <summary>
         /// JavaScript callback when HTTP stream receives a chunk.
         /// </summary>
-        /// <param name="base64Chunk">Chunk data as base64-encoded string.</param>
+        /// <param name="chunk">Chunk data as byte array.</param>
         [JSInvokable]
-        public void OnChunk(string base64Chunk)
+        public void OnChunk(byte[] chunk)
         {
-            if (string.IsNullOrEmpty(base64Chunk))
+            if (chunk == null || chunk.Length == 0)
             {
                 return;
             }
 
             try
             {
-                // Decode base64 to byte array
-                byte[] chunk = Convert.FromBase64String(base64Chunk);
 
                 lock (_bufferLock)
                 {

src/Centrifugal.Centrifuge/Transports/BrowserWebSocketTransport.cs

@@ -219,19 +219,17 @@ public void OnOpen()
         /// <summary>
         /// JavaScript callback when WebSocket receives a message.
         /// </summary>
-        /// <param name="base64Data">Message data as base64-encoded string.</param>
+        /// <param name="data">Message data as byte array.</param>
         [JSInvokable]
-        public void OnMessage(string base64Data)
+        public void OnMessage(byte[] data)
         {
-            if (string.IsNullOrEmpty(base64Data))
+            if (data == null || data.Length == 0)
             {
                 return;
             }
 
             try
             {
-                // Decode base64 to byte array
-                byte[] data = Convert.FromBase64String(base64Data);
 
                 // Process varint-delimited messages
                 using var ms = new MemoryStream(data);

src/Centrifugal.Centrifuge/Transports/ITransport.cs

@@ -21,7 +21,7 @@ internal interface ITransport : IDisposable
 
         /// <summary>
         /// Gets whether this transport uses emulation mode.
-        /// Emulation mode is used for unidirectional transports (SSE, HTTP Stream)
+        /// Emulation mode is used for unidirectional transports (HTTP Stream at this point)
         /// where sends go through a separate emulation endpoint.
         /// </summary>
         bool UsesEmulation { get; }

src/Centrifugal.Centrifuge/wwwroot/centrifuge-httpstream.js

@@ -107,9 +107,8 @@ window.CentrifugeHttpStream = {
                     break;
                 }
 
-                // Send chunk to .NET as base64 string (JSInterop limitation)
-                const base64 = btoa(String.fromCharCode.apply(null, value));
-                dotnetRef.invokeMethodAsync('OnChunk', base64);
+                // Pass Uint8Array directly - Blazor marshals it to byte[]
+                dotnetRef.invokeMethodAsync('OnChunk', value);
             }
         } catch (error) {
             self.debugLog(debug, '[CentrifugeHttpStream] Read loop error for stream', id, ':', error.name, error.message);

src/Centrifugal.Centrifuge/wwwroot/centrifuge-websocket.js

@@ -42,16 +42,14 @@ window.CentrifugeWebSocket = {
 
             socket.onmessage = (event) => {
                 if (event.data instanceof ArrayBuffer) {
-                    // Convert ArrayBuffer to base64 for .NET interop
+                    // Pass Uint8Array directly - Blazor marshals it to byte[]
                     const bytes = new Uint8Array(event.data);
-                    const base64 = btoa(String.fromCharCode.apply(null, bytes));
-                    dotnetRef.invokeMethodAsync('OnMessage', base64);
+                    dotnetRef.invokeMethodAsync('OnMessage', bytes);
                 } else if (typeof event.data === 'string') {
-                    // Text message - convert to UTF-8 bytes then base64
+                    // Text message - convert to UTF-8 bytes
                     const encoder = new TextEncoder();
                     const bytes = encoder.encode(event.data);
-                    const base64 = btoa(String.fromCharCode.apply(null, bytes));
-                    dotnetRef.invokeMethodAsync('OnMessage', base64);
+                    dotnetRef.invokeMethodAsync('OnMessage', bytes);
                 }
             };