Broadcast channel to replace SMS

Orleans.sln

@@ -203,6 +203,8 @@ Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "DistributedTests.Server", "
 EndProject
 Project("{9A19103F-16F7-4668-BE54-9A1E7A4F7556}") = "Orleans.Serialization.SystemTextJson", "src\Orleans.Serialization.SystemTextJson\Orleans.Serialization.SystemTextJson.csproj", "{5CFBC7AC-C9AE-4C6C-943C-4A157396E427}"
 EndProject
+Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Orleans.BroadcastChannel", "src\Orleans.BroadcastChannel\Orleans.BroadcastChannel.csproj", "{497D472A-0BA8-4306-A110-C4D871FD5918}"
+EndProject
 Global
 	GlobalSection(SolutionConfigurationPlatforms) = preSolution
 		Debug|Any CPU = Debug|Any CPU
@@ -545,6 +547,10 @@ Global
 		{5CFBC7AC-C9AE-4C6C-943C-4A157396E427}.Debug|Any CPU.Build.0 = Debug|Any CPU
 		{5CFBC7AC-C9AE-4C6C-943C-4A157396E427}.Release|Any CPU.ActiveCfg = Release|Any CPU
 		{5CFBC7AC-C9AE-4C6C-943C-4A157396E427}.Release|Any CPU.Build.0 = Release|Any CPU
+		{497D472A-0BA8-4306-A110-C4D871FD5918}.Debug|Any CPU.ActiveCfg = Debug|Any CPU
+		{497D472A-0BA8-4306-A110-C4D871FD5918}.Debug|Any CPU.Build.0 = Debug|Any CPU
+		{497D472A-0BA8-4306-A110-C4D871FD5918}.Release|Any CPU.ActiveCfg = Release|Any CPU
+		{497D472A-0BA8-4306-A110-C4D871FD5918}.Release|Any CPU.Build.0 = Release|Any CPU
 	EndGlobalSection
 	GlobalSection(SolutionProperties) = preSolution
 		HideSolutionNode = FALSE
@@ -646,6 +652,7 @@ Global
 		{25D20278-8901-47CC-AD1D-F3C4BEB845BF} = {FFEC9FEE-FEDF-4510-B7D2-0B0B3374ED2F}
 		{E8335DC9-9A7F-45C1-AFA3-0AA93ABD4FA5} = {FFEC9FEE-FEDF-4510-B7D2-0B0B3374ED2F}
 		{5CFBC7AC-C9AE-4C6C-943C-4A157396E427} = {4CD3AA9E-D937-48CA-BB6C-158E12257D23}
+		{497D472A-0BA8-4306-A110-C4D871FD5918} = {4CD3AA9E-D937-48CA-BB6C-158E12257D23}
 	EndGlobalSection
 	GlobalSection(ExtensibilityGlobals) = postSolution
 		SolutionGuid = {7BFB3429-B5BB-4DB1-95B4-67D77A864952}

src/Orleans.BroadcastChannel/BroadcastChannelConsumerExtension.cs

