Add ICrc32 interface

If a library user wishes to have Crc32 checking, they must implement the
`ICrc32` interface and set an implementation on the configuration.

References:
* #19
* #285

RabbitMQ.Stream.Client/IConsumer.cs

@@ -69,5 +69,5 @@ public ushort InitialCredits
     // enables the check of the crc on the delivery.
     // the server will send the crc for each chunk and the client will check it.
     // It is not enabled by default because it is could reduce the performance.
-    public bool CheckCrcOnDelivery { get; set; } = false;
+    public ICrc32 Crc32 { get; set; } = null;
 }

RabbitMQ.Stream.Client/ICrc32.cs

@@ -0,0 +1,17 @@
+// This source code is dual-licensed under the Apache License, version
+// 2.0, and the Mozilla Public License, version 2.0.
+// Copyright (c) 2007-2023 VMware, Inc.
+
+namespace RabbitMQ.Stream.Client
+{
+    /// <summary>
+    /// ICrc32 defines an interface for implementing crc32 hashing.
+    /// Library users who wish to perform crc32 checks on data from RabbitMQ
+    /// should implement this interface and assign an instance to
+    /// <see cref="IConsumerConfig.Crc32"><code>IConsumerConfig.Crc32</code></see>.
+    /// </summary>
+    public interface ICrc32
+    {
+        byte[] Hash(byte[] data);
+    }
+}

RabbitMQ.Stream.Client/PublicAPI.Unshipped.txt

@@ -23,8 +23,8 @@ RabbitMQ.Stream.Client.ConsumerFilter.PostFilter.set -> void
 RabbitMQ.Stream.Client.ConsumerFilter.Values.get -> System.Collections.Generic.List<string>
 RabbitMQ.Stream.Client.ConsumerFilter.Values.set -> void
 RabbitMQ.Stream.Client.HashRoutingMurmurStrategy.Route(RabbitMQ.Stream.Client.Message message, System.Collections.Generic.List<string> partitions) -> System.Threading.Tasks.Task<System.Collections.Generic.List<string>>
-RabbitMQ.Stream.Client.IConsumerConfig.CheckCrcOnDelivery.get -> bool
-RabbitMQ.Stream.Client.IConsumerConfig.CheckCrcOnDelivery.set -> void
+RabbitMQ.Stream.Client.IConsumerConfig.Crc32.get -> RabbitMQ.Stream.Client.ICrc32
+RabbitMQ.Stream.Client.IConsumerConfig.Crc32.set -> void
 RabbitMQ.Stream.Client.IConsumerConfig.ConsumerFilter.get -> RabbitMQ.Stream.Client.ConsumerFilter
 RabbitMQ.Stream.Client.CommandVersions
 RabbitMQ.Stream.Client.CommandVersions.Command.get -> ushort
@@ -50,6 +50,8 @@ RabbitMQ.Stream.Client.ICommandVersions.MinVersion.get -> ushort
 RabbitMQ.Stream.Client.IConsumerConfig.ConsumerFilter.set -> void
 RabbitMQ.Stream.Client.IConsumerConfig.InitialCredits.get -> ushort
 RabbitMQ.Stream.Client.IConsumerConfig.InitialCredits.set -> void
+RabbitMQ.Stream.Client.ICrc32
+RabbitMQ.Stream.Client.ICrc32.Hash(byte[] data) -> byte[]
 RabbitMQ.Stream.Client.IProducerConfig.Filter.get -> RabbitMQ.Stream.Client.ProducerFilter
 RabbitMQ.Stream.Client.IProducerConfig.Filter.set -> void
 RabbitMQ.Stream.Client.IRoutingStrategy.Route(RabbitMQ.Stream.Client.Message message, System.Collections.Generic.List<string> partitions) -> System.Threading.Tasks.Task<System.Collections.Generic.List<string>>
@@ -69,8 +71,8 @@ RabbitMQ.Stream.Client.PublishFilter.SizeNeeded.get -> int
 RabbitMQ.Stream.Client.PublishFilter.Write(System.Span<byte> span) -> int
 RabbitMQ.Stream.Client.RawSuperStreamProducerConfig.RoutingStrategyType.get -> RabbitMQ.Stream.Client.RoutingStrategyType
 RabbitMQ.Stream.Client.RawSuperStreamProducerConfig.RoutingStrategyType.set -> void
