Added option to read messages from syncreply (#740)

* Add a new single queue sync messaging

* Startet på MessageListener

* Ser ut til at ting kommer inn på SynchronousReplyListener som ventet nå. Tar også med lokale hacks for å kopiere dll. Skal bort før merge

* Ryddet opp i ny listener som er testet i featuremiljø

* Lagt til tester og uppet versjonsnummer

* ryddet bort kopiering av DLLs

* Mergefeil

* Remove dlls

* fix test

---------

Co-authored-by: Martin Tysdahl Gedde-Dahl <[email protected]>

Author: GrevlingKake

Directory.build.props

@@ -1,7 +1,7 @@
 <Project>
   <PropertyGroup>
     <LangVersion>10.0</LangVersion>
-    <Version>5.2.1</Version>
+    <Version>6.0.0</Version>
   </PropertyGroup>
   <PropertyGroup>
     <Authors>Norsk Helsenett SF</Authors>

Documentation/MigrateFrom5To6.md

@@ -0,0 +1,19 @@
+## Migrere fra 5.0 til 6.0
+
+Hovedendringen for 6.0 er utvidelse av IMessagingClient og IMessagingServer for å kunne kontrollere synkron meldingsutveksling utenfor Helsenorge.Messaging. Dette er spesielt implementert for et konkret behov for bruk på helsenorge.no. 
+For brukere som ikke er avhengig av å kunne kontrollere syncreply-køen, så kan det oppgraderes til 6.0 uten å gjøre noen konfigurasjonsendringer.
+
+Alle endringer er dokumentert i dette migreringsdokumentet.
+
+`IMessagingClient` er utvidet med `SendWithoutWaitingAsync` for å kunne sende en melding til en sync-kø. Hvis denne brukes tar man selv ansvar for å hente ned svaret fra 
+køen spesifisert i `MessagingSettings:AmqpSettings:Synchronous:StaticReplyQueue`
+Hvis man ønsker å bruke den allerede eksisterende `SendAndWaitAsync` metoden, vil denne fortsatt fungere som før og trenger ikke populere opp den nye verdien.
+
+### Konfigurasjonsrelaterte endringer
+
+`MessagingSettings:AmqpSettings:Synchronous:StaticReplyQueue` er lagt til for å kunne spesifisere en svar-kø for synkron meldingsutveksling. Denne er ikke påkrevd
+
+### Nye features
+
+`IMessagingClient` er utvidet med `SendWithoutWaitingAsync`
+IMessagingServer er utvidet med muligheten for å lytte på syncreply-køen spesifisert i `MessagingSettings:AmqpSettings:Synchronous:StaticReplyQueue`

Messaging.sln

@@ -25,14 +25,14 @@ EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "workflows", "workflows", "{FB9DDD90-D5A5-4F25-8C81-4E6C9B4A18EE}"
 	ProjectSection(SolutionItems) = preProject
 		.github\workflows\build.yml = .github\workflows\build.yml
-		.github\workflows\nuget-deploy.yml = .github\workflows\nuget-deploy.yml
 		.github\workflows\codeql-analysis.yml = .github\workflows\codeql-analysis.yml
+		.github\workflows\nuget-deploy.yml = .github\workflows\nuget-deploy.yml
 	EndProjectSection
 EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "SolutionItems", "SolutionItems", "{C78E1F1C-6AE6-4E64-8211-90EEC7A5FCAD}"
 	ProjectSection(SolutionItems) = preProject
-		readme.md = readme.md
 		LICENSE = LICENSE
+		readme.md = readme.md
 	EndProjectSection
 EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Documentation", "Documentation", "{D65E8495-DAAD-46D6-8053-8D911A5C24F3}"
@@ -47,6 +47,7 @@ Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "Documentation", "Documentat
 		Documentation\KrypteringOgSignering.md = Documentation\KrypteringOgSignering.md
 		Documentation\MigrateFrom3To4.md = Documentation\MigrateFrom3To4.md
 		Documentation\MigrateFrom4To5.md = Documentation\MigrateFrom4To5.md
+		Documentation\MigrateFrom5To6.md = Documentation\MigrateFrom5To6.md
 		Documentation\MottaMeldinger.md = Documentation\MottaMeldinger.md
 		Documentation\OppdaterProxyKlasser.md = Documentation\OppdaterProxyKlasser.md
 		Documentation\Registre.md = Documentation\Registre.md

readme.md

@@ -21,7 +21,8 @@ Dette biblioteket støtter oppunder basisbehovet for en kommunikasjonsløsning v
 12. [Biztalk](Documentation/Biztalk.md "Biztalk")
 13. [Migrere fra 3.0 til 4.0](Documentation/MigrateFrom3To4.md)
 14. [Migrere fra 4.0 til 5.0](Documentation/MigrateFrom4To5.md)
-14. [Migrere fra SOAP til REST for 5.2 og senere versjoner](Documentation/MigrateFromSoapToRest.md)
+15. [Migrere fra SOAP til REST for 5.2 og senere versjoner](Documentation/MigrateFromSoapToRest.md)
+15. [Migrere fra 5.0 til 6.0](Documentation/MigrateFrom5To6.md)
 
 
 ## Contributing

src/Helsenorge.Messaging/Abstractions/IMessagingClient.cs

@@ -6,10 +6,8 @@
  * available at https://raw.githubusercontent.com/helsenorge/Helsenorge.Messaging/master/LICENSE
  */
 
-using System;
 using System.Threading.Tasks;
 using System.Xml.Linq;
-using Microsoft.Extensions.Logging;
 
 namespace Helsenorge.Messaging.Abstractions
 {
@@ -32,6 +30,14 @@ public interface IMessagingClient
         /// <returns></returns>
         Task<XDocument> SendAndWaitAsync(OutgoingMessage message);
 
+        /// <summary>
+        /// Send a message without waiting for a reply (synchronous messaging)
+        /// </summary>
+        /// <param name="message">Details about the message being sent</param>
+        /// <param name="correlationId">The correlation id to use when sending the message. Only relevant in synchronous messaging</param>
+        /// <returns></returns>
+        Task SendWithoutWaitingAsync(OutgoingMessage message, string correlationId = null);
+
         /// <summary>
         /// Closes down links, sessions and connections.
         /// </summary>

src/Helsenorge.Messaging/Abstractions/IMessagingNotification.cs

@@ -70,7 +70,27 @@ public interface IMessagingNotification
         /// The client can do any necessary set up it needs, like setting up CorrelationIds, etc.
         /// </summary>
         /// <param name="message">The actual message, contains the payload in addition to metadata.</param>
-        Task NotifySynchronousMessageReceivedStartingAsync(IncomingMessage message);
+        Task NotifySynchronousMessageReceivedStartingAsync(IncomingMessage message);
+        
+        /// <summary>
+        /// Called when the syncreply message is ready for processing.
+        /// </summary>
+        /// <param name="message">The actual message, contains the payload in addition to metadata.</param>
+        Task NotifySynchronousReplyMessageReceivedAsync(IncomingMessage message);
+
+        /// <summary>
+        /// Called to notify that processing of an syncreply message has started. The client can do any necessary
+        /// set up it needs, like setting up CorrelationIds, etc.
+        /// </summary>
+        /// <param name="listener">Reference to the listener invoking the callback.</param>
+        /// <param name="message">The actual message, contains the payload in addition to metadata.</param>
+        Task NotifySynchronousReplyMessageReceivedStartingAsync(MessageListener listener, IncomingMessage message);
+
+        /// <summary>
+        /// Called when the syncreply message has been successfully processed.
+        /// </summary>
+        /// <param name="message">The actual message, contains the payload in addition to metadata.</param>
+        Task NotifySynchronousReplyMessageReceivedCompletedAsync(IncomingMessage message);
 
         /// <summary>
         /// Called to notifiy an unhandled exception has occurred. The client need to assert what it needs to do.

src/Helsenorge.Messaging/Abstractions/IMessagingServer.cs

@@ -1,7 +1,7 @@
-/* 
+/*
  * Copyright (c) 2020-2023, Norsk Helsenett SF and contributors
  * See the file CONTRIBUTORS for details.
- * 
+ *
  * This file is licensed under the MIT license
  * available at https://raw.githubusercontent.com/helsenorge/Helsenorge.Messaging/master/LICENSE
  */
@@ -22,6 +22,7 @@ public interface IMessagingServer
         /// Start message processing
         /// </summary>
         Task StartAsync();
+
         /// <summary>
         /// Terminate message processing
         /// </summary>
@@ -38,17 +39,20 @@ public interface IMessagingServer
         /// </summary>
         /// <param name="action">The delegate that should be called</param>
         void RegisterAsynchronousMessageReceivedStartingCallback(Action<MessageListener, IncomingMessage> action);
+
         /// <summary>
         /// Registers a delegate that should be called as we start processing a message
         /// </summary>
         /// <param name="action">The delegate that should be called</param>
-        void RegisterAsynchronousMessageReceivedStartingCallbackAsync(Func<MessageListener, IncomingMessage, Task> action);
+        void RegisterAsynchronousMessageReceivedStartingCallbackAsync(
+            Func<MessageListener, IncomingMessage, Task> action);
 
         /// <summary>
         /// Registers a delegate that should be called when we have enough information to process the message. This is where the main processing logic hooks in.
         /// </summary>
         /// <param name="action">The delegate that should be called</param>
         void RegisterAsynchronousMessageReceivedCallback(Action<IncomingMessage> action);
+
         /// <summary>
         /// Registers a delegate that should be called when we have enough information to process the message. This is where the main processing logic hooks in.
         /// </summary>
@@ -60,6 +64,7 @@ public interface IMessagingServer
         /// </summary>
         /// <param name="action">The delegate that should be called</param>
         void RegisterAsynchronousMessageReceivedCompletedCallback(Action<IncomingMessage> action);
+
         /// <summary>
         /// Registers a delegate that should be called when we are finished processing the message.
         /// </summary>
@@ -71,6 +76,7 @@ public interface IMessagingServer
         /// </summary>
         /// <param name="action">The delegate that should be called</param>
         void RegisterSynchronousMessageReceivedStartingCallback(Action<IncomingMessage> action);
+
         /// <summary>
         /// Registers a delegate that should be called as we start processing a message
         /// </summary>
@@ -82,6 +88,7 @@ public interface IMessagingServer
         /// </summary>
         /// <param name="action">The delegate that should be called</param>
         void RegisterSynchronousMessageReceivedCallback(Func<IncomingMessage, XDocument> action);
+
         /// <summary>
         /// Registers a delegate that should be called when we have enough information to process the message. This is where the main processing logic hooks in.
         /// </summary>
@@ -93,17 +100,56 @@ public interface IMessagingServer
         /// </summary>
         /// <param name="action">The delegate that should be called</param>
         void RegisterSynchronousMessageReceivedCompletedCallback(Action<IncomingMessage> action);
+
         /// <summary>
         /// Registers a delegate that should be called when we are finished processing the message.
         /// </summary>
         /// <param name="action">The delegate that should be called</param>
         void RegisterSynchronousMessageReceivedCompletedCallbackAsync(Func<IncomingMessage, Task> action);
 
+        /// <summary>
+        /// Registers a delegate that should be called as we start processing a message
+        /// </summary>
+        /// <param name="action">The delegate that should be called</param>
+        void RegisterSynchronousReplyMessageReceivedStartingCallback(Action<MessageListener, IncomingMessage> action);
+
+        /// <summary>
+        /// Registers a delegate that should be called as we start processing a message
+        /// </summary>
+        /// <param name="action">The delegate that should be called</param>
+        void RegisterSynchronousReplyMessageReceivedStartingCallbackAsync(
+            Func<MessageListener, IncomingMessage, Task> action);
+
+        /// <summary>
+        /// Registers a delegate that should be called when we have enough information to process the message. This is where the main processing logic hooks in.
+        /// </summary>
+        /// <param name="action">The delegate that should be called</param>
+        void RegisterSynchronousReplyMessageReceivedCallback(Action<IncomingMessage> action);
+
+        /// <summary>
+        /// Registers a delegate that should be called when we have enough information to process the message. This is where the main processing logic hooks in.
+        /// </summary>
+        /// <param name="action">The delegate that should be called</param>
+        void RegisterSynchronousReplyMessageReceivedCallbackAsync(Func<IncomingMessage, Task> action);
+
+        /// <summary>
+        /// Registers a delegate that should be called when we are finished processing the message.
+        /// </summary>
+        /// <param name="action">The delegate that should be called</param>
+        void RegisterSynchronousReplyMessageReceivedCompletedCallback(Action<IncomingMessage> action);
+
+        /// <summary>
+        /// Registers a delegate that should be called when we are finished processing the message.
+        /// </summary>
+        /// <param name="action">The delegate that should be called</param>
+        void RegisterSynchronousReplyMessageReceivedCompletedCallbackAsync(Func<IncomingMessage, Task> action);
+
         /// <summary>
         /// Registers a delegate that should be called when we receive an error message
         /// </summary>
         /// <param name="action">The delegate that should be called</param>
         void RegisterErrorMessageReceivedCallback(Action<IAmqpMessage> action);
+
         /// <summary>
         /// Registers a delegate that should be called when we receive an error message
         /// </summary>
@@ -115,6 +161,7 @@ public interface IMessagingServer
         /// </summary>
         /// <param name="action">The delegate that should be called</param>
         void RegisterHandledExceptionCallback(Action<IAmqpMessage, Exception> action);
+
         /// <summary>
         /// Registers a delegate that should be called when we have an handled exception
         /// </summary>
@@ -126,10 +173,11 @@ public interface IMessagingServer
         /// </summary>
         /// <param name="action">The delegate that should be called</param>
         void RegisterUnhandledExceptionCallback(Action<IAmqpMessage, Exception> action);
+
         /// <summary>
         /// Registers a delegate that should be called when we have an unhandled exception
         /// </summary>
         /// <param name="action">The delegate that should be called</param>
         void RegisterUnhandledExceptionCallbackAsync(Func<IAmqpMessage, Exception, Task> action);
     }
-}
+}