@@ -0,0 +1,104 @@
+using System;
+using System.Collections.Concurrent;
+using System.Threading.Tasks;
+using Orleans.Runtime;
+
+namespace Orleans.BroadcastChannel
+{
+    internal interface IBroadcastChannelConsumerExtension : IGrainExtension
+    {
+        Task OnError(InternalChannelId streamId, Exception exception);
+        Task OnPublished(InternalChannelId streamId, object item);
+    }
+
+    internal class BroadcastChannelConsumerExtension : IBroadcastChannelConsumerExtension
+    {
+        private readonly ConcurrentDictionary<InternalChannelId, ICallback> _handlers = new();
+        private readonly IOnBroadcastChannelSubscribed _subscriptionObserver;
+        private AsyncLock _lock = new AsyncLock();
+
+        private interface ICallback
+        {
+            Task OnError(Exception exception);
+
+            Task OnPublished(object item);
+        }
+
+        private class Callback<T> : ICallback
+        {
+            private readonly Func<T, Task> _onPublished;
+            private readonly Func<Exception, Task> _onError;
+
+            private static Task NoOp(Exception _) => Task.CompletedTask;
+
+            public Callback(Func<T, Task> onPublished, Func<Exception, Task> onError)
+            {
+                _onPublished = onPublished;
+                _onError = onError ?? NoOp;
+            }
+
+            public Task OnError(Exception exception) => _onError(exception);
+
+            public Task OnPublished(object item)
+            {
+                return item is T typedItem
+                    ? _onPublished(typedItem)
+                    : _onError(new InvalidCastException($"Received an item of type {item.GetType().Name}, expected {typeof(T).FullName}"));
+            }
+        }
+
+        public BroadcastChannelConsumerExtension(IGrainContextAccessor grainContextAccessor)
+        {
+            _subscriptionObserver = grainContextAccessor.GrainContext?.GrainInstance as IOnBroadcastChannelSubscribed;
+            if (_subscriptionObserver == null)
+            {
+                throw new ArgumentException($"The grain doesn't implement interface {nameof(IOnBroadcastChannelSubscribed)}");
+            }
+        }
+
+        public async Task OnError(InternalChannelId streamId, Exception exception)
+        {
+            var callback = await GetStreamCallback(streamId);
+            if (callback != default)
+            {
+                await callback.OnError(exception);
+            }
+        }
+
+        public async Task OnPublished(InternalChannelId streamId, object item)
+        {
+            var callback = await GetStreamCallback(streamId);
+            if (callback != default)
+            {
+                await callback.OnPublished(item);
+            }
+        }
+
+        public void Attach<T>(InternalChannelId streamId, Func<T, Task> onPublished, Func<Exception, Task> onError)
+        {
+            _handlers.TryAdd(streamId, new Callback<T>(onPublished, onError));
+        }
+
+        private async ValueTask<ICallback> GetStreamCallback(InternalChannelId streamId)
+        {
+            ICallback callback;
+            if (_handlers.TryGetValue(streamId, out callback))
+            {
+                return callback;
+            }
+            using (await _lock.LockAsync())
+            {
+                if (_handlers.TryGetValue(streamId, out callback))
+                {
+                    return callback;
+                }
+                // Give a chance to the grain to attach a handler for this streamId
+                var subscription = new BroadcastChannelSubscription(this, streamId);
+                await _subscriptionObserver.OnSubscribed(subscription);
+            }
+            _handlers.TryGetValue(streamId, out callback);
+            return callback;
+        }
+    }
+}
+

src/Orleans.BroadcastChannel/BroadcastChannelOptions.cs

@@ -0,0 +1,8 @@
+namespace Orleans.BroadcastChannel
+{
+    public class BroadcastChannelOptions
+    {
+        public bool FireAndForgetDelivery { get; set; } = true;
+    }
+}
+

src/Orleans.BroadcastChannel/BroadcastChannelProvider.cs

@@ -0,0 +1,55 @@
+using System;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+using Orleans.BroadcastChannel.SubscriberTable;
+using Orleans.Providers;
+using Orleans.Runtime;
+
+namespace Orleans.BroadcastChannel
+{
+    public interface IBroadcastChannelProvider
+    {
+        IBroadcastChannelWriter<T> GetChannelWriter<T>(ChannelId streamId);
+    }
+
+    internal class BroadcastChannelProvider : IBroadcastChannelProvider
+    {
+        private readonly string _providerName;
+        private readonly BroadcastChannelOptions _options;
+        private readonly IGrainFactory _grainFactory;
+        private readonly ImplicitChannelSubscriberTable _subscriberTable;
+        private readonly ILoggerFactory _loggerFactory;
+
+        public BroadcastChannelProvider(
+            string providerName,
+            BroadcastChannelOptions options,
+            IGrainFactory grainFactory,
+            ImplicitChannelSubscriberTable subscriberTable,
+            ILoggerFactory loggerFactory)
+        {
+            _providerName = providerName;
+            _options = options;
+            _grainFactory = grainFactory;
+            _subscriberTable = subscriberTable;
+            _loggerFactory = loggerFactory;
+        }
+
+        public IBroadcastChannelWriter<T> GetChannelWriter<T>(ChannelId streamId)
+        {
+            return new BroadcastChannelWriter<T>(
+                new InternalChannelId(_providerName, streamId),
+                _grainFactory,
+                _subscriberTable,
+                _options.FireAndForgetDelivery,
+                _loggerFactory);
+        }
+
+        public static IBroadcastChannelProvider Create(IServiceProvider sp, string name)
+        {
+            var opt = sp.GetOptionsByName<BroadcastChannelOptions>(name);
+            return ActivatorUtilities.CreateInstance<BroadcastChannelProvider>(sp, name, sp.GetOptionsByName<BroadcastChannelOptions>(name));
+        }
+    }
+}
+

src/Orleans.BroadcastChannel/BroadcastChannelSubscription.cs

