Add streams-based trigger, include new tests that ensure multiple functions instances don't duplicate events

mapalan · web-flow · commit 048433d14879 · 2023-02-15T19:46:18.000-08:00
diff --git a/samples/RedisSamples.cs b/samples/RedisSamples.cs
@@ -46,5 +46,22 @@ public static void ListsMultipleTrigger(
         {
             logger.LogInformation(JsonSerializer.Serialize(model));
         }
+
+        [FunctionName(nameof(StreamsTrigger))]
+        public static void StreamsTrigger(
+            [RedisStreamsTrigger(ConnectionString = localhost, Keys = "streamTest")] RedisMessageModel model,
+            ILogger logger)
+        {
+            logger.LogInformation(JsonSerializer.Serialize(model));
+        }
+
+        [FunctionName(nameof(StreamsMultipleTriggers))]
+        public static void StreamsMultipleTriggers(
+            [RedisStreamsTrigger(ConnectionString = localhost, Keys = "streamTest1 streamTest2")] RedisMessageModel model,
+            ILogger logger)
+        {
+            logger.LogInformation(JsonSerializer.Serialize(model));
+        }
+
     }
 }
diff --git a/src/RedisExtensionConfigProvider.cs b/src/RedisExtensionConfigProvider.cs
@@ -32,11 +32,14 @@ public void Initialize(ExtensionConfigContext context)
             }
 
 #pragma warning disable CS0618
-            FluentBindingRule<RedisPubSubTriggerAttribute> rule = context.AddBindingRule<RedisPubSubTriggerAttribute>();
-            rule.BindToTrigger<RedisMessageModel>(new RedisPubSubTriggerBindingProvider(configuration));
+            FluentBindingRule<RedisPubSubTriggerAttribute> pubsubTriggerRule = context.AddBindingRule<RedisPubSubTriggerAttribute>();
+            pubsubTriggerRule.BindToTrigger<RedisMessageModel>(new RedisPubSubTriggerBindingProvider(configuration));
 
             FluentBindingRule<RedisListsTriggerAttribute> listsTriggerRule = context.AddBindingRule<RedisListsTriggerAttribute>();
             listsTriggerRule.BindToTrigger<RedisMessageModel>(new RedisListsTriggerBindingProvider(configuration));
+
+            FluentBindingRule<RedisStreamsTriggerAttribute> streamsTriggerRule = context.AddBindingRule<RedisStreamsTriggerAttribute>();
+            streamsTriggerRule.BindToTrigger<RedisMessageModel>(new RedisStreamsTriggerBindingProvider(configuration));
 #pragma warning restore CS0618
         }
     }
