fix: Consume notification when breaking in await foreach of Stream

Note that currently there might still be notifications left after removing the listener. Tangentially related to #68.
obecto · Dec 14, 2020 · d8e0664 · d8e0664
1 parent 66bbbba
commit d8e0664
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 5 deletions.
diff --git a/functions/runtime/src/Perper.WebJobs.Extensions/Model/Stream.cs b/functions/runtime/src/Perper.WebJobs.Extensions/Model/Stream.cs
@@ -143,11 +143,16 @@ private async IAsyncEnumerable<T> Impl([EnumeratorCancellation] CancellationToke
 
                             await ((State)_stream._state).LoadStateEntries();
 
-                            yield return (T)converter.Invoke(value)!;
-
-                            await ((State)_stream._state).StoreStateEntries();
+                            try
+                            {
+                                yield return (T)converter.Invoke(value)!;
+                            }
+                            finally
+                            {
+                                await ((State)_stream._state).StoreStateEntries();
 
-                            await _stream._fabric.ConsumeNotification(key);
+                                await _stream._fabric.ConsumeNotification(key);
+                            }
                         }
                     }
                 }

diff --git a/functions/runtime/src/Perper.WebJobs.Extensions/Triggers/PerperTriggerValueBinder.cs b/functions/runtime/src/Perper.WebJobs.Extensions/Triggers/PerperTriggerValueBinder.cs
@@ -35,7 +35,7 @@ public PerperTriggerValueBinder(JObject trigger, IIgniteClient ignite, PerperBin
             return Task.FromResult<object?>(null);
         }
 
-        public async Task SetValueAsync(object value, CancellationToken cancellationToken)
+        public async Task SetValueAsync(object? value, CancellationToken cancellationToken)
         {
             if (_trigger.ContainsKey("Call"))
             {
@@ -50,6 +50,11 @@ public async Task SetValueAsync(object value, CancellationToken cancellationToke
             {
                 var stream = (string)_trigger["Stream"]!;
 
+                if (value == null)
+                {
+                    return;
+                }
+
                 var asyncEnumerableInterface = PerperTypeUtils.GetGenericInterface(value.GetType(), typeof(IAsyncEnumerable<>));
                 if (asyncEnumerableInterface == null)
                 {