-RabbitMQ.Stream.Client.Reliable.ConsumerConfig.CheckCrcOnDelivery.get -> bool
-RabbitMQ.Stream.Client.Reliable.ConsumerConfig.CheckCrcOnDelivery.set -> void
+RabbitMQ.Stream.Client.Reliable.ConsumerConfig.Crc32.get -> RabbitMQ.Stream.Client.ICrc32
+RabbitMQ.Stream.Client.Reliable.ConsumerConfig.Crc32.set -> void
 RabbitMQ.Stream.Client.Reliable.ConsumerConfig.Filter.get -> RabbitMQ.Stream.Client.ConsumerFilter
 RabbitMQ.Stream.Client.Reliable.ConsumerConfig.Filter.set -> void
 RabbitMQ.Stream.Client.Reliable.ConsumerConfig.InitialCredits.get -> ushort

RabbitMQ.Stream.Client/RabbitMQ.Stream.Client.csproj

@@ -45,7 +45,6 @@
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
       <PrivateAssets>all</PrivateAssets>
     </PackageReference>
-    <PackageReference Include="System.IO.Hashing" />
     <PackageReference Include="System.IO.Pipelines" />
     <PackageReference Include="Microsoft.Extensions.Logging.Abstractions" />
   </ItemGroup>

RabbitMQ.Stream.Client/RawConsumer.cs

@@ -6,7 +6,6 @@
 using System.Buffers;
 using System.Collections.Generic;
 using System.Diagnostics;
-using System.IO.Hashing;
 using System.Runtime.CompilerServices;
 using System.Threading.Channels;
 using System.Threading.Tasks;
@@ -467,10 +466,10 @@ private async Task Init()
                     if (Token.IsCancellationRequested)
                         return;
 