src/Helsenorge.Messaging/Abstractions/MessageProtection.cs

@@ -9,7 +9,6 @@
 using System;
 using System.IO;
 using System.Security.Cryptography.X509Certificates;
-using System.Xml.Linq;
 
 namespace Helsenorge.Messaging.Abstractions
 {

src/Helsenorge.Messaging/Abstractions/MessagingException.cs

@@ -8,7 +8,6 @@
 
 using System;
 using System.Diagnostics.CodeAnalysis;
-using System.Runtime.Serialization;
 using Microsoft.Extensions.Logging;
 
 namespace Helsenorge.Messaging.Abstractions

src/Helsenorge.Messaging/Abstractions/OutgoingMessage.cs

@@ -6,7 +6,6 @@
  * available at https://raw.githubusercontent.com/helsenorge/Helsenorge.Messaging/master/LICENSE
  */
 
-using System;
 using System.Xml.Linq;
 
 namespace Helsenorge.Messaging.Abstractions

src/Helsenorge.Messaging/Amqp/AmqpCore.cs

@@ -260,7 +260,7 @@ private void ExtractAndAddMetaDataPropertiesIfEnabled(IAmqpMessage amqpMessage,
                     amqpMessage.SetApplicationPropertyValue(metaDataProperty.Key, metaDataProperty.Value);
                 }
                 sw.Stop();
-                logger.LogInformation($"After-AddingPayloadMetadata: {amqpMessage.MessageFunction} FromHerId: {amqpMessage.FromHerId} ToHerId: {amqpMessage.ToHerId}  MessageId: {amqpMessage.MessageId} Elapsed:{sw.ElapsedMilliseconds}");
+                logger.LogInformation($"After-AddingPayloadMetadata: {amqpMessage.MessageFunction} FromHerId: {amqpMessage.FromHerId} ToHerId: {amqpMessage.ToHerId}  MessageId: {amqpMessage.MessageId} Elapsed: {sw.ElapsedMilliseconds} ms");
             }
             catch (Exception e)
             {

src/Helsenorge.Messaging/Amqp/AmqpReceiver.cs

@@ -8,7 +8,6 @@
 
 using System;
 using System.Diagnostics.CodeAnalysis;
-using System.Threading;
 using System.Threading.Tasks;
 using Amqp;
 using Helsenorge.Messaging.Abstractions;

src/Helsenorge.Messaging/Amqp/Receivers/MessageListener.cs

@@ -67,14 +67,17 @@ protected virtual string GetQueueName()
             {
                 QueueType.Asynchronous => _queueNames.Async,
                 QueueType.Synchronous => _queueNames.Sync,
+                QueueType.SynchronousReply => _queueNames.SyncReply,
                 QueueType.Error => _queueNames.Error,
                 _ => throw new UnknownQueueTypeException(QueueType)
             };
         }
