Add support for WALRUS AccessToken Pushes on every heartbeat see [#12…

…](supabase-community/supabase-csharp#12)

Realtime/Channel.cs

@@ -95,6 +95,11 @@ public enum ChannelState
         /// </summary>
         public string Topic { get => Utils.GenerateChannelTopic(database, schema, table, col, value); }
 
+        /// <summary>
+        /// Flag stating whether a channel has been joined once or not.
+        /// </summary>
+        public bool HasJoinedOnce { get; private set; }
+
         private string database;
         private string schema;
         private string table;
@@ -108,7 +113,7 @@ public enum ChannelState
 
         private bool canPush => IsJoined && Socket.IsConnected;
         private bool hasJoinedOnce = false;
-        private List<Push> buffer = new List<Push>();
+        internal List<Push> buffer = new List<Push>();
         private Timer rejoinTimer;
         private bool isRejoining = false;
 
@@ -166,6 +171,7 @@ public Task<Channel> Subscribe(int timeoutMs = Constants.DEFAULT_TIMEOUT)
                 {
                     // Success!
                     case ChannelState.Joined:
+                        HasJoinedOnce = true;
                         StateChanged -= channelCallback;
                         JoinPush.OnTimeout -= joinPushTimeoutCallback;
 

Realtime/Client.cs

@@ -1,5 +1,6 @@
 using System;
 using System.Collections.Generic;
+using System.Collections.ObjectModel;
 using System.Diagnostics;
 using System.Threading.Tasks;
 using Newtonsoft.Json;
@@ -24,7 +25,12 @@ public class Client
         /// Keys are of encoded value: `{database}{:schema?}{:table?}{:col.eq.:value?}`
         /// Values are of type `Channel<T> where T : BaseModel, new()`;
         /// </summary>
-        private Dictionary<string, Channel> subscriptions { get; set; }
+        public Dictionary<string, Channel> subscriptions { get; set; }
+
+        /// <summary>
+        /// Exposes all Realtime Channel Subscriptions for R/O public consumption 
+        /// </summary>
+        public ReadOnlyDictionary<string, Channel> Subscriptions => new ReadOnlyDictionary<string, Channel>(subscriptions);
 
         /// <summary>
         /// The backing Socket class.
@@ -106,6 +112,11 @@ public JsonSerializerSettings SerializerSettings
 
         private string realtimeUrl;
 
+        /// <summary>
+        /// JWT Access token for WALRUS security
+        /// </summary>
+        internal string AccessToken { get; private set; }
+
         /// <summary>
         /// Initializes a Client instance, this method should be called prior to any other method.
         /// </summary>
@@ -242,6 +253,34 @@ public Client Disconnect(CloseStatusCode code = WebSocketSharp.CloseStatusCode.N
             return this;
         }
 
+        /// <summary>
+        /// Sets the JWT access token used for channel subscription authorization and Realtime RLS.
+        /// Ref: https://github.com/supabase/realtime-js/pull/117 | https://github.com/supabase/realtime-js/pull/117
+        /// </summary>
+        /// <param name="jwt"></param>
+        public void SetAuth(string jwt)
+        {
+            AccessToken = jwt;
+
+            try
+            {
+                foreach (var channel in subscriptions.Values)
+                {
+                    if (channel.HasJoinedOnce && channel.IsJoined)
+                    {
+                        channel.Push(Constants.CHANNEL_ACCESS_TOKEN, new Dictionary<string, string>
+                        {
+                            { "access_token", AccessToken }
+                        });
+                    }
+                }
+            }
+            catch (Exception ex)
+            {
+                Debug.WriteLine(ex);
+            }
+        }
+
         /// <summary>
         /// Adds a Channel subscription - if a subscription exists with the same signature, the existing subscription will be returned.
         /// </summary>
@@ -295,6 +334,9 @@ private void HandleSocketStateChanged(object sender, SocketStateChangedEventArgs
             switch (args.State)
             {
                 case SocketStateChangedEventArgs.ConnectionState.Open:
+                    // Ref: https://github.com/supabase/realtime-js/pull/116/files
+                    SetAuth(AccessToken);
+
                     OnOpen?.Invoke(this, args);
                     break;
                 case SocketStateChangedEventArgs.ConnectionState.Close:

Realtime/Constants.cs

@@ -56,5 +56,7 @@ public enum EventType
         public static string PHEONIX_STATUS_ERROR = "error";
 
         public static string TRANSPORT_WEBSOCKET = "websocket";
+
+        public static string CHANNEL_ACCESS_TOKEN = "access_token";
     }
 }

Realtime/Socket.cs

@@ -160,6 +160,12 @@ private void SendHeartbeat()
             pendingHeartbeatRef = MakeMsgRef();
 
             Push(new SocketRequest { Topic = "phoenix", Event = "heartbeat", Ref = pendingHeartbeatRef.ToString() });
+
+            // Ref: https://github.com/supabase/realtime-js/pull/116
+            if (!string.IsNullOrEmpty(Client.Instance.AccessToken))
+            {
+                Client.Instance.SetAuth(Client.Instance.AccessToken);
+            }
         }
 
         /// <summary>