add message broker (dapr provider) #5

Author: thangchung

coolstore.sln.DotSettings

@@ -8,4 +8,5 @@
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=exprs/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=Grpc/@EntryIndexedValue">True</s:Boolean>
 	<s:Boolean x:Key="/Default/UserDictionary/Words/=Mediat/@EntryIndexedValue">True</s:Boolean>
-	<s:Boolean x:Key="/Default/UserDictionary/Words/=Migrator/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>
+	<s:Boolean x:Key="/Default/UserDictionary/Words/=Migrator/@EntryIndexedValue">True</s:Boolean>
+	<s:Boolean x:Key="/Default/UserDictionary/Words/=pubsub/@EntryIndexedValue">True</s:Boolean></wpf:ResourceDictionary>

src/BuildingBlocks/CoolStore.AppContracts/CoolStore.AppContracts.csproj

@@ -12,8 +12,6 @@
     <PackageReference Include="System.Diagnostics.DiagnosticSource" Version="5.0.1" />
     <PackageReference Include="AspNetCore.HealthChecks.NpgSql" Version="5.0.1" />
     <PackageReference Include="AspNetCore.HealthChecks.Uris" Version="5.0.1" />
-    <PackageReference Include="RestEase" Version="1.5.2" />
-    <PackageReference Include="RestEase.HttpClientFactory" Version="1.5.2" />
     <PackageReference Include="RestEase.SourceGenerator" Version="1.5.2">
       <PrivateAssets>all</PrivateAssets>
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>

src/BuildingBlocks/CoolStore.AppContracts/IntegrationEvents/CustomerCreatedIntegrationEvent.cs

@@ -0,0 +1,13 @@
+using N8T.Core.Domain;
+using N8T.Infrastructure.Bus.Dapr;
+
+namespace CoolStore.AppContracts.IntegrationEvents
+{
+    [DaprPubSubName(PubSubName = "pubsub")]
+    public class CustomerCreatedIntegrationEvent : IntegrationEventBase
+    {
+        public override void Flatten()
+        {
+        }
+    }
+}

src/BuildingBlocks/N8T.Core/Domain/DomainEventBase.cs

@@ -17,7 +17,7 @@ public interface IDomainEventContext
 
     public abstract class EventBase : IDomainEvent
     {
-        public string EventType { get { return GetType().Name; } }
+        public string EventType { get { return GetType().AssemblyQualifiedName; } }
         public DateTime CreatedAt { get; } = DateTime.UtcNow;
         public string CorrelationId { get; init; }
         public IDictionary<string, object> MetaData { get; } = new Dictionary<string, object>();

src/BuildingBlocks/N8T.Infrastructure/Bus/Dapr/DaprEventBus.cs

@@ -0,0 +1,57 @@
+using System;
+using System.Threading;
+using System.Threading.Tasks;
+using Dapr.Client;
+using Microsoft.Extensions.Logging;
+using N8T.Core.Domain;
+
+namespace N8T.Infrastructure.Bus.Dapr
+{
+    public class DaprEventBus : IEventBus
+    {
+        private readonly DaprClient _daprClient;
+        private readonly ILogger<DaprEventBus> _logger;
+
+        public DaprEventBus(DaprClient daprClient, ILogger<DaprEventBus> logger)
+        {
+            _daprClient = daprClient ?? throw new ArgumentNullException(nameof(daprClient));
+            _logger = logger ?? throw new ArgumentNullException(nameof(logger));
+        }
+
+        public async Task PublishAsync<TEvent>(TEvent @event, string[] topics = default,
+            CancellationToken token = default) where TEvent : IntegrationEventBase
+        {
+            var attr = (DaprPubSubNameAttribute)Attribute.GetCustomAttribute(typeof(TEvent),
+                typeof(DaprPubSubNameAttribute));
+
+            var pubsubName = "pubsub";
+
+            if (attr is not null)
+            {
+                pubsubName = attr.PubSubName;
+            }
+
+            if (topics is null)
+            {
+                var topicName = @event.GetType().Name;
+
+                _logger.LogInformation("Publishing event {@Event} to {PubsubName}.{TopicName}", @event, pubsubName, topicName);
+                await _daprClient.PublishEventAsync(pubsubName, topicName, @event, token);
+            }
+            else
+            {
+                foreach (var topicName in topics)
+                {
+                    _logger.LogInformation("Publishing event {@Event} to {PubsubName}.{TopicName}", @event, pubsubName,
+                        topicName);
+                    await _daprClient.PublishEventAsync(pubsubName, topicName, @event, token);
+                }
+            }
+        }
+
+        public Task SubscribeAsync<TMessage>(string[] topics = default, CancellationToken token = default) where TMessage : IntegrationEventBase
+        {
+            throw new System.NotImplementedException();
+        }
+    }
+}

src/BuildingBlocks/N8T.Infrastructure/Bus/Dapr/DaprPubSubNameAttribute.cs

@@ -0,0 +1,9 @@
+using System;
+
+namespace N8T.Infrastructure.Bus.Dapr
+{
+    public class DaprPubSubNameAttribute : Attribute
+    {
+        public string PubSubName { get; set; } = "pubsub";
+    }
+}

src/BuildingBlocks/N8T.Infrastructure/Bus/Extensions.cs

@@ -0,0 +1,22 @@
+using Microsoft.Extensions.DependencyInjection;
+using N8T.Infrastructure.Bus.Dapr;
+
+namespace N8T.Infrastructure.Bus
+{
+    public static class Extensions
+    {
+        public static IServiceCollection AddMessageBroker(this IMvcBuilder mvcBuilder,
+            string messageBrokerType = "dapr")
+        {
+            switch (messageBrokerType)
+            {
+                case "dapr":
+                    mvcBuilder.AddDapr();
+                    mvcBuilder.Services.AddScoped<IEventBus, DaprEventBus>();
+                    break;
+            }
+
+            return mvcBuilder.Services;
+        }
+    }
+}

src/BuildingBlocks/N8T.Infrastructure/Bus/IEventBus.cs

@@ -0,0 +1,15 @@
+using System.Threading;
+using System.Threading.Tasks;
+using N8T.Core.Domain;
+
+namespace N8T.Infrastructure.Bus
+{
+    public interface IEventBus
+    {
+        Task PublishAsync<TEvent>(TEvent @event, string[] topics = default, CancellationToken token = default)
+            where TEvent : IntegrationEventBase;
+
+        Task SubscribeAsync<TEvent>(string[] topics = default, CancellationToken token = default)
+            where TEvent : IntegrationEventBase;
+    }
+}

src/BuildingBlocks/N8T.Infrastructure/Dapr/Extensions.cs

@@ -21,6 +21,7 @@ public static class Extensions
         public static IServiceCollection AddCore(this IServiceCollection services, Type[] types = null,
             Action<IServiceCollection> doMoreActions = null)
         {
+            services.AddHttpContextAccessor();
             services.AddCustomMediatR(types);
             services.AddCustomValidators(types);