+
         /// <summary>
         /// Specifies what type of queue this listener is processing
         /// </summary>
         protected abstract QueueType QueueType { get; }
+
         /// <summary>
         /// Reference to service bus related setting
         /// </summary>
@@ -84,6 +87,7 @@ protected virtual string GetQueueName()
         /// Gets a reference to the server
         /// </summary>
         protected IMessagingNotification MessagingNotification { get; }
+
         /// <summary>
         /// Gets the logger used for diagnostics purposes
         /// </summary>
@@ -190,6 +194,7 @@ public async Task<IncomingMessage> ReadAndProcessMessageAsync(bool alwaysRemoveM
             }
             return await HandleRawMessageAsync(message, alwaysRemoveMessage).ConfigureAwait(false);
         }
+
         private async Task<IncomingMessage> HandleRawMessageAsync(IAmqpMessage message, bool alwaysRemoveMessage)
         {
             if (message == null) return null;
@@ -223,9 +228,10 @@ private async Task<IncomingMessage> HandleRawMessageAsync(IAmqpMessage message,
                     DeliveryCount = message.DeliveryCount,
                     LockedUntil = message.LockedUntil,
                 };
+               
                 await NotifyMessageProcessingStartedAsync(this, incomingMessage).ConfigureAwait(false);
 
-                SetCorrelationIdAction?.Invoke(incomingMessage.MessageId);
+                SetCorrelationIdAction?.Invoke(incomingMessage.CorrelationId);
 
                 Logger.LogStartReceive(QueueType, incomingMessage, $"Message received from host and queue: {AmqpCore.HostnameAndPath}/{queueName}");
 

src/Helsenorge.Messaging/Amqp/Receivers/SenderHerIdMismatchException.cs

@@ -7,7 +7,6 @@
  */
 
 using System;
-using System.Runtime.Serialization;
 
 namespace Helsenorge.Messaging.Amqp.Receivers
 {

src/Helsenorge.Messaging/Amqp/Receivers/SynchronousReplyListener.cs

@@ -48,6 +48,7 @@ protected override Task NotifyMessageProcessingStartedAsync(MessageListener list
             // Not relevant for this implementation
             return Task.CompletedTask;
         }
+
         /// <summary>
         /// Called when message processing is complete
         /// </summary>
@@ -58,6 +59,7 @@ protected override Task NotifyMessageProcessingCompletedAsync(IncomingMessage me
             // Not relevant for this implementation
             return Task.CompletedTask;
         }
+
         /// <summary>
         /// Called to process message
         /// </summary>