add async support for RegisterQueue in c#

Author: skadefro

dotnet/CallbackRegistry.cs

@@ -23,10 +23,14 @@ public bool TryGetCallback<T>(int id, out TaskCompletionSource<T>? tcs)
                 tcs = typedTcs;
                 return true;
             } else {
+                #if DEBUG
                 Console.WriteLine("Failed to get callback for id: " + id + " is of wrong type " + obj.GetType() + " expected " + typeof(TaskCompletionSource<T>));
+                #endif
             }
         } else {
+            #if DEBUG
             Console.WriteLine("Failed to get callback for id: " + id);
+            #endif
         }
         tcs = null;
         return false;
@@ -39,10 +43,14 @@ public bool TryGetCallback(int id, out TaskCompletionSource? tcs)
                 tcs = typedTcs;
                 return true;
             } else {
+                #if DEBUG
                 Console.WriteLine("Failed to get callback for id: " + id + " is of wrong type " + obj.GetType() + " expected " + typeof(TaskCompletionSource));
+                #endif
             }
         } else {
+            #if DEBUG
             Console.WriteLine("Failed to get callback for id: " + id);
+            #endif
         }
         tcs = null;
         return false;
@@ -128,10 +136,14 @@ public bool TryGetCallback<T>(int id, out Action<T>? tcs)
                 tcs = typedTcs;
                 return true;
             } else {
+                #if DEBUG
                 Console.WriteLine("Failed to get callback for id: " + id + " is of wrong type " + obj.GetType() + " expected " + typeof(Action<T>));
+                #endif
             }
         } else {
+            #if DEBUG
             Console.WriteLine("Failed to get callback for id: " + id);
+            #endif
         }
         tcs = null;
         return false;
@@ -145,10 +157,14 @@ public bool TryGetCallback(int id, out Action? tcs)
                 tcs = typedTcs;
                 return true;
             } else {
+                #if DEBUG
                 Console.WriteLine("Failed to get callback for id: " + id + " is of wrong type " + obj.GetType() + " expected " + typeof(Action));
+                #endif
             }
         } else {
+            #if DEBUG
             Console.WriteLine("Failed to get callback for id: " + id);
+            #endif
         }
         tcs = null;
         return false;
@@ -214,7 +230,9 @@ public bool TryGetCallback<TIn, TOut>(int id, out Func<TIn, TOut>? func)
                 func = typedFunc;
                 return true;
             }
+            #if DEBUG
             Console.WriteLine($"Failed to get callback for id: {id} is of wrong type {obj.GetType()} expected {typeof(Func<TIn, TOut>)}");
+            #endif
         }
         func = null;
         return false;

dotnet/Client.cs

@@ -1235,7 +1235,7 @@ public Client(string libPath = "")
             _WatchCallbackDelegate = _WatchCallback;
             _WatchEventCallbackDelegate = _WatchEventCallback;
             _DropCollectionCallbackDelegate = _DropCollectionCallback;
-            _QueueEventCallbackDelegate = _QueueEventCallback;
+            _QueueEventCallbackDelegate = _QueueEventCallbackWrapper;
             _ExchangeEventCallbackDelegate = _ExchangeEventCallback;
             _RpcResponseCallbackDelegate = _RpcResponseCallback;
             _CustomCommandCallbackDelegate = _CustomCommandCallback;
@@ -3086,7 +3086,11 @@ public void UnWatch(string watchid)
                 Marshal.FreeHGlobal(watchidPtr);
             }
         }
-        IntPtr _QueueEventCallback(IntPtr QueueEventWrapperptr)
+        private IntPtr _QueueEventCallbackWrapper(IntPtr QueueEventWrapperptr)
+        {
+            return _QueueEventCallback(QueueEventWrapperptr).GetAwaiter().GetResult();
+        }
+        private async Task<IntPtr> _QueueEventCallback(IntPtr QueueEventWrapperptr)
         {
             try
             {
@@ -3115,6 +3119,26 @@ IntPtr _QueueEventCallback(IntPtr QueueEventWrapperptr)
                     {
                         return Marshal.StringToHGlobalAnsi(result);
                     }
+                } else if (QueueFuncRegistry.TryGetCallback<QueueEvent, Task<string>>(eventObj.request_id, out var asyncFuncHandler))
+                {
+                    if (asyncFuncHandler == null)
+                    {
+                        return IntPtr.Zero;
+                    }
+                    var queueEvent = new QueueEvent
+                    {
+                        queuename = Marshal.PtrToStringAnsi(eventObj.queuename) ?? string.Empty,
+                        correlation_id = Marshal.PtrToStringAnsi(eventObj.correlation_id) ?? string.Empty,
+                        replyto = Marshal.PtrToStringAnsi(eventObj.replyto) ?? string.Empty,
+                        routingkey = Marshal.PtrToStringAnsi(eventObj.routingkey) ?? string.Empty,
+                        exchangename = Marshal.PtrToStringAnsi(eventObj.exchangename) ?? string.Empty,
+                        data = Marshal.PtrToStringAnsi(eventObj.data) ?? string.Empty,
+                    };
+                    var result = await asyncFuncHandler(queueEvent);
+                    if (!string.IsNullOrEmpty(result))
+                    {
+                        return Marshal.StringToHGlobalAnsi(result);
+                    }
                 }
                 return IntPtr.Zero;
             }
@@ -3129,7 +3153,7 @@ IntPtr _QueueEventCallback(IntPtr QueueEventWrapperptr)
             }
         }
 
-        public string RegisterQueue(string queuename, Func<QueueEvent, string> eventHandler)
+        public string RegisterQueue(string queuename, Func<QueueEvent, Task<string>> eventHandler)
         {
             if (eventHandler == null) throw new ArgumentNullException(nameof(eventHandler));
             IntPtr queuenamePtr = Marshal.StringToHGlobalAnsi(queuename);
@@ -3172,7 +3196,7 @@ public string RegisterQueueAction(string queuename, Action<QueueEvent> eventHand
         {
             return RegisterQueue(queuename, (queueEvent) => {
                 eventHandler(queueEvent);
-                return string.Empty;
+                return Task.FromResult(string.Empty);
             });
         }
 

dotnet/Program.cs

@@ -647,13 +647,8 @@ await client.CreateCollection("testdotnettscollection", timeseries:
             var queueId = client.RegisterQueue("test2queue", e =>
             {
               Console.WriteLine("Queue event received from " + e.queuename + " with data: " + e.data);
-              // if(!string.IsNullOrEmpty(e.replyto)) {
-              //     var t = System.Threading.Tasks.Task.Run(() => {
-              //         var message = "{\"payload\": \"Bettina\"}";
-              //         _ = client.QueueMessage(message, e.replyto, striptoken: true, correlation_id: e.correlation_id);
-              //     });
-              // }
-              return "{\"payload\": \"Bettina\"}";
+              //return "{\"payload\": \"Bettina\"}";
+              return Task.FromResult("{\"payload\": \"Bettina\"}");
             });
             Console.WriteLine("Queue registered with id: " + queueId);
           }