code cleanup

cleaned up some code as well as modified the NATS.io connector to use headers to hide the additional message properties to avoid using additional external libraries and for cleaner usage.
roger-castaldo · Jul 30, 2024 · 14b9920 · 14b9920
1 parent eb9afe4
commit 14b9920
Show file tree

Hide file tree

Showing 35 changed files with 99 additions and 332 deletions.
diff --git a/Abstractions/Interfaces/Messages/IEncodedMessage.cs b/Abstractions/Interfaces/Messages/IEncodedMessage.cs
@@ -1,9 +1,4 @@
 using MQContract.Messages;
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
 
 namespace MQContract.Interfaces.Messages
 {

diff --git a/Abstractions/Messages/MessageHeader.cs b/Abstractions/Messages/MessageHeader.cs
@@ -1,10 +1,4 @@
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
-
-namespace MQContract.Messages
+namespace MQContract.Messages
 {
     public sealed class MessageHeader(IEnumerable<KeyValuePair<string,string>> data)
         : IMessageHeader

diff --git a/Abstractions/Messages/PingResult.cs b/Abstractions/Messages/PingResult.cs
@@ -1,10 +1,4 @@
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
-
-namespace MQContract.Messages
+namespace MQContract.Messages
 {
     public record PingResult(string Host,string Version, TimeSpan ResponseTime)
     {}

diff --git a/Abstractions/Messages/QueryResult.cs b/Abstractions/Messages/QueryResult.cs
@@ -1,10 +1,4 @@
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
-
-namespace MQContract.Messages
+namespace MQContract.Messages
 {
     public record QueryResult<T>(string ID,IMessageHeader Header,T? Result=null,string? Error=null)
         : TransmissionResult(ID,Error)

diff --git a/Abstractions/Messages/RecievedServiceMessage.cs b/Abstractions/Messages/RecievedServiceMessage.cs
@@ -1,10 +1,4 @@
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
-
-namespace MQContract.Messages
+namespace MQContract.Messages
 {
     public record RecievedServiceMessage(string ID, string MessageTypeID, string Channel, IMessageHeader Header, ReadOnlyMemory<byte> Data)
         : ServiceMessage(ID,MessageTypeID,Channel,Header,Data)

diff --git a/Abstractions/Messages/ServiceMessage.cs b/Abstractions/Messages/ServiceMessage.cs
@@ -1,9 +1,4 @@
 using MQContract.Interfaces.Messages;
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
 
 namespace MQContract.Messages
 {

diff --git a/Abstractions/Messages/ServiceQueryResult.cs b/Abstractions/Messages/ServiceQueryResult.cs
@@ -1,9 +1,4 @@
 using MQContract.Interfaces.Messages;
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
 
 namespace MQContract.Messages
 {

diff --git a/Abstractions/Messages/TransmissionResult.cs b/Abstractions/Messages/TransmissionResult.cs
@@ -1,10 +1,4 @@
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
-
-namespace MQContract.Messages
+namespace MQContract.Messages
 {
     public record TransmissionResult(string ID,string? Error=null)
     {

diff --git a/AutomatedTesting/ChannelMapperTests.cs b/AutomatedTesting/ChannelMapperTests.cs
@@ -3,14 +3,8 @@
 using MQContract.Attributes;
 using MQContract.Interfaces.Service;
 using MQContract;
-using System;
-using System.Collections.Generic;
 using System.Diagnostics;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
 using System.Reflection;
-using MQContract.Interfaces;
 using System.Text.Json;
 
 namespace AutomatedTesting

diff --git a/AutomatedTesting/ContractConnectionTests/PublishTests.cs b/AutomatedTesting/ContractConnectionTests/PublishTests.cs
@@ -6,7 +6,6 @@
 using Moq;
 using MQContract;
 using MQContract.Attributes;
-using MQContract.Interfaces;
 using MQContract.Interfaces.Encoding;
 using MQContract.Interfaces.Encrypting;
 using MQContract.Interfaces.Service;

diff --git a/AutomatedTesting/ContractConnectionTests/QueryTests.cs b/AutomatedTesting/ContractConnectionTests/QueryTests.cs
@@ -9,7 +9,6 @@
 using MQContract.Interfaces.Encoding;
 using MQContract.Interfaces.Encrypting;
 using MQContract.Interfaces.Service;
-using MQContract.Messages;
 using System.Diagnostics;
 using System.IO.Compression;
 using System.Reflection;

diff --git a/AutomatedTesting/Helper.cs b/AutomatedTesting/Helper.cs
@@ -1,8 +1,5 @@
 using AutomatedTesting.ServiceInjection;
 using Microsoft.Extensions.DependencyInjection;
-using Moq;
-using System.Threading.Tasks;
-using System.Threading;
 
 namespace AutomatedTesting
 {

diff --git a/Connectors/KubeMQ/Connection.cs b/Connectors/KubeMQ/Connection.cs
@@ -10,7 +10,6 @@
 using MQContract.KubeMQ.Subscriptions;
 using MQContract.Messages;
 using System.Diagnostics;
-using System.IO.MemoryMappedFiles;
 using System.Text.RegularExpressions;
 
 namespace MQContract.KubeMQ

diff --git a/Connectors/KubeMQ/Interfaces/IKubeMQPingResult.cs b/Connectors/KubeMQ/Interfaces/IKubeMQPingResult.cs
@@ -1,11 +1,4 @@
-using MQContract.Messages;
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
-
-namespace MQContract.KubeMQ.Interfaces
+namespace MQContract.KubeMQ.Interfaces
 {
     public interface IKubeMQPingResult
     {

diff --git a/Connectors/KubeMQ/Subscriptions/PubSubscription.cs b/Connectors/KubeMQ/Subscriptions/PubSubscription.cs
@@ -1,6 +1,5 @@
 using Grpc.Core;
 using Microsoft.Extensions.Logging;
-using MQContract.KubeMQ.Messages;
 using MQContract.KubeMQ.Options;
 using MQContract.KubeMQ.SDK.Connection;
 using MQContract.KubeMQ.SDK.Grpc;

diff --git a/Connectors/KubeMQ/Subscriptions/QuerySubscription.cs b/Connectors/KubeMQ/Subscriptions/QuerySubscription.cs
@@ -1,6 +1,5 @@
 using Grpc.Core;
 using Microsoft.Extensions.Logging;
-using MQContract.KubeMQ.Messages;
 using MQContract.KubeMQ.SDK.Connection;
 using MQContract.KubeMQ.SDK.Grpc;
 using MQContract.Messages;

diff --git a/Connectors/NATS/Connection.cs b/Connectors/NATS/Connection.cs
@@ -1,26 +1,21 @@
 using Microsoft.Extensions.Logging;
-using Microsoft.Extensions.Logging.Abstractions;
-using Microsoft.Extensions.Options;
 using MQContract.Interfaces.Service;
 using MQContract.Messages;
-using MQContract.NATS.Messages;
 using MQContract.NATS.Options;
-using MQContract.NATS.Serialization;
 using MQContract.NATS.Subscriptions;
 using NATS.Client.Core;
 using NATS.Client.JetStream;
 using NATS.Client.JetStream.Models;
-using System;
-using System.Collections.Generic;
-using System.Linq;
 using System.Text;
-using System.Threading.Channels;
-using System.Threading.Tasks;
 
 namespace MQContract.NATS
 {
     public class Connection : IMessageServiceConnection,IDisposable
     {
+        private const string MESSAGE_IDENTIFIER_HEADER = "_MessageID";
+        private const string MESSAGE_TYPE_HEADER = "_MessageTypeID";
+        private const string QUERY_RESPONSE_ERROR_TYPE = "NatsQueryError";
+
         private readonly NatsConnection natsConnection;
         private readonly NatsJSContext natsJSContext;
         private readonly ILogger? logger;
@@ -101,14 +96,42 @@ public async Task<PingResult> PingAsync()
                 await natsConnection.PingAsync()
             );
 
-        public static NatsHeaders ExtractHeader(IMessageHeader header)
+        internal static NatsHeaders ExtractHeader(ServiceMessage message)
             => new(new Dictionary<string, Microsoft.Extensions.Primitives.StringValues>(
-                header.Keys.Select(k=>
+                message.Header.Keys.Select(k=>
                     new KeyValuePair<string, Microsoft.Extensions.Primitives.StringValues>(
                         k,
-                        new Microsoft.Extensions.Primitives.StringValues(header[k]))
+                        new Microsoft.Extensions.Primitives.StringValues(message.Header[k]))
                 )
+                .Concat([
+                    new(MESSAGE_IDENTIFIER_HEADER,message.ID),
+                    new(MESSAGE_TYPE_HEADER,message.MessageTypeID)
+                ])
             ));
+        internal static IMessageHeader ExtractHeader(NatsHeaders? header,out string? messageID,out string? messageTypeID)
+        {
+            if (header?.TryGetValue(MESSAGE_IDENTIFIER_HEADER, out var mid)??false)
+                messageID = mid.ToString();
+            else
+                messageID=null;
+            if (header?.TryGetValue(MESSAGE_TYPE_HEADER, out var mti)??false)
+                messageTypeID=mti.ToString();
+            else
+                messageTypeID=null;
+            return new MessageHeader(header?
+                .Where(pair=>!Equals(pair.Key,MESSAGE_IDENTIFIER_HEADER)&&!Equals(pair.Key,MESSAGE_TYPE_HEADER))
+                .Select(pair => new KeyValuePair<string, string>(pair.Key, pair.Value.ToString()))?? []
+            );
+        }
+
+        internal static NatsHeaders ProduceQueryError(Exception exception,string messageID,out byte[] data)
+        {
+            data = UTF8Encoding.UTF8.GetBytes(exception.Message);
+            return new(new Dictionary<string,Microsoft.Extensions.Primitives.StringValues>([
+                    new KeyValuePair<string,Microsoft.Extensions.Primitives.StringValues>(MESSAGE_IDENTIFIER_HEADER,messageID),
+                    new KeyValuePair<string,Microsoft.Extensions.Primitives.StringValues>(MESSAGE_TYPE_HEADER,QUERY_RESPONSE_ERROR_TYPE)
+            ]));
+        }
 
         public async Task<TransmissionResult> PublishAsync(ServiceMessage message, TimeSpan timeout, IServiceChannelOptions? options = null, CancellationToken cancellationToken = default)
         {
@@ -120,32 +143,20 @@ public async Task<TransmissionResult> PublishAsync(ServiceMessage message, TimeS
                 {
                     if (publishChannelOptions.Config!=null)
                         await CreateStreamAsync(publishChannelOptions.Config, cancellationToken);
-                    var ack = await natsJSContext.PublishAsync<NatsMessage>(
+                    var ack = await natsJSContext.PublishAsync<byte[]>(
                         message.Channel,
-                        new NatsMessage()
-                        {
-                            ID=message.ID,
-                            MessageTypeID=message.MessageTypeID,
-                            Data=message.Data
-                        },
-                        headers: ExtractHeader(message.Header),
-                        serializer: MessageSerializer<NatsMessage>.Default,
+                        message.Data.ToArray(),
+                        headers: ExtractHeader(message),
                         cancellationToken: cancellationToken
                     );
                     return new TransmissionResult(message.ID, (ack.Error!=null ? $"{ack.Error.Code}:{ack.Error.Description}" : null));
                 }
                 else
                 {
-                    await natsConnection.PublishAsync<NatsMessage>(
+                    await natsConnection.PublishAsync<byte[]>(
                         message.Channel,
-                        new NatsMessage()
-                        {
-                            ID=message.ID,
-                            MessageTypeID=message.MessageTypeID,
-                            Data=message.Data
-                        },
-                        headers: ExtractHeader(message.Header),
-                        serializer: MessageSerializer<NatsMessage>.Default,
+                        message.Data.ToArray(),
+                        headers: ExtractHeader(message),
                         cancellationToken: cancellationToken
                     );
                     return new TransmissionResult(message.ID);
@@ -158,26 +169,20 @@ await natsConnection.PublishAsync<NatsMessage>(
 
         public async Task<ServiceQueryResult> QueryAsync(ServiceMessage message, TimeSpan timeout, IServiceChannelOptions? options = null, CancellationToken cancellationToken = default)
         {
-            var result = await natsConnection.RequestAsync<NatsMessage, NatsQueryResponseMessage>(
+            var result = await natsConnection.RequestAsync<byte[], byte[]>(
                 message.Channel,
-                new NatsMessage()
-                {
-                    ID=message.ID,
-                    MessageTypeID=message.MessageTypeID,
-                    Data=message.Data
-                },
-                headers: ExtractHeader(message.Header),
-                requestSerializer: MessageSerializer<NatsMessage>.Default,
-                replySerializer: MessageSerializer<NatsQueryResponseMessage>.Default,
+                message.Data.ToArray(),
+                headers: ExtractHeader(message),
                 cancellationToken: cancellationToken
             );
-            if (!string.IsNullOrWhiteSpace(result.Data?.Error))
-                throw new QueryAsyncReponseException(result.Data?.Error);
+            if (Equals(result.Headers?[MESSAGE_TYPE_HEADER], QUERY_RESPONSE_ERROR_TYPE))
+                throw new QueryAsyncReponseException(UTF8Encoding.UTF8.GetString(result.Data!));
+            var headers = ExtractHeader(result.Headers, out var messageID, out var messageTypeID);
             return new ServiceQueryResult(
-                result.Data?.ID??string.Empty,
-                new MQContract.NATS.Messages.MessageHeader(result.Headers),
-                result.Data?.MessageTypeID??string.Empty,
-                result.Data?.Data??new ReadOnlyMemory<byte>()
+                messageID??string.Empty,
+                headers,
+                messageTypeID??string.Empty,
+                result.Data??new ReadOnlyMemory<byte>()
             );
         }
 
@@ -195,10 +200,9 @@ public async Task<ServiceQueryResult> QueryAsync(ServiceMessage message, TimeSpa
             }
             else
                 subscription = new PublishSubscription(
-                    natsConnection.SubscribeAsync<NatsMessage>(
+                    natsConnection.SubscribeAsync<byte[]>(
                         channel,
                         queueGroup: group,
-                        serializer: MessageSerializer<NatsMessage>.Default,
                         cancellationToken: cancellationToken
                     ),
                     messageRecieved,
@@ -215,10 +219,9 @@ public async Task<ServiceQueryResult> QueryAsync(ServiceMessage message, TimeSpa
         public async Task<IServiceSubscription?> SubscribeQueryAsync(Func<RecievedServiceMessage, Task<ServiceMessage>> messageRecieved, Action<Exception> errorRecieved, string channel, string group, IServiceChannelOptions? options = null, CancellationToken cancellationToken = default)
         {
             var sub = new QuerySubscription(
-                natsConnection.SubscribeAsync<NatsMessage>(
+                natsConnection.SubscribeAsync<byte[]>(
                     channel,
                     queueGroup: group,
-                    serializer: MessageSerializer<NatsMessage>.Default,
                     cancellationToken: cancellationToken
                 ),
                 messageRecieved,

diff --git a/Connectors/NATS/Exceptions.cs b/Connectors/NATS/Exceptions.cs
@@ -1,10 +1,4 @@
-using System;
-using System.Collections.Generic;
-using System.Linq;
-using System.Text;
-using System.Threading.Tasks;
-
-namespace MQContract.NATS
+namespace MQContract.NATS
 {
     /// <summary>
     /// Thrown when an error occurs attempting to connect to the NATS server.  

diff --git a/Connectors/NATS/Messages/MessageHeader.cs b/Connectors/NATS/Messages/MessageHeader.cs
diff --git a/Connectors/NATS/Messages/NatsMessage.cs b/Connectors/NATS/Messages/NatsMessage.cs