diff --git a/src/StreamsTrigger/RedisStreamsListener.cs b/src/StreamsTrigger/RedisStreamsListener.cs
@@ -0,0 +1,107 @@
+﻿using System;
+using System.Linq;
+using System.Text.Json;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Azure.WebJobs.Host.Executors;
+using StackExchange.Redis;
+
+
+namespace Microsoft.Azure.WebJobs.Extensions.Redis
+{
+    /// <summary>
+    /// Responsible for managing connections and listening to a given Azure Redis Cache.
+    /// </summary>
+    internal sealed class RedisStreamsListener : RedisPollingListenerBase
+    {
+        internal bool deleteAfterProcess;
+        internal string consumerGroup;
+        internal StreamPosition[] positions;
+        internal string consumerName;
+
+        public RedisStreamsListener(string connectionString, string keys, TimeSpan pollingInterval, int messagesPerWorker, int batchSize, string consumerGroup, bool deleteAfterProcess, ITriggeredFunctionExecutor executor)
+            : base(connectionString, keys, pollingInterval, messagesPerWorker, batchSize, executor)
+        {
+            this.consumerGroup = consumerGroup;
+            this.deleteAfterProcess = deleteAfterProcess;
+            this.positions = this.keys.Select((key) => new StreamPosition(key, StreamPosition.NewMessages)).ToArray();
+            this.consumerName = Guid.NewGuid().ToString();
+        }
+
+        public override async void BeforePolling()
+        {
+            IDatabase db = multiplexer.GetDatabase();
+
+            // create consumer group for each stream key
+            foreach (RedisKey key in keys)
+            {
+                try
+                {
+                    if (!await db.StreamCreateConsumerGroupAsync(key, consumerGroup))
+                    {
+                        throw new Exception($"Could not create consumer group for stream key {key}");
+                    }
+                }
+                catch (RedisServerException e)
+                {
+                    // consumer group already exists
+                    if (!e.Message.Contains("BUSYGROUP"))
+                    {
+                        throw;
+                    }
+                }
+            }
+        }
+
+        public override async Task PollAsync(CancellationToken cancellationToken)
+        {
+            IDatabase db = multiplexer.GetDatabase();
+            RedisStream[] streams = await db.StreamReadGroupAsync(positions, consumerGroup, consumerName, batchSize);
+
+            for (int i = 0; i < streams.Length; i++)
+            {
+                if (streams[i].Entries.Length > 0)
+                {
+                    foreach (StreamEntry entry in streams[i].Entries)
+                    {
+                        var triggerValue = new RedisMessageModel
+                        {
+                            Trigger = streams[i].Key,
+                            Message = JsonSerializer.Serialize(entry.Values.ToDictionary(value => value.Name.ToString(), value => value.Value.ToString()))
+                        };
+
+                        await executor.TryExecuteAsync(new TriggeredFunctionData() { TriggerValue = triggerValue }, cancellationToken);
+                    };
+
+                    RedisValue[] entryIds = streams[i].Entries.Select(entry => entry.Id).ToArray();
+                    await db.StreamAcknowledgeAsync(streams[i].Key, consumerGroup, entryIds);
+
+                    if (deleteAfterProcess)
+                    {
+                        await db.StreamDeleteAsync(streams[i].Key, entryIds);
+                    }
+                }
+            };
+        }
+
+        public override void BeforeClosing()
+        {
+            IDatabase db = multiplexer.GetDatabase();
+            foreach (RedisKey key in keys)
+            {
+                db.StreamDeleteConsumerAsync(key, consumerGroup, consumerName);
+            }
+        }
+
+        public override Task<RedisPollingMetrics> GetMetricsAsync()
+        {
+            var metrics = new RedisPollingMetrics
+            {
+                Remaining = keys.Sum((key) => multiplexer.GetDatabase().StreamLength(key)),
+                Timestamp = DateTime.UtcNow,
+            };
+
+            return Task.FromResult(metrics);
+        }
+    }
+}
diff --git a/src/StreamsTrigger/RedisStreamsTriggerAttribute.cs b/src/StreamsTrigger/RedisStreamsTriggerAttribute.cs
@@ -0,0 +1,24 @@
+﻿using System;
+using Microsoft.Azure.WebJobs.Description;
+
+namespace Microsoft.Azure.WebJobs.Extensions.Redis
+{
+    /// <summary>
+    /// Streams trigger binding attributes.
+    /// </summary>
+    [Binding]
+    [AttributeUsage(AttributeTargets.Parameter)]
+    public class RedisStreamsTriggerAttribute : RedisPollingTriggerAttributeBase
+    {
+        /// <summary>
+        /// Name of the consumer group to use when reading the streams.
+        /// </summary>
+        public string ConsumerGroup { get; set; } = "AzureFunctionRedisExtension";
+
+        /// <summary>
+        /// If true, the function will delete the stream entries after processing.
+        /// </summary>
+        public bool DeleteAfterProcess { get; set; } = false;
+
+    }
+}
diff --git a/src/StreamsTrigger/RedisStreamsTriggerBinding.cs b/src/StreamsTrigger/RedisStreamsTriggerBinding.cs
@@ -0,0 +1,60 @@
+﻿using System;
+using System.Threading.Tasks;
+using System.Collections.Generic;
+using Microsoft.Azure.WebJobs.Host.Bindings;
+using Microsoft.Azure.WebJobs.Host.Listeners;
+using Microsoft.Azure.WebJobs.Host.Protocols;
+using Microsoft.Azure.WebJobs.Host.Triggers;
+
+namespace Microsoft.Azure.WebJobs.Extensions.Redis
+{
+    /// <summary>
+    /// Trigger Binding, manages and binds context to listener.
+    /// </summary>
+    internal class RedisStreamsTriggerBinding : ITriggerBinding
+    {
+        private readonly string connectionString;
+        private readonly TimeSpan pollingInterval;
+        private readonly int messagesPerWorker;
+        private readonly string keys;
+        private readonly int count;
+        private readonly string consumerGroup;
+        private readonly bool deleteAfterProcess;
+
+        public RedisStreamsTriggerBinding(string connectionString, string keys, TimeSpan pollingInterval, int messagesPerWorker, int count, string consumerGroup, bool deleteAfterProcess)
+        {
+            this.connectionString = connectionString;
+            this.keys = keys;
+            this.pollingInterval = pollingInterval;
+            this.messagesPerWorker = messagesPerWorker;
+            this.count = count;
+            this.consumerGroup = consumerGroup;
+            this.deleteAfterProcess = deleteAfterProcess;
+        }
+
+        public Type TriggerValueType => typeof(RedisMessageModel);
+
+        public IReadOnlyDictionary<string, Type> BindingDataContract => new Dictionary<string, Type>();
+
+        public Task<ITriggerData> BindAsync(object value, ValueBindingContext context)
+        {
+            IReadOnlyDictionary<string, object> bindingData = new Dictionary<string, object>();
+            return Task.FromResult<ITriggerData>(new TriggerData(null, bindingData));
+        }
+
+        public Task<IListener> CreateListenerAsync(ListenerFactoryContext context)
+        {
+            if (context == null)
+            {
+                throw new ArgumentNullException("context");
+            }
+
+            return Task.FromResult<IListener>(new RedisStreamsListener(connectionString, keys, pollingInterval, messagesPerWorker, count, consumerGroup, deleteAfterProcess, context.Executor));
+        }
+
+        public ParameterDescriptor ToParameterDescriptor()
+        {
+            return new ParameterDescriptor();
+        }
+    }
+}
diff --git a/src/StreamsTrigger/RedisStreamsTriggerBindingProvider.cs b/src/StreamsTrigger/RedisStreamsTriggerBindingProvider.cs
@@ -0,0 +1,47 @@
+﻿using System;
+using System.Reflection;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Azure.WebJobs.Host.Triggers;
+
+namespace Microsoft.Azure.WebJobs.Extensions.Redis
+{
+    /// <summary>
+    /// Provides trigger binding, variables configured in local.settings.json are being retrieved here.
+    /// </summary>
+    internal class RedisStreamsTriggerBindingProvider : ITriggerBindingProvider
+    {
+        private readonly IConfiguration configuration;
+
+        public RedisStreamsTriggerBindingProvider(IConfiguration configuration)
+        {
+            this.configuration = configuration;
+        }
+
+        public Task<ITriggerBinding> TryCreateAsync(TriggerBindingProviderContext context)
+        {
+            if (context == null)
+            {
+                throw new ArgumentNullException("context");
+            }
+
+            ParameterInfo parameter = context.Parameter;
+            RedisStreamsTriggerAttribute attribute = parameter.GetCustomAttribute<RedisStreamsTriggerAttribute>(inherit: false);
+
+            if (attribute == null)
+            {
+                return Task.FromResult<ITriggerBinding>(null);
+            }
+
+            string connectionString = RedisUtilities.ResolveString(configuration, attribute.ConnectionString, "ConnectionString");
+            string keys = RedisUtilities.ResolveString(configuration, attribute.Keys, "Keys");
+            int messagesPerWorker = attribute.MessagesPerWorker;
+            int batchSize = attribute.BatchSize;
+            TimeSpan pollingInterval = TimeSpan.FromMilliseconds(attribute.PollingIntervalInMs);
+            string consumerGroup = RedisUtilities.ResolveString(configuration, attribute.ConsumerGroup, "ConsumerGroup");
+            bool deleteAfterProcess = attribute.DeleteAfterProcess;
+
+            return Task.FromResult<ITriggerBinding>(new RedisStreamsTriggerBinding(connectionString, keys, pollingInterval, messagesPerWorker, batchSize, consumerGroup, deleteAfterProcess));
+        }
+    }
+}
diff --git a/test/Integration/IntegrationTestFunctions.cs b/test/Integration/IntegrationTestFunctions.cs
@@ -10,9 +10,14 @@ public static class IntegrationTestFunctions
         public const string pubsubChannel = "testChannel";
         public const string keyspaceChannel = "__keyspace@0__:testKey";
         public const string keyeventChannel = "__keyevent@0__:set";
