Fixes #109

src/Core/src/Eventuous.Producers/BaseProducer.cs

@@ -7,8 +7,8 @@ namespace Eventuous.Producers;
 
 public abstract class BaseProducer<TProduceOptions> : BaseProducer, IEventProducer<TProduceOptions>
     where TProduceOptions : class {
-    protected BaseProducer(bool requiresInit, ProducerTracingOptions? tracingOptions = null) 
-        : base(requiresInit, tracingOptions) { }
+    protected BaseProducer(ProducerTracingOptions? tracingOptions = null) 
+        : base(tracingOptions) { }
 
     protected abstract Task ProduceMessages(
         StreamName                   stream,
@@ -47,10 +47,9 @@ protected override Task ProduceMessages(
 public abstract class BaseProducer : IEventProducer {
     protected KeyValuePair<string, object?>[] DefaultTags { get; }
 
-    protected BaseProducer(bool requiresInit, ProducerTracingOptions? tracingOptions = null) {
+    protected BaseProducer(ProducerTracingOptions? tracingOptions = null) {
         var options = tracingOptions ?? new ProducerTracingOptions();
         DefaultTags = options.AllTags.Concat(EventuousDiagnostics.Tags).ToArray();
-        if (!requiresInit) ReadyNow();
     }
 
     protected abstract Task ProduceMessages(
@@ -85,8 +84,4 @@ public async Task Produce(
             return (act?.Start(), new[] { producedMessage });
         }
     }
-
-    public bool Ready { get; private set; }
-
-    protected void ReadyNow() => Ready = true;
 }

src/Core/src/Eventuous.Producers/IEventProducer.cs

@@ -1,3 +1,5 @@
+using Microsoft.Extensions.Hosting;
+
 namespace Eventuous.Producers;
 
 [PublicAPI]
@@ -15,8 +17,6 @@ Task Produce(
         IEnumerable<ProducedMessage> messages,
         CancellationToken            cancellationToken = default
     );
-
-    bool Ready { get; }
 }
 
 [PublicAPI]
@@ -36,4 +36,8 @@ Task Produce(
         TProduceOptions?             options,
         CancellationToken            cancellationToken = default
     );
+}
+
+public interface IHostedProducer : IHostedService {
+    bool Ready { get; }
 }

src/EventStore/src/Eventuous.EventStore/EsdbEventStore.cs

@@ -1,3 +1,6 @@
+// Copyright (C) 2021-2022 Ubiquitous AS. All rights reserved
+// Licensed under the Apache License, Version 2.0.
+
 // ReSharper disable CoVariantArrayConversion
 
 using System.Runtime.CompilerServices;

src/EventStore/src/Eventuous.EventStore/Producers/EventStoreProducer.cs

@@ -22,7 +22,7 @@ public EventStoreProducer(
         EventStoreClient     eventStoreClient,
         IEventSerializer?    serializer     = null,
         IMetadataSerializer? metaSerializer = null
-    ) : base(false, TracingOptions) {
+    ) : base(TracingOptions) {
         _client         = Ensure.NotNull(eventStoreClient);
         _serializer     = serializer     ?? DefaultEventSerializer.Instance;
         _metaSerializer = metaSerializer ?? DefaultMetadataSerializer.Instance;

src/Experimental/src/Eventuous.ElasticSearch/Index/IndexConfig.cs

@@ -1,3 +1,6 @@
+// Copyright (C) 2021-2022 Ubiquitous AS. All rights reserved
+// Licensed under the Apache License, Version 2.0.
+
 // ReSharper disable UnusedAutoPropertyAccessor.Global
 
 namespace Eventuous.ElasticSearch.Index;

src/Experimental/src/Eventuous.ElasticSearch/Producers/ElasticProducer.cs

@@ -11,7 +11,7 @@ namespace Eventuous.ElasticSearch.Producers;
 public class ElasticProducer : BaseProducer<ElasticProduceOptions> {
     readonly IElasticClient _elasticClient;
 
-    public ElasticProducer(IElasticClient elasticClient) : base(false, TracingOptions) => _elasticClient = elasticClient;
+    public ElasticProducer(IElasticClient elasticClient) : base(TracingOptions) => _elasticClient = elasticClient;
 
     static readonly ProducerTracingOptions TracingOptions = new() {
         MessagingSystem  = "elasticsearch",

src/Gateway/src/Eventuous.Gateway/GatewayProducer.cs

@@ -13,11 +13,7 @@ public async Task Produce(
         T?                           options,
         CancellationToken            cancellationToken = default
     ) {
-        while (!_inner.Ready) {
-            EventuousEventSource.Log.Warn("Producer not ready, waiting...");
-            await Task.Delay(1000, cancellationToken);
-        }
-
+        await WaitForInner(_inner, cancellationToken);
         await _inner.Produce(stream, messages, options, cancellationToken);
     }
 }
@@ -32,10 +28,16 @@ public async Task Produce(
         IEnumerable<ProducedMessage> messages,
         CancellationToken            cancellationToken = default
     ) {
-        while (!_inner.Ready) await Task.Delay(10, cancellationToken);
-
+        await WaitForInner(_inner, cancellationToken);
         await _inner.Produce(stream, messages, cancellationToken);
     }
 
-    public bool Ready => true;
+    protected static async ValueTask WaitForInner(IEventProducer inner, CancellationToken cancellationToken) {
+        if (inner is not IHostedProducer hosted) return;
+
+        while (!hosted.Ready) {
+            EventuousEventSource.Log.Warn("Producer not ready, waiting...");
+            await Task.Delay(1000, cancellationToken);
+        }
+    }
 }

src/Gateway/test/Eventuous.Gateway.Tests/RegistrationTests.cs

@@ -66,6 +66,6 @@ protected override Task ProduceMessages(
             return Task.CompletedTask;
         }
 
-        public TestProducer() : base(false) { }
+        public TestProducer() : base() { }
     }
 }

src/Gateway/test/Eventuous.Gateway.Tests/RegistrationTestsWithOptions.cs

@@ -66,7 +66,7 @@ protected override Task ProduceMessages(
             return Task.CompletedTask;
         }
 
-        public TestProducer() : base(false) { }
+        public TestProducer() : base() { }
     }
 
     record TestProduceOptions { }

