Add support for browser platform

sdks/csharp/SpacetimeDB.ClientSDK.csproj

@@ -26,6 +26,7 @@
 
   <ItemGroup>
     <PackageReference Include="SpacetimeDB.BSATN.Runtime" Version="1.2.*" />
+	<PackageReference Include="System.Threading.Channels" Version="9.0.7" />
 
     <InternalsVisibleTo Include="SpacetimeDB.Tests" />
   </ItemGroup>

sdks/csharp/src/SpacetimeDBClient.cs

@@ -1,16 +1,16 @@
 using System;
 using System.Collections;
-using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.IO;
 using System.IO.Compression;
 using System.Linq;
+using System.Runtime.CompilerServices;
 using System.Threading;
+using System.Threading.Channels;
 using System.Threading.Tasks;
 using SpacetimeDB.BSATN;
 using SpacetimeDB.Internal;
 using SpacetimeDB.ClientApi;
-using Thread = System.Threading.Thread;
 using System.Diagnostics;
 
 
@@ -167,7 +167,7 @@ public abstract class DbConnectionBase<DbConnection, Tables, Reducer> : IDbConne
         private readonly Dictionary<Guid, TaskCompletionSource<OneOffQueryResponse>> waitingOneOffQueries = new();
 
         private bool isClosing;
-        private readonly Thread networkMessageParseThread;
+        private readonly ConfiguredTaskAwaitable networkMessageParseTask;
         public readonly Stats stats = new();
 
         protected DbConnectionBase()
@@ -198,8 +198,7 @@ protected DbConnectionBase()
 
 #if !(UNITY_WEBGL && !UNITY_EDITOR)
             // For targets other than webgl we start a thread to parse messages
-            networkMessageParseThread = new Thread(ParseMessages);
-            networkMessageParseThread.Start();
+            networkMessageParseTask = ParseMessages().ConfigureAwait(false);
 #endif
         }
 
@@ -257,14 +256,22 @@ internal struct ParsedMessage
             public ReducerEvent<Reducer>? reducerEvent;
         }
 
-        private readonly BlockingCollection<UnparsedMessage> _parseQueue =
-            new(new ConcurrentQueue<UnparsedMessage>());
+        private readonly Channel<UnparsedMessage> _parseQueue =
+            Channel.CreateUnbounded<UnparsedMessage>(new UnboundedChannelOptions
+            {
+                SingleReader = false,
+                SingleWriter = false
+            });
 
-        private readonly BlockingCollection<ParsedMessage> _applyQueue =
-            new(new ConcurrentQueue<ParsedMessage>());
+        private readonly Channel<ParsedMessage> _applyQueue =
+            Channel.CreateUnbounded<ParsedMessage>(new UnboundedChannelOptions
+            {
+                SingleReader = false,
+                SingleWriter = false
+            });
 
         internal static bool IsTesting;
-        internal bool HasMessageToApply => _applyQueue.Count > 0;
+        internal bool HasMessageToApply => _applyQueue.Reader.Count > 0;
 
         private readonly CancellationTokenSource _parseCancellationTokenSource = new();
         private CancellationToken _parseCancellationToken => _parseCancellationTokenSource.Token;
@@ -413,7 +420,7 @@ private static (BinaryReader reader, int rowCount) ParseRowList(BsatnRowList lis
 #if UNITY_WEBGL && !UNITY_EDITOR
         internal IEnumerator ParseMessages()
 #else
-        internal void ParseMessages()
+        internal async Task ParseMessages()
 #endif
         {
             while (!isClosing)
@@ -425,9 +432,9 @@ internal void ParseMessages()
 #endif
                 try
                 {
-                    var message = _parseQueue.Take(_parseCancellationToken);
+                    var message = await _parseQueue.Reader.ReadAsync(_parseCancellationToken);
                     var parsedMessage = ParseMessage(message);
-                    _applyQueue.Add(parsedMessage, _parseCancellationToken);
+                    await _applyQueue.Writer.WriteAsync(parsedMessage, _parseCancellationToken);
                 }
                 catch (OperationCanceledException)
                 {
@@ -751,7 +758,7 @@ async Task Function()
                 }
                 _ = Function();
 #else
-                });
+                }).ConfigureAwait(false);
 #endif
             }
         }
@@ -939,7 +946,7 @@ private void ApplyMessage(ParsedMessage parsed)
         // Note: this method is called from unit tests.
         internal void OnMessageReceived(byte[] bytes, DateTime timestamp)
         {
-            _parseQueue.Add(new UnparsedMessage { bytes = bytes, timestamp = timestamp, parseQueueTrackerId = stats.ParseMessageQueueTracker.StartTrackingRequest() });
+            _parseQueue.Writer.TryWrite(new UnparsedMessage { bytes = bytes, timestamp = timestamp, parseQueueTrackerId = stats.ParseMessageQueueTracker.StartTrackingRequest() });
         }
 
         // TODO: this should become [Obsolete] but for now is used by autogenerated code.
@@ -1071,7 +1078,7 @@ T[] LogAndThrow(string error)
         public void FrameTick()
         {
             webSocket.Update();
-            while (_applyQueue.TryTake(out var parsedMessage))
+            while (_applyQueue.Reader.TryRead(out var parsedMessage))
             {
                 ApplyMessage(parsedMessage);
             }

sdks/csharp/src/WebSocket.cs

@@ -167,19 +167,24 @@ public async Task Connect(string? auth, string host, string nameOrAddress, Conne
             {
                 uri += "&light=true";
             }
-            var url = new Uri(uri);
             Ws.Options.AddSubProtocol(_options.Protocol);
 
             var source = new CancellationTokenSource(10000);
             if (!string.IsNullOrEmpty(auth))
             {
-                Ws.Options.SetRequestHeader("Authorization", $"Bearer {auth}");
-            }
-            else
-            {
-                Ws.Options.UseDefaultCredentials = true;
+                if (RuntimeInformation.IsOSPlatform(OSPlatform.Create("Browser")))
+                {
+                    // For WebAssembly, we need to pass the auth token as a query parameter
+                    uri += $"&token={Uri.EscapeDataString(auth)}";
+                }
+                else
+                {
+                    Ws.Options.SetRequestHeader("Authorization", $"Bearer {auth}");
+                }
             }
 
+            var url = new Uri(uri);
+
             try
             {
                 await Ws.ConnectAsync(url, source.Token);