Async queue/dequeue

Author: jjxtra

IPBanProSDKExtensionMethods.cs

@@ -171,7 +171,7 @@ public static IReadOnlyCollection<string> SplitWithNoEmptyEntries(this string te
         /// <param name="socket">Web socket</param>
         /// <param name="subscription">Subscription type</param>
         /// <returns>True if message queued, false if not</returns>
-        public static bool SubscribeWebSocket(this ClientWebSocket socket, IPBanProAPIWebSocketSubscription subscription)
+        public static System.Threading.Tasks.Task<bool> SubscribeWebSocket(this ClientWebSocket socket, IPBanProAPIWebSocketSubscription subscription)
         {
             return socket.QueueMessage(new Message
             {
@@ -189,7 +189,7 @@ public static bool SubscribeWebSocket(this ClientWebSocket socket, IPBanProAPIWe
         /// <param name="socket">Web socket</param>
         /// <param name="subscription">Subscription type</param>
         /// <returns>True if message queued, false if not</returns>
-        public static bool UnsubscribeWebSocket(this ClientWebSocket socket, IPBanProAPIWebSocketSubscription subscription)
+        public static System.Threading.Tasks.Task<bool> UnsubscribeWebSocket(this ClientWebSocket socket, IPBanProAPIWebSocketSubscription subscription)
         {
             return socket.QueueMessage(new Message
             {

Interfaces/IQueueMessage.cs

@@ -31,7 +31,7 @@ public interface IQueueMessage : IDisposable
         /// <param name="message">Message to send, should be a Message or byte[]</param>
         /// <param name="groupId">Group id, or 0 for none</param>
         /// <returns>True if success, false if error</returns>
-        bool QueueMessage(object message, int groupId = 0);
+        System.Threading.Tasks.Task<bool> QueueMessage(object message, int groupId = 0);
     }
 
     /// <summary>
@@ -46,7 +46,7 @@ public interface IQueueMessageServer : IQueueMessage
         /// <param name="message">Message to queue, should be Message or byte[]</param>
         /// <param name="clients">Clients to send to (null for all)</param>
         /// <returns>True if queued, false if no group with the specified id or disposed</returns>
-        bool QueueMessageForClients(int groupId, object message, System.Collections.Generic.IEnumerable<IQueueMessage> clients = null);
+        System.Threading.Tasks.Task<bool> QueueMessageForClients(int groupId, object message, System.Collections.Generic.IEnumerable<IQueueMessage> clients = null);
     }
 
     /// <summary>

WebSocket/ClientWebSocket.cs

@@ -330,7 +330,7 @@ public void Dispose()
         /// <param name="message">Message to send</param>
         /// <param name="groupId">Group id, ignored for now</param>
         /// <returns>True if success, false if error</returns>
-        public bool QueueMessage(object message, int groupId = IPBanProBaseAPI.WebSocketGroupIdNone)
+        public async Task<bool> QueueMessage(object message, int groupId = IPBanProBaseAPI.WebSocketGroupIdNone)
         {
             if (webSocket is null ||
                 message is null ||
@@ -413,7 +413,7 @@ async Task enqueueCallback(IQueueMessage socket)
                 }
             }
 
-            QueueActions(enqueueCallback);
+            await QueueActions(enqueueCallback);
             return true;
         }
 
@@ -444,7 +444,7 @@ public void WaitForAck(string id, int timeoutMilliseconds = 60000)
             }
         }
 
-        private void QueueActions(params Func<IQueueMessage, Task>[] actions)
+        private async Task QueueActions(params Func<IQueueMessage, Task>[] actions)
         {
             if (actions != null && actions.Length != 0)
             {
@@ -467,16 +467,16 @@ private void QueueActions(params Func<IQueueMessage, Task>[] actions)
                     }
                 };
 
-                messageQueue.Enqueue(enqueueActionsFunc);
+                await messageQueue.EnqueueAsync(enqueueActionsFunc);
             }
         }
 
-        private void QueueActionsWithNoExceptions(params Func<IQueueMessage, Task>[] actions)
+        private async Task QueueActionsWithNoExceptions(params Func<IQueueMessage, Task>[] actions)
         {
             if (actions != null && actions.Length != 0)
             {
                 Func<IQueueMessage, Task>[] actionsCopy = actions;
-                messageQueue.Enqueue((Func<Task>)(async () =>
+                await messageQueue.EnqueueAsync((Func<Task>)(async () =>
                 {
                     foreach (var action in actionsCopy.Where(a => a != null))
                     {
@@ -554,7 +554,7 @@ private async Task ReadTask()
 
                     // on connect may make additional calls that must succeed, such as rest calls
                     // for lists, etc.
-                    QueueActionsWithNoExceptions(InvokeConnected);
+                    await QueueActionsWithNoExceptions(InvokeConnected);
 
                     while (webSocket.State == WebSocketState.Open)
                     {
@@ -566,7 +566,7 @@ private async Task ReadTask()
                                 if (result.MessageType == WebSocketMessageType.Close)
                                 {
                                     await webSocket.CloseAsync(WebSocketCloseStatus.NormalClosure, string.Empty, cancellationTokenSource.Token);
-                                    QueueActions(InvokeDisconnected);
+                                    await QueueActions(InvokeDisconnected);
                                 }
                                 else
                                 {
@@ -586,7 +586,7 @@ private async Task ReadTask()
                             if (result.MessageType == WebSocketMessageType.Text)
                             {
                                 string text = Encoding.UTF8.GetString(stream.GetBuffer(), 0, (int)stream.Length);
-                                messageQueue.Enqueue(text);
+                                await messageQueue.EnqueueAsync(text);
                             }
                             // otherwise treat message as binary
                             else
@@ -618,12 +618,12 @@ private async Task ReadTask()
                                     }
                                     else
                                     {
-                                        messageQueue.Enqueue(message);
+                                        await messageQueue.EnqueueAsync(message);
                                     }
                                 }
                                 else
                                 {
-                                    messageQueue.Enqueue(stream.ToArray());
+                                    await messageQueue.EnqueueAsync(stream.ToArray());
                                 }
                             }
                         }
@@ -648,7 +648,7 @@ private async Task ReadTask()
                 {
                     if (wasConnected)
                     {
-                        QueueActions(InvokeDisconnected);
+                        await QueueActions(InvokeDisconnected);
                     }
 
                     Logger.Info("Client web socket was disconnected from {0}, attempting reconnection...", uri);
@@ -701,7 +701,7 @@ private async Task MessageTask()
                 {
                     Logger.Debug("Sending ping from client web socket connection to {0}", Uri);
                     lastPing = IPBanService.UtcNow;
-                    QueueMessage("ping");
+                    await QueueMessage("ping");
                 }
 
                 if ((result = await messageQueue.TryDequeueAsync(messageQueueTimeout, cancellationTokenSource.Token)).Key)
@@ -755,7 +755,7 @@ private async Task MessageTask()
                     lastCheck = IPBanService.UtcNow;
 
                     // this must succeed, the callback may be requests lists or other resources that must not fail
-                    QueueActionsWithNoExceptions(InvokeConnected);
+                    await QueueActionsWithNoExceptions(InvokeConnected);
                 }
             }
         }