src/GooglePubSub/src/Eventuous.GooglePubSub/Producers/GooglePubSubProducer.cs

@@ -1,7 +1,8 @@
-using Eventuous.Diagnostics;
+// Copyright (C) 2021-2022 Ubiquitous AS. All rights reserved
+// Licensed under the Apache License, Version 2.0.
+
 using Eventuous.Producers;
 using Eventuous.Producers.Diagnostics;
-using Microsoft.Extensions.Hosting;
 using Microsoft.Extensions.Options;
 using static Google.Cloud.PubSub.V1.PublisherClient;
 
@@ -13,7 +14,7 @@ namespace Eventuous.GooglePubSub.Producers;
 /// Producer for Google PubSub
 /// </summary>
 [PublicAPI]
-public class GooglePubSubProducer : BaseProducer<PubSubProduceOptions>, IHostedService {
+public class GooglePubSubProducer : BaseProducer<PubSubProduceOptions>, IHostedProducer {
     readonly IEventSerializer _serializer;
     readonly ClientCache      _clientCache;
     readonly PubSubAttributes _attributes;
@@ -47,7 +48,7 @@ public GooglePubSubProducer(
     public GooglePubSubProducer(
         PubSubProducerOptions options,
         IEventSerializer?     serializer = null
-    ) : base(true, TracingOptions) {
+    ) : base(TracingOptions) {
         Ensure.NotNull(options);
 
         _serializer  = serializer ?? DefaultEventSerializer.Instance;
@@ -66,7 +67,7 @@ public GooglePubSubProducer(
     ) : this(options.Value, serializer) { }
 
     public Task StartAsync(CancellationToken cancellationToken = default) {
-        ReadyNow();
+        Ready = true;
         return Task.CompletedTask;
     }
 
@@ -108,9 +109,9 @@ PubsubMessage CreateMessage(ProducedMessage message, PubSubProduceOptions? optio
             OrderingKey = options?.OrderingKey ?? "",
             Attributes = {
                 { _attributes.ContentType, contentType },
-                { _attributes.EventType, eventType }
+                { _attributes.EventType, eventType },
+                { _attributes.MessageId, message.MessageId.ToString() }
             },
-            MessageId = message.MessageId.ToString()
         };
 
         if (message.Metadata != null) {
@@ -131,4 +132,6 @@ PubsubMessage CreateMessage(ProducedMessage message, PubSubProduceOptions? optio
 
         return psm;
     }
+
+    public bool Ready { get; private set; }
 }

src/GooglePubSub/src/Eventuous.GooglePubSub/PubSubAttributes.cs

@@ -4,4 +4,5 @@ namespace Eventuous.GooglePubSub;
 public class PubSubAttributes {
     public string EventType   { get; set; } = "eventType";
     public string ContentType { get; set; } = "contentType";
+    public string MessageId   { get; set; } = "messageId";
 }

src/Kafka/src/Eventuous.Kafka/Producers/KafkaBasicProducer.cs

@@ -1,3 +1,6 @@
+// Copyright (C) 2021-2022 Ubiquitous AS. All rights reserved
+// Licensed under the Apache License, Version 2.0.
+
 using Confluent.Kafka;
 using Eventuous.Producers;
 using Eventuous.Producers.Diagnostics;
@@ -9,13 +12,13 @@ namespace Eventuous.Kafka.Producers;
 /// Produces messages with byte[] payload without using the schema registry. The message type is specified in the
 /// headers, so the type mapping is required.
 /// </summary>
-public class KafkaBasicProducer : BaseProducer<KafkaProduceOptions>, IHostedService {
+public class KafkaBasicProducer : BaseProducer<KafkaProduceOptions>, IHostedProducer {
     readonly IProducer<string, byte[]> _producerWithKey;
     readonly IProducer<Null, byte[]>   _producerWithoutKey;
     readonly IEventSerializer          _serializer;
 
     public KafkaBasicProducer(KafkaProducerOptions options, IEventSerializer? serializer = null) :
-        base(true, TracingOptions) {
+        base(TracingOptions) {
         _producerWithKey    = new ProducerBuilder<string, byte[]>(options.ProducerConfig).Build();
         _producerWithoutKey = new DependentProducerBuilder<Null, byte[]>(_producerWithKey.Handle).Build();
         _serializer         = serializer ?? DefaultEventSerializer.Instance;
@@ -91,7 +94,7 @@ void Report(Error error) {
     }
 
     public Task StartAsync(CancellationToken cancellationToken) {
-        ReadyNow();
+        Ready = true;
         return Task.CompletedTask;
     }
 
@@ -103,4 +106,6 @@ public async Task StopAsync(CancellationToken cancellationToken) {
             await Task.Delay(100, cancellationToken).NoContext();
         }
     }
+
+    public bool Ready { get; private set; }
 }

src/RabbitMq/src/Eventuous.RabbitMq/Producers/RabbitMqProducer.cs

@@ -1,3 +1,6 @@
+// Copyright (C) 2021-2022 Ubiquitous AS. All rights reserved
+// Licensed under the Apache License, Version 2.0.
+
 using System.Diagnostics;
 using Eventuous.Diagnostics;
 using Eventuous.Producers;
@@ -11,7 +14,7 @@ namespace Eventuous.RabbitMq.Producers;
 /// RabbitMQ producer
 /// </summary>
 [PublicAPI]
-public class RabbitMqProducer : BaseProducer<RabbitMqProduceOptions>, IHostedService {
+public class RabbitMqProducer : BaseProducer<RabbitMqProduceOptions>, IHostedProducer {
     readonly RabbitMqExchangeOptions? _options;
     readonly IEventSerializer         _serializer;
     readonly ConnectionFactory        _connectionFactory;
@@ -29,7 +32,7 @@ public RabbitMqProducer(
         ConnectionFactory        connectionFactory,
         IEventSerializer?        serializer = null,
         RabbitMqExchangeOptions? options    = null
-    ) : base(true, TracingOptions) {
+    ) : base(TracingOptions) {
         _options           = options;
         _serializer        = serializer ?? DefaultEventSerializer.Instance;
         _connectionFactory = Ensure.NotNull(connectionFactory);
@@ -39,9 +42,7 @@ public Task StartAsync(CancellationToken cancellationToken = default) {
         _connection = _connectionFactory.CreateConnection();
         _channel    = _connection.CreateModel();
         _channel.ConfirmSelect();
-
-        ReadyNow();
-
+        Ready = true;
         return Task.CompletedTask;
     }
 
@@ -146,4 +147,6 @@ public Task StopAsync(CancellationToken cancellationToken = default) {
 
         return Task.CompletedTask;
     }
+
+    public bool Ready { get; private set; }
 }