+        public const string keyeventChannelAll = "__keyevent@0__:*";
+        public const string keyspaceChannelAll = "__keyspace@0__:*";
         public const string all = "*";
         public const string listSingleKey = "listSingleKey";
         public const string listMultipleKeys = "listKey1 listKey2 listKey3";
+        public const string streamSingleKey = "streamSingleKey";
+        public const string streamMultipleKeys = "streamKey1 streamKey2 streamKey3";
+        public const int pollingInterval = 100;
         public const int count = 100;
 
         [FunctionName(nameof(PubSubTrigger_SingleChannel))]
@@ -25,7 +30,7 @@ public static void PubSubTrigger_SingleChannel(
 
         [FunctionName(nameof(PubSubTrigger_MultipleChannels))]
         public static void PubSubTrigger_MultipleChannels(
-            [RedisPubSubTrigger(ConnectionString = connectionString, Channel = pubsubChannel + all)] RedisMessageModel model,
+            [RedisPubSubTrigger(ConnectionString = connectionString, Channel = pubsubChannel + "*")] RedisMessageModel model,
             ILogger logger)
         {
             logger.LogInformation(JsonSerializer.Serialize(model));
@@ -49,15 +54,15 @@ public static void KeySpaceTrigger_SingleKey(
 
         [FunctionName(nameof(KeySpaceTrigger_MultipleKeys))]
         public static void KeySpaceTrigger_MultipleKeys(
-            [RedisPubSubTrigger(ConnectionString = connectionString, Channel = keyspaceChannel + all)] RedisMessageModel model,
+            [RedisPubSubTrigger(ConnectionString = connectionString, Channel = keyspaceChannel + "*")] RedisMessageModel model,
             ILogger logger)
         {
             logger.LogInformation(JsonSerializer.Serialize(model));
         }
 
         [FunctionName(nameof(KeySpaceTrigger_AllKeys))]
         public static void KeySpaceTrigger_AllKeys(
-            [RedisPubSubTrigger(ConnectionString = connectionString, Channel = all)] RedisMessageModel model,
+            [RedisPubSubTrigger(ConnectionString = connectionString, Channel = keyspaceChannelAll)] RedisMessageModel model,
             ILogger logger)
         {
             logger.LogInformation(JsonSerializer.Serialize(model));
@@ -73,23 +78,39 @@ public static void KeyEventTrigger_SingleEvent(
 
         [FunctionName(nameof(KeyEventTrigger_AllEvents))]
         public static void KeyEventTrigger_AllEvents(
-            [RedisPubSubTrigger(ConnectionString = connectionString, Channel = all)] RedisMessageModel model,
+            [RedisPubSubTrigger(ConnectionString = connectionString, Channel = keyeventChannelAll)] RedisMessageModel model,
             ILogger logger)
         {
             logger.LogInformation(JsonSerializer.Serialize(model));
         }
 
         [FunctionName(nameof(ListsTrigger_SingleKey))]
         public static void ListsTrigger_SingleKey(
-            [RedisListsTrigger(ConnectionString = connectionString, Keys = listSingleKey)] RedisMessageModel result,
+            [RedisListsTrigger(ConnectionString = connectionString, Keys = listSingleKey, PollingIntervalInMs = pollingInterval)] RedisMessageModel result,
             ILogger logger)
         {
             logger.LogInformation(JsonSerializer.Serialize(result));
         }
 
         [FunctionName(nameof(ListsTrigger_MultipleKeys))]
         public static void ListsTrigger_MultipleKeys(
-            [RedisListsTrigger(ConnectionString = connectionString, Keys = listMultipleKeys)] RedisMessageModel result,
+            [RedisListsTrigger(ConnectionString = connectionString, Keys = listMultipleKeys, PollingIntervalInMs = pollingInterval)] RedisMessageModel result,
+            ILogger logger)
+        {
+            logger.LogInformation(JsonSerializer.Serialize(result));
+        }
+
+        [FunctionName(nameof(StreamsTrigger_DefaultGroup_SingleKey))]
+        public static void StreamsTrigger_DefaultGroup_SingleKey(
+            [RedisStreamsTrigger(ConnectionString = connectionString, Keys = streamSingleKey, PollingIntervalInMs = pollingInterval)] RedisMessageModel result,
+            ILogger logger)
+        {
+            logger.LogInformation(JsonSerializer.Serialize(result));
+        }
+
+        [FunctionName(nameof(StreamsTrigger_DefaultGroup_MultipleKeys))]
+        public static void StreamsTrigger_DefaultGroup_MultipleKeys(
+            [RedisStreamsTrigger(ConnectionString = connectionString, Keys = streamMultipleKeys, PollingIntervalInMs = pollingInterval)] RedisMessageModel result,
             ILogger logger)
         {
             logger.LogInformation(JsonSerializer.Serialize(result));
diff --git a/test/Integration/IntegrationTestHelpers.cs b/test/Integration/IntegrationTestHelpers.cs
@@ -6,6 +6,7 @@
 using System.Collections.Generic;
 using System.Linq;
 using System.Reflection;
+using System.Collections.Concurrent;
 
 namespace Microsoft.Azure.WebJobs.Extensions.Redis.Tests.Integration
 {
@@ -64,7 +65,7 @@ void functionLoadedHandler(object sender, DataReceivedEventArgs e)
             return functionsProcess;
         }
 
-        internal static DataReceivedEventHandler CounterHandlerCreator(Dictionary<string, int> counts, TaskCompletionSource<bool> functionExecuted)
+        internal static DataReceivedEventHandler CounterHandlerCreator(IDictionary<string, int> counts)
         {
             return (object sender, DataReceivedEventArgs e) =>
             {
@@ -74,11 +75,6 @@ internal static DataReceivedEventHandler CounterHandlerCreator(Dictionary<string
                     {
                         counts[key] -= 1;
                     }
-
-                    if (counts.Values.Sum() == 0)
-                    {
-                        functionExecuted.TrySetResult(true);
-                    }
                 }
             };
         }
diff --git a/test/Integration/RedisTriggerTests.cs b/test/Integration/RedisTriggerTests.cs

Original file line number	Diff line number	Diff line change
`@@ -46,5 +46,22 @@ public static void ListsMultipleTrigger(`
`46`	`46`	`{`
`47`	`47`	`logger.LogInformation(JsonSerializer.Serialize(model));`
`48`	`48`	`}`
	`49`	`+`
	`50`	`+ [FunctionName(nameof(StreamsTrigger))]`
	`51`	`+ public static void StreamsTrigger(`
	`52`	`+ [RedisStreamsTrigger(ConnectionString = localhost, Keys = "streamTest")] RedisMessageModel model,`
	`53`	`+ ILogger logger)`
	`54`	`+ {`
	`55`	`+ logger.LogInformation(JsonSerializer.Serialize(model));`
	`56`	`+ }`
	`57`	`+`
	`58`	`+ [FunctionName(nameof(StreamsMultipleTriggers))]`
	`59`	`+ public static void StreamsMultipleTriggers(`
	`60`	`+ [RedisStreamsTrigger(ConnectionString = localhost, Keys = "streamTest1 streamTest2")] RedisMessageModel model,`
	`61`	`+ ILogger logger)`
	`62`	`+ {`
	`63`	`+ logger.LogInformation(JsonSerializer.Serialize(model));`
	`64`	`+ }`
	`65`	`+`
`49`	`66`	`}`
`50`	`67`	`}`
Original file line number	Diff line number	Diff line change
`@@ -32,11 +32,14 @@ public void Initialize(ExtensionConfigContext context)`
`32`	`32`	`}`
`33`	`33`
`34`	`34`	`#pragma warning disable CS0618`
`35`		`- FluentBindingRule<RedisPubSubTriggerAttribute> rule = context.AddBindingRule<RedisPubSubTriggerAttribute>();`
`36`		`- rule.BindToTrigger<RedisMessageModel>(new RedisPubSubTriggerBindingProvider(configuration));`
	`35`	`+ FluentBindingRule<RedisPubSubTriggerAttribute> pubsubTriggerRule = context.AddBindingRule<RedisPubSubTriggerAttribute>();`
	`36`	`+ pubsubTriggerRule.BindToTrigger<RedisMessageModel>(new RedisPubSubTriggerBindingProvider(configuration));`
`37`	`37`
`38`	`38`	`FluentBindingRule<RedisListsTriggerAttribute> listsTriggerRule = context.AddBindingRule<RedisListsTriggerAttribute>();`
`39`	`39`	`listsTriggerRule.BindToTrigger<RedisMessageModel>(new RedisListsTriggerBindingProvider(configuration));`
	`40`	`+`
	`41`	`+ FluentBindingRule<RedisStreamsTriggerAttribute> streamsTriggerRule = context.AddBindingRule<RedisStreamsTriggerAttribute>();`
	`42`	`+ streamsTriggerRule.BindToTrigger<RedisMessageModel>(new RedisStreamsTriggerBindingProvider(configuration));`
`40`	`43`	`#pragma warning restore CS0618`
`41`	`44`	`}`
`42`	`45`	`}`
Original file line number	Diff line number	Diff line change
`@@ -6,6 +6,7 @@`
`6`	`6`	`using System.Collections.Generic;`
`7`	`7`	`using System.Linq;`
`8`	`8`	`using System.Reflection;`
	`9`	`+using System.Collections.Concurrent;`
`9`	`10`
`10`	`11`	`namespace Microsoft.Azure.WebJobs.Extensions.Redis.Tests.Integration`
`11`	`12`	`{`
`@@ -64,7 +65,7 @@ void functionLoadedHandler(object sender, DataReceivedEventArgs e)`
`64`	`65`	`return functionsProcess;`
`65`	`66`	`}`
`66`	`67`
`67`		`- internal static DataReceivedEventHandler CounterHandlerCreator(Dictionary<string, int> counts, TaskCompletionSource<bool> functionExecuted)`
	`68`	`+ internal static DataReceivedEventHandler CounterHandlerCreator(IDictionary<string, int> counts)`
`68`	`69`	`{`
`69`	`70`	`return (object sender, DataReceivedEventArgs e) =>`
`70`	`71`	`{`
`@@ -74,11 +75,6 @@ internal static DataReceivedEventHandler CounterHandlerCreator(Dictionary<string`
`74`	`75`	`{`
`75`	`76`	`counts[key] -= 1;`
`76`	`77`	`}`
`77`		`-`
`78`		`- if (counts.Values.Sum() == 0)`
`79`		`- {`
`80`		`- functionExecuted.TrySetResult(true);`
`81`		`- }`
`82`	`78`	`}`
`83`	`79`	`};`
`84`	`80`	`}`