-                    if (_config.CheckCrcOnDelivery)
+                    if (_config.Crc32 is not null)
                     {
                         var crcCalculated = BitConverter.ToUInt32(
-                            Crc32.Hash(deliver.Chunk.Data.ToArray())
+                            _config.Crc32.Hash(deliver.Chunk.Data.ToArray())
                         );
                         if (crcCalculated != deliver.Chunk.Crc)
                         {

RabbitMQ.Stream.Client/RawSuperStreamConsumer.cs

@@ -70,7 +70,7 @@ private RawConsumerConfig FromStreamConfig(string stream)
             IsSingleActiveConsumer = _config.IsSingleActiveConsumer,
             ConsumerUpdateListener = _config.ConsumerUpdateListener,
             ConsumerFilter = _config.ConsumerFilter,
-            CheckCrcOnDelivery = _config.CheckCrcOnDelivery,
+            Crc32 = _config.Crc32,
             ConnectionClosedHandler = async (string s) =>
             {
                 // if the stream is still in the consumer list

RabbitMQ.Stream.Client/Reliable/Consumer.cs

@@ -113,12 +113,12 @@ public record ConsumerConfig : ReliableConfig
     public ConsumerFilter Filter { get; set; } = null;
 
     /// <summary>
-    /// CheckCrcOnDelivery enables the check of the crc on the delivery.
-    /// the server will send the crc for each chunk and the client will check it.
+    /// Eenables the check of the crc on the delivery when set to an implementation
+    /// of <see cref="ICrc32"><code>ICrc32</code></see>.
+    /// >he server will send the crc for each chunk and the client will check it.
     /// It is not enabled by default because it is could reduce the performance.
-    /// Default value is false.
     /// </summary>
-    public bool CheckCrcOnDelivery { get; set; } = false;
+    public ICrc32 Crc32 { get; set; } = null;
 
     public ConsumerConfig(StreamSystem streamSystem, string stream) : base(streamSystem, stream)
     {

RabbitMQ.Stream.Client/Reliable/ConsumerFactory.cs

@@ -51,7 +51,7 @@ private async Task<IConsumer> StandardConsumer(bool boot)
             InitialCredits = _consumerConfig.InitialCredits,
             OffsetSpec = offsetSpec,
             ConsumerFilter = _consumerConfig.Filter,
-            CheckCrcOnDelivery = _consumerConfig.CheckCrcOnDelivery,
+            Crc32 = _consumerConfig.Crc32,
             ConnectionClosedHandler = async _ =>
             {
                 await TryToReconnect(_consumerConfig.ReconnectStrategy).ConfigureAwait(false);
@@ -111,7 +111,7 @@ private async Task<IConsumer> SuperConsumer(bool boot)
                 IsSingleActiveConsumer = _consumerConfig.IsSingleActiveConsumer,
                 InitialCredits = _consumerConfig.InitialCredits,
                 ConsumerFilter = _consumerConfig.Filter,
-                CheckCrcOnDelivery = _consumerConfig.CheckCrcOnDelivery,
+                Crc32 = _consumerConfig.Crc32,
                 OffsetSpec = offsetSpecs,
                 MessageHandler = async (stream, consumer, ctx, message) =>
                 {

Tests/Crc32.cs

@@ -0,0 +1,15 @@
+// This source code is dual-licensed under the Apache License, version
+// 2.0, and the Mozilla Public License, version 2.0.
+// Copyright (c) 2007-2023 VMware, Inc.
+
+using RabbitMQ.Stream.Client;
+
+namespace Tests;
+
+public class Crc32 : ICrc32
+{
+    public byte[] Hash(byte[] data)
+    {
+        return System.IO.Hashing.Crc32.Hash(data);
+    }
+}

Tests/RawConsumerSystemTests.cs

@@ -19,6 +19,7 @@ namespace Tests
 {
     public class ConsumerSystemTests
     {
+        private readonly ICrc32 _crc32 = new Crc32();
         private readonly ITestOutputHelper testOutputHelper;
 
         public ConsumerSystemTests(ITestOutputHelper testOutputHelper)
@@ -416,7 +417,7 @@ public async void Amqp091MessagesConsumer()
             var rawConsumer = await system.CreateRawConsumer(
                 new RawConsumerConfig(stream)
                 {
-                    CheckCrcOnDelivery = true,
+                    Crc32 = _crc32,
                     Reference = "consumer",
                     MessageHandler = async (consumer, ctx, message) =>
                     {
@@ -463,7 +464,7 @@ public async void ConsumerQueryOffset()
             var rawConsumer = await system.CreateRawConsumer(
                 new RawConsumerConfig(stream)
                 {
-                    CheckCrcOnDelivery = true,
+                    Crc32 = _crc32,
                     Reference = reference,
                     OffsetSpec = new OffsetTypeOffset(),
                     MessageHandler = async (consumer, ctx, message) =>
@@ -534,7 +535,7 @@ public async void ShouldConsumeFromStoredOffset()
             var rawConsumer = await system.CreateRawConsumer(
                 new RawConsumerConfig(stream)
                 {
-                    CheckCrcOnDelivery = false,
+                    Crc32 = _crc32,
                     Reference = Reference,
                     OffsetSpec = new OffsetTypeOffset(),
                     MessageHandler = async (consumer, ctx, _) =>

Tests/ReliableTests.cs

@@ -18,6 +18,7 @@ namespace Tests;
 
 public class ReliableTests
 {
+    private readonly ICrc32 _crc32 = new Crc32();
     private readonly ITestOutputHelper _testOutputHelper;
 
     public ReliableTests(ITestOutputHelper testOutputHelper)
@@ -323,7 +324,7 @@ public async void FirstConsumeAfterKillConnectionShouldContinueToWork()
         var messagesReceived = 0;
         var consumer = await Consumer.Create(new ConsumerConfig(system, stream)
         {
-            CheckCrcOnDelivery = true,
+            Crc32 = _crc32,
             Reference = reference,
             ClientProvidedName = clientProviderName,
             OffsetSpec = new OffsetTypeFirst(),
@@ -369,7 +370,7 @@ public async void ConsumeAfterKillConnectionShouldContinueToWork()
         var messagesReceived = 0;
         var consumer = await Consumer.Create(new ConsumerConfig(system, stream)
         {
-            CheckCrcOnDelivery = true,
+            Crc32 = _crc32,
             Reference = reference,
             ClientProvidedName = clientProviderName,
             OffsetSpec = new OffsetTypeFirst(),

Tests/SuperStreamConsumerTests.cs

@@ -19,6 +19,7 @@ namespace Tests;
 
 public class SuperStreamConsumerTests
 {
+    private readonly ICrc32 _crc32 = new Crc32();
     private readonly ITestOutputHelper _testOutputHelper;
 
     public SuperStreamConsumerTests(ITestOutputHelper testOutputHelper)
@@ -64,7 +65,7 @@ public async void NumberOfMessagesConsumedShouldBeEqualsToPublished()
         var consumer = await system.CreateSuperStreamConsumer(
             new RawSuperStreamConsumerConfig(SystemUtils.InvoicesExchange)
             {
-                CheckCrcOnDelivery = true,
+                Crc32 = _crc32,
                 ClientProvidedName = clientProvidedName,
                 OffsetSpec = await SystemUtils.OffsetsForSuperStreamConsumer(system, SystemUtils.InvoicesExchange, new OffsetTypeFirst()),
                 MessageHandler = (stream, consumer1, context, message) =>

Tests/Tests.csproj

@@ -10,6 +10,7 @@
     <PackageReference Include="AmqpNetLite" />
     <PackageReference Include="Microsoft.NET.Test.Sdk" />
     <PackageReference Include="RabbitMQ.Client" />
+    <PackageReference Include="System.IO.Hashing" />
     <PackageReference Include="xunit" />
     <PackageReference Include="xunit.runner.visualstudio">
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>