[MWB] - pub-sub spike - more poc code

src/modules/MouseWithoutBorders/App/Class/Program.cs

@@ -31,6 +31,7 @@
 using ManagedCommon;
 using Microsoft.PowerToys.Settings.UI.Library.Utilities;
 using Microsoft.PowerToys.Telemetry;
+using MouseWithoutBorders.Messaging;
 using Newtonsoft.Json;
 using StreamJsonRpc;
 
@@ -52,6 +53,9 @@ internal static class Program
 
         public static bool ShowServiceModeErrorTooltip;
 
+        public static PacketProducer PacketProducer
+            => Program.CreatePacketProducer();
+
         [STAThread]
         private static void Main()
         {
@@ -229,6 +233,8 @@ private static void Main()
 
                 Application.Run(formScreen);
 
+                Program.PacketProducer.Queue.Complete();
+
                 etwTrace?.Dispose();
             }
             catch (Exception e)
@@ -237,6 +243,25 @@ private static void Main()
             }
         }
 
+        private static PacketProducer CreatePacketProducer()
+        {
+            // make a producer that we'll use to push messages onto a queue
+            var producer = new PacketProducer();
+
+            // create a demo consumer to log the types of packages,
+            // subscribe it to the producer's queue and then start it
+            PacketConsumer loggingConsumer = new(
+                (DATA packet, CancellationToken cancellationToken) =>
+                {
+                    Logger.Log($"received packet of type {packet.Type}");
+                    return Task.CompletedTask;
+                });
+            producer.Queue.Subscribe(loggingConsumer);
+            _ = Task.Run(() => loggingConsumer.StartAsync());
+
+            return producer;
+        }
+
         private interface ISettingsSyncHelper
         {
             [JsonObject(MemberSerialization.OptIn)]

src/modules/MouseWithoutBorders/App/Core/Receiver.cs

@@ -85,6 +85,11 @@ private static bool PreProcess(DATA package)
 
     internal static void ProcessPackage(DATA package, TcpSk tcp)
     {
+#pragma warning disable VSTHRD002
+        Program.PacketProducer.WriteAsync(package)
+            .AsTask().GetAwaiter().GetResult();
+#pragma warning restore VSTHRD002
+
         if (!PreProcess(package))
         {
             return;

src/modules/MouseWithoutBorders/App/Messaging/PacketConsumer.cs

@@ -85,7 +85,7 @@ public async Task DrainAsync(CancellationToken cancellationToken = default)
         }
     }
 
-    public void Stop()
+    public void Complete()
     {
         this.Channel.Writer.Complete();
     }

src/modules/MouseWithoutBorders/App/Messaging/PacketQueue.cs

@@ -77,4 +77,25 @@ public bool TryWrite(DATA packet)
 
         return result;
     }
+
+    public void Complete()
+    {
+        foreach (var consumer in this.Consumers)
+        {
+            consumer.Complete();
+        }
+    }
+
+    /// <summary>
+    /// Reads and processes all messages currently on the queue until it is empty.
+    /// Any messages that arrive while draining will be read and processed as well.
+    /// Does *not* "Complete" the queue, just leaves it empty.
+    /// </summary>
+    public async Task DrainAsync(CancellationToken cancellationToken = default)
+    {
+        foreach (var consumer in this.Consumers)
+        {
+            await consumer.DrainAsync(cancellationToken);
+        }
+    }
 }