Merge pull request #4 from ledjon-behluli/fix-pass-through-stream

Fix missing streamChannel write for BaseAgent
ledjon-behluli · Aug 26, 2023 · 99afa47 · 99afa47
2 parents 6ba4772 + 69e79e5
commit 99afa47
Show file tree

Hide file tree

Showing 3 changed files with 18 additions and 12 deletions.
diff --git a/src/OrleanSpaces/Agents/BaseAgent.cs b/src/OrleanSpaces/Agents/BaseAgent.cs
@@ -50,10 +50,7 @@ async ValueTask ISpaceRouter<TTuple, TTemplate>.RouteAction(TupleAction<TTuple>
                 case TupleActionType.Insert:
                     {
                         tuples = tuples.Add(action.Tuple);
-                        if (streamChannel is not null)
-                        {
-                            await streamChannel.Writer.WriteAsync(action.Tuple);
-                        }
+                        await streamChannel.WriteIfNotNull(action.Tuple);
                     }
                     break;
                 case TupleActionType.Remove:
@@ -86,6 +83,8 @@ public async Task WriteAsync(TTuple tuple)
         ThrowHelpers.EmptyTuple(tuple);
 
         await tupleStore.Insert(new(agentId, tuple, TupleActionType.Insert));
+        await streamChannel.WriteIfNotNull(tuple);
+
         tuples = tuples.Add(tuple);
     }
 

diff --git a/src/OrleanSpaces/Agents/SpaceAgent.cs b/src/OrleanSpaces/Agents/SpaceAgent.cs
@@ -47,10 +47,7 @@ async ValueTask ISpaceRouter<SpaceTuple, SpaceTemplate>.RouteAction(TupleAction<
                 case TupleActionType.Insert:
                     {
                         tuples = tuples.Add(action.Tuple);
-                        if (streamChannel is not null)
-                        {
-                            await streamChannel.Writer.WriteAsync(action.Tuple);
-                        }
+                        await streamChannel.WriteIfNotNull(action.Tuple);
                     }
                     break;
                 case TupleActionType.Remove:
@@ -81,12 +78,11 @@ public void Unsubscribe(Guid observerId)
     public async Task WriteAsync(SpaceTuple tuple)
     {
         ThrowHelpers.EmptyTuple(tuple);
+
         await tupleStore.Insert(new(agentId, tuple, TupleActionType.Insert));
+        await streamChannel.WriteIfNotNull(tuple);
+
         tuples = tuples.Add(tuple);
-        if (streamChannel is not null)
-        {
-            await streamChannel.Writer.WriteAsync(tuple);
-        }
     }
 
     public ValueTask EvaluateAsync(Func<Task<SpaceTuple>> evaluation)

diff --git a/src/OrleanSpaces/Helpers/Helpers.cs b/src/OrleanSpaces/Helpers/Helpers.cs
@@ -1,5 +1,6 @@
 using System.Buffers;
 using System.Runtime.CompilerServices;
+using System.Threading.Channels;
 using OrleanSpaces.Helpers;
 using OrleanSpaces.Tuples;
 
@@ -46,4 +47,14 @@ public static bool AllocateAndExecute<T, TConsumer>(this TConsumer consumer, int
             return result;
         }
     }
+
+    [MethodImpl(MethodImplOptions.AggressiveInlining)]
+    public static async Task WriteIfNotNull<T>(this Channel<T>? channel, T tuple)
+        where T : ISpaceTuple
+    {
+        if (channel is not null)
+        {
+            await channel.Writer.WriteAsync(tuple);
+        }
+    }
 }