Queue test

Libraries/Esiur/Core/AsyncQueue.cs

@@ -93,6 +93,9 @@ public class AsyncQueue<T> : AsyncReply<T>
     }
 
     public void Add(AsyncReply<T> reply)
+        => Add(reply, !reply.Ready);
+
+    public void Add(AsyncReply<T> reply, bool hasResource)
     {
         lock (queueLock)
         {
@@ -103,7 +106,7 @@ public class AsyncQueue<T> : AsyncReply<T>
                 NotificationsCountWaitingInTheQueueAtEnqueueing = list.Count,
                 Reply = reply,
                 Arrival = DateTime.Now,
-                HasResource = !reply.Ready
+                HasResource = hasResource
             });
         }
 

Libraries/Esiur/Protocol/EpConnectionProtocol.cs

@@ -606,12 +606,13 @@ partial class EpConnection
             if (pt == null)
                 return;
 
-            Codec.ParseAsync(tdu.Data, valueOffset, this, null).Then(pr =>
+            void EnqueueParsedProperty(ParsedTdu parsed)
             {
-                if (pr.Value is AsyncReply asyncReply)
+                var value = Codec.ParseAsync(parsed, this, null);
+                if (value is AsyncReply asyncReply)
                 {
                     var item = new AsyncReply<EpResourceQueueItem>();
-                    _queue.Add(item);
+                    _queue.Add(item, hasResource: true);
 
                     asyncReply.Then((result) =>
                     {
@@ -624,14 +625,27 @@ partial class EpConnection
                 {
                     _queue.Add(new AsyncReply<EpResourceQueueItem>(new EpResourceQueueItem((EpResource)r,
                                                     EpResourceQueueItem.DistributedResourceQueueItemType.Propery,
-                                                    pr.Value, index)));
+                                                    value, index)), hasResource: false);
+                }
             }
 
-            }).Error((ex) =>
+            var parsed = ParsedTdu.Parse(tdu.Data, valueOffset, (uint)tdu.Data.Length, this);
+            if (parsed is ParsedTdu parsedTdu)
+            {
+                EnqueueParsedProperty(parsedTdu);
+            }
+            else if (parsed is AsyncReply<ParsedTdu> parsedReply)
+            {
+                parsedReply.Then(EnqueueParsedProperty).Error((ex) =>
                 {
                     //.Error(x => SendError(ErrorType.Management, callback, (ushort)ExceptionCode.ParseError));
                     throw ex;
                 });
+            }
+            else
+            {
+                throw new NullReferenceException("DataType can't be parsed.");
+            }
 
         });
     }

Tests/Distribution/Queueing/Server/Program.cs

@@ -29,8 +29,10 @@ await wh.Open();
 long memAfter = GC.GetTotalMemory(forceFullCollection: true);
 double memMB = (memAfter - memBefore) / (1024.0 * 1024.0);
 
-Console.WriteLine("Press ENTER to stop.");
+Console.WriteLine("Ready. Press Ctrl+C to stop.");
-Console.ReadLine();
+var stop = new TaskCompletionSource();
+Console.CancelKeyPress += (_, e) => { e.Cancel = true; stop.TrySetResult(); };
+await stop.Task;
 await wh.Close();
 
 

Tests/Unit/AsyncQueueTests.cs

@@ -35,6 +35,26 @@ public class AsyncQueueTests
         Assert.Empty(queue.DrainProcessed());
     }
 
+    [Fact]
+    public void DrainProcessed_PreservesExplicitResourceWorkFlag()
+    {
+        var queue = new AsyncQueue<int>();
+        queue.SetProcessedCapture(true);
+
+        queue.Add(new AsyncReply<int>(1), hasResource: false);
+
+        var pending = new AsyncReply<int>();
+        queue.Add(pending, hasResource: true);
+        pending.Trigger(2);
+
+        var processed = queue.DrainProcessed();
+
+        Assert.Collection(
+            processed,
+            x => Assert.False(x.HasResource),
+            x => Assert.True(x.HasResource));
+    }
+
     [Fact]
     public void DisablingCapture_DiscardsHistoryAndStopsCapturing()
     {