Updated MartenEventPublisher to publish events with metadata.

Core.ElasticSearch/Projections/ElasticSearchProjection.cs

@@ -9,7 +9,7 @@ namespace Core.ElasticSearch.Projections;
 
 public class ElasticSearchProjection<TEvent, TView> : IEventHandler<EventEnvelope<TEvent>>
     where TView : class, IProjection
-    where TEvent : IEvent
+    where TEvent : notnull
 {
     private readonly IElasticClient elasticClient;
     private readonly Func<TEvent, string> getId;
@@ -46,7 +46,7 @@ public static class ElasticSearchProjectionConfig
     public static IServiceCollection Project<TEvent, TView>(this IServiceCollection services,
         Func<TEvent, string> getId)
         where TView : class, IProjection
-        where TEvent : IEvent
+        where TEvent : notnull
     {
         services.AddTransient<IEventHandler<EventEnvelope<TEvent>>>(sp =>
         {

Core.ElasticSearch/Repository/ElasticSearchRepository.cs

@@ -1,5 +1,4 @@
 using Core.ElasticSearch.Indices;
-using Core.Events;
 using Nest;
 using IAggregate = Core.Aggregates.IAggregate;
 

Core.EventStoreDB.Tests/Events/EventStoreDBEventMetadataJsonConverterTests.cs

@@ -1,5 +1,4 @@
-using Core.Events;
-using Core.EventStoreDB.Events;
+using Core.EventStoreDB.Events;
 using Core.Tracing;
 using Core.Tracing.Causation;
 using Core.Tracing.Correlation;

Core.EventStoreDB/Events/EventStoreDBEventMetadataJsonConverter.cs

@@ -1,5 +1,4 @@
-using Core.Events;
-using Core.Tracing;
+using Core.Tracing;
 using Core.Tracing.Causation;
 using Core.Tracing.Correlation;
 using Newtonsoft.Json;

Core.EventStoreDB/Subscriptions/EventStoreDBSubscriptionCheckpointRepository.cs

@@ -1,10 +1,9 @@
-using Core.Events;
-using Core.EventStoreDB.Serialization;
+using Core.EventStoreDB.Serialization;
 using EventStore.Client;
 
 namespace Core.EventStoreDB.Subscriptions;
 
-public record CheckpointStored(string SubscriptionId, ulong? Position, DateTime CheckpointedAt): IEvent;
+public record CheckpointStored(string SubscriptionId, ulong? Position, DateTime CheckpointedAt);
 
 public class EventStoreDBSubscriptionCheckpointRepository: ISubscriptionCheckpointRepository
 {

Core.Kafka/Config.cs

@@ -1,4 +1,5 @@
 using Core.BackgroundWorkers;
+using Core.Events;
 using Core.Events.External;
 using Core.Kafka.Consumers;
 using Core.Kafka.Producers;
@@ -13,7 +14,10 @@ public static class Config
     public static IServiceCollection AddKafkaProducer(this IServiceCollection services)
     {
         //using TryAdd to support mocking, without that it won't be possible to override in tests
-        services.TryAddScoped<IExternalEventProducer, KafkaProducer>();
+        services.TryAddSingleton<IExternalEventProducer, KafkaProducer>();
+        services.AddSingleton<IEventBus>(sp =>
+            new EventBusDecoratorWithExternalProducer(sp.GetRequiredService<EventBus>(),
+                sp.GetRequiredService<IExternalEventProducer>()));
         return services;
     }
 

Core.Kafka/Consumers/KafkaConsumer.cs

@@ -1,9 +1,7 @@
 using Confluent.Kafka;
-using Core.Events;
 using Core.Events.External;
 using Core.Reflection;
 using Microsoft.Extensions.Configuration;
-using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Logging;
 using Newtonsoft.Json;
 using IEventBus = Core.Events.IEventBus;
@@ -12,18 +10,18 @@ namespace Core.Kafka.Consumers;
 
 public class KafkaConsumer: IExternalEventConsumer
 {
-    private readonly IServiceProvider serviceProvider;
     private readonly ILogger<KafkaConsumer> logger;
+    private readonly IEventBus eventBus;
     private readonly KafkaConsumerConfig config;
 
     public KafkaConsumer(
-        IServiceProvider serviceProvider,
         ILogger<KafkaConsumer> logger,
-        IConfiguration configuration
+        IConfiguration configuration,
+        IEventBus eventBus
     )
     {
-        this.serviceProvider = serviceProvider ?? throw new ArgumentNullException(nameof(serviceProvider));
         this.logger = logger ?? throw new ArgumentNullException(nameof(logger));
+        this.eventBus = eventBus;
 
         if (configuration == null)
             throw new ArgumentNullException(nameof(configuration));
@@ -73,12 +71,8 @@ private async Task ConsumeNextEvent(IConsumer<string, string> consumer, Cancella
             // deserialize event
             var @event = JsonConvert.DeserializeObject(message.Message.Value, eventType)!;
 
-            using var scope = serviceProvider.CreateScope();
-            var eventBus =
-                scope.ServiceProvider.GetRequiredService<IEventBus>();
-
             // publish event to internal event bus
-            await eventBus.Publish((IEvent)@event, cancellationToken);
+            await eventBus.Publish(@event, cancellationToken);
         }
         catch (Exception e)
         {

Core.Kafka/Producers/KafkaProducer.cs

@@ -1,8 +1,9 @@
 using Confluent.Kafka;
 using Core.Events;
 using Core.Events.External;
+using Core.Serialization.Newtonsoft;
+using Core.Threading;
 using Microsoft.Extensions.Configuration;
-using Newtonsoft.Json;
 
 namespace Core.Kafka.Producers;
 
@@ -21,18 +22,46 @@ IConfiguration configuration
         config = configuration.GetKafkaProducerConfig();
     }
 
-    public async Task Publish(IExternalEvent @event, CancellationToken cancellationToken)
+    public async Task Publish(EventEnvelope @event, CancellationToken ct)
     {
-        using var p = new ProducerBuilder<string, string>(config.ProducerConfig).Build();
-        await Task.Yield();
-        // publish event to kafka topic taken from config
-        await p.ProduceAsync(config.Topic,
-            new Message<string, string>
-            {
-                // store event type name in message Key
-                Key = @event.GetType().Name,
-                // serialize event to message Value
-                Value = JsonConvert.SerializeObject(@event)
-            }, cancellationToken);
+        try
+        {
+            using var p = new ProducerBuilder<string, string>(config.ProducerConfig).Build();
+            await Task.Yield();
+            // publish event to kafka topic taken from config
+
+            // var result = p.ProduceAsync(config.Topic,
+            //     new Message<string, string>
+            //     {
+            //         // store event type name in message Key
+            //         Key = @event.GetType().Name,
+            //         // serialize event to message Value
+            //         Value = @event.ToJson()
+            //     }, ct).ConfigureAwait(false).GetAwaiter().GetResult();
+
+            p.Produce(config.Topic,
+                new Message<string, string>
+                {
+                    // store event type name in message Key
+                    Key = @event.GetType().Name,
+                    // serialize event to message Value
+                    Value = @event.ToJson()
+                });
+            p.Flush(ct);
+            // Console.WriteLine(result);
+            // var result = p.ProduceAsync(config.Topic,
+            //     new Message<string, string>
+            //     {
+            //         // store event type name in message Key
+            //         Key = @event.GetType().Name,
+            //         // serialize event to message Value
+            //         Value = @event.ToJson()
+            //     }, ct).ConfigureAwait(false).GetAwaiter().GetResult();
+        }
+        catch (Exception e)
+        {
+            Console.WriteLine(e);
+            throw;
+        }
     }
 }

Core.Marten/Subscriptions/MartenEventPublisher.cs

@@ -1,8 +1,10 @@
 using Core.Events;
+using Core.Tracing;
+using Core.Tracing.Causation;
+using Core.Tracing.Correlation;
 using Marten;
 using Marten.Events;
 using Microsoft.Extensions.DependencyInjection;
-using IEvent = Core.Events.IEvent;
 
 namespace Core.Marten.Subscriptions;
 
@@ -34,9 +36,17 @@ public async Task ConsumeAsync(IDocumentOperations documentOperations, IReadOnly
             using var scope = serviceProvider.CreateScope();
             var eventBus = scope.ServiceProvider.GetRequiredService<IEventBus>();
 
-            if (@event.Data is not IEvent mappedEvent) continue;
+            var eventMetadata = new EventMetadata(
+                @event.Id.ToString(),
+                (ulong)@event.Version,
+                (ulong)@event.Sequence,
+                new TraceMetadata(
+                    @event.CorrelationId != null ? new CorrelationId(@event.CorrelationId) : null,
+                    @event.CausationId != null ? new CausationId(@event.CausationId) : null
+                )
+            );
 
-            await eventBus.Publish(mappedEvent, ct);
+            await eventBus.Publish(new EventEnvelope(@event.Data, eventMetadata), ct);
         }
     }
 }

Core.Serialization/Newtonsoft/SerializationExtensions.cs

@@ -53,7 +53,8 @@ public static object FromJson(this string json, Type type)
     /// <returns>json string</returns>
     public static string ToJson(this object obj)
     {
-        return JsonConvert.SerializeObject(obj);
+        return JsonConvert.SerializeObject(obj,
+            new JsonSerializerSettings().WithNonDefaultConstructorContractResolver());
     }
 
     /// <summary>

Core.Testing/AggregateExtensions.cs

@@ -5,7 +5,7 @@ namespace Core.Testing;
 
 public static class AggregateExtensions
 {
-    public static T? PublishedEvent<T>(this IAggregate aggregate) where T : class, IEvent
+    public static T? PublishedEvent<T>(this IAggregate aggregate) where T : class
     {
         return aggregate.DequeueUncommittedEvents().LastOrDefault() as T;
     }

Core.Testing/ApiFixture.cs

@@ -22,13 +22,14 @@ public override TestContext CreateTestContext() =>
             services.AddSingleton(eventsLog);
             services.AddSingleton(typeof(IEventHandler<>), typeof(EventListener<>));
             services.AddSingleton<IExternalEventProducer>(externalEventProducer);
+            services.AddSingleton<IEventBus>(sp =>
+                new EventBusDecoratorWithExternalProducer(sp.GetRequiredService<EventBus>(),
+                    sp.GetRequiredService<IExternalEventProducer>()));
             services.AddSingleton<IExternalCommandBus>(externalCommandBus);
             services.AddSingleton<IExternalEventConsumer, DummyExternalEventConsumer>();
-
         }, SetupWebHostBuilder);
 
 
-
     public IReadOnlyCollection<TEvent> PublishedExternalEventsOfType<TEvent>() where TEvent : IExternalEvent
     {
         return externalEventProducer.PublishedEvents.OfType<TEvent>().ToList();
@@ -39,7 +40,7 @@ public IReadOnlyCollection<TCommand> PublishedExternalCommandOfType<TCommand>()
         return externalCommandBus.SentCommands.OfType<TCommand>().ToList();
     }
 
-    public async Task PublishInternalEvent(IEvent @event, CancellationToken ct = default)
+    public async Task PublishInternalEvent(object @event, CancellationToken ct = default)
     {
         using var scope = Server.Host.Services.CreateScope();
         var eventBus = scope.ServiceProvider.GetRequiredService<IEventBus>();
@@ -78,8 +79,6 @@ public async Task ShouldPublishInternalEventOfType<TEvent>(
             retryCount--;
         } while (!finished);
     }
-
-
 }
 
 public abstract class ApiWithEventsFixture: ApiFixture

Core.Testing/DummyExternalEventPublisher.cs

@@ -5,11 +5,11 @@ namespace Core.Testing;
 
 public class DummyExternalEventProducer: IExternalEventProducer
 {
-    public IList<IExternalEvent> PublishedEvents { get; } = new List<IExternalEvent>();
+    public IList<object> PublishedEvents { get; } = new List<object>();
 
-    public Task Publish(IExternalEvent @event, CancellationToken ct)
+    public Task Publish(EventEnvelope @event, CancellationToken ct)
     {
-        PublishedEvents.Add(@event);
+        PublishedEvents.Add(@event.Data);
 
         return Task.CompletedTask;
     }

Core.Testing/EventListener.cs

@@ -4,11 +4,11 @@ namespace Core.Testing;
 
 public class EventsLog
 {
-    public List<IEvent> PublishedEvents { get; } = new();
+    public List<object> PublishedEvents { get; } = new();
 }
 
 public class EventListener<TEvent>: IEventHandler<TEvent>
-    where TEvent : IEvent
+    where TEvent : notnull
 {
     private readonly EventsLog eventsLog;
 

Core.WebApi/Tracing/TracingMiddleware.cs

@@ -1,5 +1,4 @@
-using Core.Events;
-using Core.Tracing;
+using Core.Tracing;
 using Microsoft.AspNetCore.Builder;
 using Microsoft.AspNetCore.Http;
 using Microsoft.Extensions.DependencyInjection;

Core/Aggregates/Aggregate.cs

@@ -1,5 +1,3 @@
-using Core.Events;
-
 namespace Core.Aggregates;
 
 public abstract class Aggregate: Aggregate<Guid>, IAggregate
@@ -12,11 +10,11 @@ public abstract class Aggregate<T>: IAggregate<T> where T : notnull
 
     public int Version { get; protected set; }
 
-    [NonSerialized] private readonly Queue<IEvent> uncommittedEvents = new();
+    [NonSerialized] private readonly Queue<object> uncommittedEvents = new();
 
     public virtual void When(object @event) { }
 
-    public IEvent[] DequeueUncommittedEvents()
+    public object[] DequeueUncommittedEvents()
     {
         var dequeuedEvents = uncommittedEvents.ToArray();
 
@@ -25,7 +23,7 @@ public IEvent[] DequeueUncommittedEvents()
         return dequeuedEvents;
     }
 
-    protected void Enqueue(IEvent @event)
+    protected void Enqueue(object @event)
     {
         uncommittedEvents.Enqueue(@event);
     }

Core/Aggregates/IAggregate.cs

@@ -1,4 +1,3 @@
-using Core.Events;
 using Core.Projections;
 
 namespace Core.Aggregates;
@@ -12,5 +11,5 @@ public interface IAggregate<out T>: IProjection
     T Id { get; }
     int Version { get; }
 
-    IEvent[] DequeueUncommittedEvents();
+    object[] DequeueUncommittedEvents();
 }

Core/Config.cs

@@ -1,6 +1,5 @@
 using Core.Commands;
 using Core.Events;
-using Core.Events.External;
 using Core.Ids;
 using Core.Queries;
 using Core.Requests;
@@ -23,15 +22,13 @@ public static IServiceCollection AddCoreServices(this IServiceCollection service
             .AddTracing()
             .AddEventBus();
 
-        services.TryAddScoped<IExternalEventProducer, NulloExternalEventProducer>();
         services.TryAddScoped<IExternalCommandBus, ExternalCommandBus>();
 
         services.TryAddScoped<IIdGenerator, NulloIdGenerator>();
 
         return services;
     }
 
-
     public static IServiceCollection AddTracing(this IServiceCollection services)
     {
         services.TryAddSingleton<ICorrelationIdFactory, GuidCorrelationIdFactory>();

Core/Core.csproj

@@ -12,6 +12,7 @@
         <PackageReference Include="Newtonsoft.Json" Version="13.0.1" />
         <PackageReference Include="Polly" Version="7.2.3" />
         <PackageReference Include="RestSharp" Version="107.3.0" />
+        <PackageReference Include="Scrutor" Version="3.3.0" />
     </ItemGroup>
 
     <ItemGroup>

Core/Events/Config.cs

@@ -7,7 +7,6 @@ public static class Config
     public static IServiceCollection AddEventHandler<TEvent, TEventHandler>(
         this IServiceCollection services
     )
-        where TEvent : IEvent
         where TEventHandler : class, IEventHandler<TEvent>
     {
         return services