Refactor Classes names (#177)

* Refactor Classes names

Closes: #173
Closes: #171

Signed-off-by: Gabriele Santomaggio <[email protected]>

Signed-off-by: Gabriele Santomaggio <[email protected]>
Co-authored-by: Luke Bakken <[email protected]>

Author: Gsantomaggio

.ci/versions.json

@@ -1,4 +1,4 @@
 {
-  "erlang": "25.0.4",
-  "rabbitmq": "3.11.0"
+  "erlang": "25.1.2",
+  "rabbitmq": "3.11.2"
 }

Examples/Performances/BatchVsBatchSend.cs

@@ -35,10 +35,8 @@ private static async Task RProducerSend(string stream, StreamSystem system)
         var total = 0;
         var confirmed = 0;
         var error = 0;
-        var reliableProducer = await ReliableProducer.CreateReliableProducer(new ReliableProducerConfig()
+        var reliableProducer = await Producer.Create(new ProducerConfig(system, stream)
         {
-            Stream = stream,
-            StreamSystem = system,
             MaxInFlight = 1_000_000,
             ConfirmationHandler = messagesConfirmed =>
             {
@@ -87,10 +85,8 @@ private static async Task RProducerBatchSend(string stream, StreamSystem system)
         var total = 0;
         var confirmed = 0;
         var error = 0;
-        var reliableProducer = await ReliableProducer.CreateReliableProducer(new ReliableProducerConfig()
+        var producer = await Producer.Create(new ProducerConfig(system,stream)
         {
-            Stream = stream,
-            StreamSystem = system,
             MaxInFlight = 1_000_000,
             ConfirmationHandler = messagesConfirmed =>
             {
@@ -122,7 +118,7 @@ private static async Task RProducerBatchSend(string stream, StreamSystem system)
             messages.Add(new Message(array));
             if (i % AggregateBatchSize == 0)
             {
-                await reliableProducer.BatchSend(messages);
+                await producer.Send(messages);
                 messages.Clear();
             }
 
@@ -132,23 +128,22 @@ private static async Task RProducerBatchSend(string stream, StreamSystem system)
             }
         }
 
-        await reliableProducer.BatchSend(messages);
+        await producer.Send(messages);
         messages.Clear();
 
         Console.WriteLine(
             $"*****Reliable Producer Batch Send***** time: {DateTime.Now - start}, messages sent: {TotalMessages}");
         Thread.Sleep(1000);
-        await reliableProducer.Close();
+        await producer.Close();
     }
 
 
     private static async Task StandardProducerSend(string stream, StreamSystem system)
     {
         Console.WriteLine("*****Standard Producer Send*****");
         var confirmed = 0;
-        var producer = await system.CreateProducer(new ProducerConfig()
+        var producer = await system.CreateRawProducer(new RawProducerConfig(stream)
         {
-            Stream = stream,
             MaxInFlight = 1_000_000,
             ConfirmHandler = _ =>
             {
@@ -182,9 +177,8 @@ private static async Task BatchSend(StreamSystem system, string stream)
     {
         Console.WriteLine("*****Standard Batch Send*****");
         var confirmed = 0;
-        var producer = await system.CreateProducer(new ProducerConfig()
+        var producer = await system.CreateRawProducer(new RawProducerConfig(stream)
         {
-            Stream = stream,
             MaxInFlight = 1_000_000,
             ConfirmHandler = _ =>
             {
@@ -202,7 +196,7 @@ private static async Task BatchSend(StreamSystem system, string stream)
             messages.Add((i, new Message(array)));
             if (i % AggregateBatchSize == 0)
             {
-                await producer.BatchSend(messages);
+                await producer.Send(messages);
                 messages.Clear();
             }
 
@@ -212,7 +206,7 @@ private static async Task BatchSend(StreamSystem system, string stream)
             }
         }
 
-        await producer.BatchSend(messages);
+        await producer.Send(messages);
         messages.Clear();
 
         Console.WriteLine(

RabbitMQ.Stream.Client.PerfTest/Program.fs

@@ -20,7 +20,7 @@ let main argv =
     let mutable confirmed = 0
     let mutable prod = null
     let streamName = "dotnet-perftest"
-    let consumerConfig = ConsumerConfig(Stream = streamName,
+    let consumerConfig = RawConsumerConfig(streamName,
                                         Reference = Guid.NewGuid().ToString(),
                                         MessageHandler =
                                             fun c ctx m ->
@@ -33,12 +33,12 @@ let main argv =
         let! system = StreamSystem.Create config
         let! stream = system.CreateStream(StreamSpec(streamName))
         printfn $"Stream: {streamName}"
-        let! consumer = system.CreateConsumer(consumerConfig)
-        let producerConfig = ProducerConfig(Stream = streamName,
+        let! consumer = system.CreateRawConsumer(consumerConfig)
+        let producerConfig = RawProducerConfig(streamName,
                                             Reference = null,
                                             MaxInFlight = 10000,
                                             ConfirmHandler = fun c -> confirmed <- confirmed + 1)
-        let! producer = system.CreateProducer producerConfig
+        let! producer = system.CreateRawProducer producerConfig
         //make producer available to metrics async
         prod <- producer
         let msg = Message "asdf"B

RabbitMQ.Stream.Client/Client.cs

@@ -296,14 +296,14 @@ await Request<DeletePublisherRequest, DeletePublisherResponse>(corr =>
             Dictionary<string, string> properties, Func<Deliver, Task> deliverHandler,
             Func<bool, Task<IOffsetType>> consumerUpdateHandler = null)
         {
-            return await Subscribe(new ConsumerConfig() { Stream = stream, OffsetSpec = offsetType },
+            return await Subscribe(new RawConsumerConfig(stream) { OffsetSpec = offsetType },
                 initialCredit,
                 properties,
                 deliverHandler,
                 consumerUpdateHandler);
         }
 
-        public async Task<(byte, SubscribeResponse)> Subscribe(ConsumerConfig config,
+        public async Task<(byte, SubscribeResponse)> Subscribe(RawConsumerConfig config,
             ushort initialCredit,
             Dictionary<string, string> properties, Func<Deliver, Task> deliverHandler,
             Func<bool, Task<IOffsetType>> consumerUpdateHandler)
@@ -614,6 +614,7 @@ public async Task<bool> StreamExists(string stream)
             return response.StreamInfos is { Count: >= 1 } &&
                    response.StreamInfos[stream].ResponseCode == ResponseCode.Ok;
         }
+
         public async ValueTask<QueryOffsetResponse> QueryOffset(string reference, string stream)
         {
             return await Request<QueryOffsetRequest, QueryOffsetResponse>(corr =>

RabbitMQ.Stream.Client/IConsumer.cs

@@ -24,7 +24,7 @@ public record IConsumerConfig : INamedEntity
     internal IOffsetType StoredOffsetSpec { get; set; }
 
     // ClientProvidedName is used to identify TCP connection name.
-    public string ClientProvidedName { get; set; } = "dotnet-stream-consumer";
+    public string ClientProvidedName { get; set; } = "dotnet-stream-raw-consumer";
 
     // SingleActiveConsumer is used to indicate that there is only one consumer active for the stream.
     // given a consumer reference. 

RabbitMQ.Stream.Client/IProducer.cs

@@ -16,13 +16,44 @@ namespace RabbitMQ.Stream.Client;
 
 public interface IProducer
 {
+    /// <summary>
+    /// Send the message to the stream in asynchronous mode.
+    /// The client will aggregate messages and send them in batches.
+    /// The batch size is configurable. See IProducerConfig.BatchSize.
+    /// </summary>
+    /// <param name="publishingId">Publishing id</param>
+    /// <param name="message"> Message </param>
+    /// <returns></returns>
     public ValueTask Send(ulong publishingId, Message message);
-    public ValueTask BatchSend(List<(ulong, Message)> messages);
 
+    /// <summary>
+    /// Send the messages in batch to the stream in synchronous mode.
+    /// The aggregation is provided by the user.
+    /// The client will send the messages in the order they are provided.
+    /// </summary>
+    /// <param name="messages">Batch messages to send</param>
+    /// <returns></returns>
+    public ValueTask Send(List<(ulong, Message)> messages);
+
+    /// <summary>
+    /// Enable sub-batch feature.
+    /// It is needed when you need to sub aggregate the messages and compress them.
+    /// For example you can aggregate 100 log messages and compress to reduce the space.
+    /// One single publishingId can have multiple sub-batches messages.
+    /// See also: https://rabbitmq.github.io/rabbitmq-stream-java-client/stable/htmlsingle/#sub-entry-batching-and-compression
+    /// </summary>
+    /// <param name="publishingId"></param>
+    /// <param name="subEntryMessages">Messages to aggregate</param>
+    /// <param name="compressionType"> Type of compression. By default the client supports GZIP and none</param>
+    /// <returns></returns>
     public ValueTask Send(ulong publishingId, List<Message> subEntryMessages, CompressionType compressionType);
 
     public Task<ResponseCode> Close();
 
+    /// <summary>
+    /// Return the last publishing id.
+    /// </summary>
+    /// <returns></returns>
     public Task<ulong> GetLastPublishingId();
 
     public bool IsOpen();
@@ -41,7 +72,7 @@ public record IProducerConfig : INamedEntity
 {
     public string Reference { get; set; }
     public int MaxInFlight { get; set; } = 1000;
-    public string ClientProvidedName { get; set; } = "dotnet-stream-producer";
+    public string ClientProvidedName { get; set; } = "dotnet-stream-raw-producer";
 
     public int BatchSize { get; set; } = 100;