@@ -0,0 +1,43 @@
+using System;
+using System.Threading.Tasks;
+using Orleans.Runtime;
+
+namespace Orleans.BroadcastChannel
+{
+    public interface IBroadcastChannelSubscription
+    {
+        public ChannelId ChannelId { get; }
+
+        public string ProviderName { get; }
+
+        Task Attach<T>(Func<T, Task> onPublished, Func<Exception, Task> onError = null);
+    }
+
+    public interface IOnBroadcastChannelSubscribed
+    {
+        public Task OnSubscribed(IBroadcastChannelSubscription streamSubscription);
+    }
+
+    internal class BroadcastChannelSubscription : IBroadcastChannelSubscription
+    {
+        private readonly BroadcastChannelConsumerExtension _consumerExtension;
+        private readonly InternalChannelId _streamId;
+
+        public ChannelId ChannelId => _streamId.ChannelId;
+
+        public string ProviderName => _streamId.ProviderName;
+
+        public BroadcastChannelSubscription(BroadcastChannelConsumerExtension consumerExtension, InternalChannelId streamId)
+        {
+            _consumerExtension = consumerExtension;
+            _streamId = streamId;
+        }
+
+        public Task Attach<T>(Func<T, Task> onPublished, Func<Exception, Task> onError = null)
+        {
+            _consumerExtension.Attach(_streamId, onPublished, onError);
+            return Task.CompletedTask;
+        }
+    }
+}
+

src/Orleans.BroadcastChannel/BroadcastChannelWriter.cs

@@ -0,0 +1,95 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using Microsoft.Extensions.Logging;
+using Orleans.BroadcastChannel.SubscriberTable;
+using Orleans.Providers;
+using Orleans.Runtime;
+
+namespace Orleans.BroadcastChannel
+{
+    public interface IBroadcastChannelWriter<T>
+    {
+        Task Publish(T item);
+    }
+
+    internal class BroadcastChannelWriter<T> : IBroadcastChannelWriter<T>
+    {
+        private static readonly string LoggingCategory = typeof(BroadcastChannelWriter<>).FullName;
+
+        private readonly InternalChannelId _channelId;
+        private readonly IGrainFactory _grainFactory;
+        private readonly ImplicitChannelSubscriberTable _subscriberTable;
+        private readonly bool _fireAndForgetDelivery;
+        private readonly ILogger _logger;
+
+        public BroadcastChannelWriter(
+            InternalChannelId channelId,
+            IGrainFactory grainFactory,
+            ImplicitChannelSubscriberTable subscriberTable,
+            bool fireAndForgetDelivery,
+            ILoggerFactory loggerFactory)
+        {
+            _channelId = channelId;
+            _grainFactory = grainFactory;
+            _subscriberTable = subscriberTable;
+            _fireAndForgetDelivery = fireAndForgetDelivery;
+            _logger = loggerFactory.CreateLogger(LoggingCategory);
+        }
+
+        public async Task Publish(T item)
+        {
+            var subscribers = _subscriberTable.GetImplicitSubscribers(_channelId, _grainFactory);
+
+            if (subscribers.Count == 0)
+            {
+                if (_logger.IsEnabled(LogLevel.Debug)) _logger.LogDebug("No consumer found for {Item}", item);
+                return;
+            }
+
+            if (_logger.IsEnabled(LogLevel.Debug)) _logger.LogDebug("Publishing item {Item} to {ConsumerCount} consumers", item, subscribers.Count);
+
+            if (_fireAndForgetDelivery)
+            {
+                foreach (var sub in subscribers)
+                {
+                    PublishToSubscriber(sub.Value, item).Ignore();
+                }
+            }
+            else
+            {
+                var tasks = new List<Task>();
+                foreach (var sub in subscribers)
+                {
+                    tasks.Add(PublishToSubscriber(sub.Value, item));
+                }
+                try
+                {
+                    await Task.WhenAll(tasks);
+                }
+                catch (Exception)
+                {
+                    throw new AggregateException(tasks.Select(t => t.Exception).Where(ex => ex != null));
+                }
+            }
+        }
+
+        private async Task PublishToSubscriber(IBroadcastChannelConsumerExtension consumer, T item)
+        {
+            try
+            {
+                await consumer.OnPublished(_channelId, item);
+            }
+            catch (Exception ex)
+            {
+                _logger.LogError(ex, "Exception when sending item to {GrainId}", consumer.GetGrainId());
+                if (!_fireAndForgetDelivery)
+                {
+                    throw;
+                }
+            }
+        }
+    }
+}
+