Jpleger/sqs

.github/pull_request_template.md

@@ -1,8 +1,3 @@
-<!--
-SPDX-FileCopyrightText: � 2021-2022 MONAI Consortium
-SPDX-License-Identifier: Apache License 2.0
--->
-
 Fixes # .
 
 ### Description
@@ -16,7 +11,5 @@ A few sentences describing the changes proposed in this pull request.
 - [ ] Non-breaking change (fix or new feature that would not break existing functionality).
 - [ ] Breaking change (fix or new feature that would cause existing functionality to change).
 - [ ] New tests added to cover the changes.
-- [ ] All tests passed locally by running `./src/run-tests-in-docker.sh`.
+- [ ] All tests passed locally.
 - [ ] [Documentation comments](https://docs.microsoft.com/en-us/dotnet/csharp/language-reference/language-specification/documentation-comments) included/updated.
-- [ ] User guide updated.
-- [ ] I have updated the changelog

.github/workflows/ci.yml

@@ -9,6 +9,11 @@ on:
 
   # Allows you to run this workflow manually from the Actions tab
   workflow_dispatch:
+    inputs:
+      nuget:
+        type: boolean
+        default: false
+        description: Publish to NuGet.org
 
 env:
   BUILD_CONFIG: "Release"
@@ -206,6 +211,7 @@ jobs:
           retention-days: 30
 
   publish:
+    name: Publish to GitHub Packages
     runs-on: ubuntu-latest
     needs: [build, unit-test]
     steps:
@@ -223,10 +229,11 @@ jobs:
           source-url: https://nuget.pkg.github.com/Project-MONAI/index.json
 
       - name: Publish to GitHub
-        run: dotnet nuget push ${{ steps.download.outputs.download-path }}/artifact/*.nupkg
+        run: dotnet nuget push ${{ steps.download.outputs.download-path }}/artifact/*.nupkg --skip-duplicate
 
   release:
-    if: ${{ contains(github.ref, 'refs/heads/main') ||contains(github.head_ref, 'release/') }}
+    name: Official Release to NuGet.org
+    if: ${{ github.event.inputs.nuget || contains(github.ref, 'refs/heads/release') }}
     runs-on: ubuntu-latest
     needs: [build, unit-test]
     env:
@@ -268,17 +275,17 @@ jobs:
           name: Release v${{ env.MAJORMINORPATCH }}
 
       - name: Publish release with GitReleaseManager
-        uses: gittools/actions/gitreleasemanager/[email protected]
         if: ${{ contains(github.ref, 'refs/heads/main') }}
+        uses: gittools/actions/gitreleasemanager/[email protected]
         with:
           token: ${{ secrets.GITHUB_TOKEN }}
           owner: ${{ steps.repo.outputs._0 }}
           repository: ${{ steps.repo.outputs._1 }}
           tagName: ${{ env.MAJORMINORPATCH }}
 
       - name: Close release with GitReleaseManager
-        uses: gittools/actions/gitreleasemanager/[email protected]
         if: ${{ contains(github.ref, 'refs/heads/main') }}
+        uses: gittools/actions/gitreleasemanager/[email protected]
         with:
           token: ${{ secrets.GITHUB_TOKEN }}
           owner: ${{ steps.repo.outputs._0 }}

GitVersion.yml

@@ -1,17 +1,57 @@
-# SPDX-FileCopyrightText: � 2022 MONAI Consortium
+# SPDX-FileCopyrightText: © 2022 MONAI Consortium
 # SPDX-License-Identifier: Apache License 2.0
 
 assembly-versioning-scheme: MajorMinorPatchTag
-mode: ContinuousDelivery
+mode: ContinuousDeployment
 branches:
   main:
     tag: ''
+    mode: ContinuousDelivery
+    increment: Patch
+    prevent-increment-of-merged-branch-version: true
+    track-merge-target: false
+    source-branches: [ 'release' ]
+    tracks-release-branches: false
+    is-release-branch: false
+    is-mainline: true
+    pre-release-weight: 55000
   release:
     tag: rc
+    regex: ^releases?[/-]
+    mode: ContinuousDeployment
+    increment: None
+    prevent-increment-of-merged-branch-version: true
+    track-merge-target: false
+    source-branches: [ 'main', 'release' ]
+    tracks-release-branches: false
+    is-release-branch: true
+    is-mainline: false
+    pre-release-weight: 30000
   feature:
     tag: alpha.{BranchName}
+    regex: ^features?[/-]
+    mode: ContinuousDeployment
+    increment: Inherit
+    prevent-increment-of-merged-branch-version: false
+    track-merge-target: false
+    source-branches: [ 'main', 'release', 'feature' ]
+    tracks-release-branches: false
+    is-release-branch: false
+    is-mainline: false
+    pre-release-weight: 30000
   pull-request:
     tag: pr
+    regex: ^(pull|pull\-requests|pr)[/-]
+    mode: ContinuousDeployment
+    increment: Inherit
+    prevent-increment-of-merged-branch-version: false
+    tag-number-pattern: '[/-](?<number>\d+)[-/]'
+    track-merge-target: false
+    source-branches: [ 'main', 'release', 'feature' ]
+    tracks-release-branches: false
+    is-release-branch: false
+    is-mainline: false
+    pre-release-weight: 30000
 
 ignore:
   sha: []

src/Messaging/API/IMessageBrokerPublisherService.cs

@@ -5,7 +5,7 @@
 
 namespace Monai.Deploy.Messaging
 {
-    public interface IMessageBrokerPublisherService
+    public interface IMessageBrokerPublisherService : IDisposable
     {
         /// <summary>
         /// Gets or sets the name of the storage service.

src/Messaging/API/IMessageBrokerSubscriberService.cs

@@ -6,33 +6,68 @@
 
 namespace Monai.Deploy.Messaging
 {
-    public interface IMessageBrokerSubscriberService
+    public interface IMessageBrokerSubscriberService : IDisposable
     {
         /// <summary>
         /// Gets or sets the name of the storage service.
         /// </summary>
         string Name { get; }
 
         /// <summary>
-        /// Subscribe to a message topic & queue.
+        /// Subscribe to a message topic & queue and executes <c>messageReceivedCallback</c> for every message that is received.
         /// Either provide a topic, a queue or both.
+        /// A queue is generated if the name of the queue is not provided.
         /// </summary>
-        /// <param name="topic">Name of the topic to subscribe to</param>
+        /// <param name="topic">Topic/routing key to bind to</param>
         /// <param name="queue">Name of the queue to consume</param>
         /// <param name="messageReceivedCallback">Action to be performed when message is received</param>
-        /// <param name="prefetchCount">Number of unacknowledged messages to receive at once.  Defaults to 0.</param>
+        /// <param name="prefetchCount">Number of unacknowledged messages to receive at once. Defaults to 0.</param>
         void Subscribe(string topic, string queue, Action<MessageReceivedEventArgs> messageReceivedCallback, ushort prefetchCount = 0);
 
+        /// <summary>
+        /// Subscribe to a message topic & queue and executes <c>messageReceivedCallback</c> for every message that is received.
+        /// Either provide a topic, a queue or both.
+        /// A queue is generated if the name of the queue is not provided.
+        /// </summary>
+        /// <param name="topics">Topics/routing keys to bind to</param>
+        /// <param name="queue">Name of the queue to consume</param>
+        /// <param name="messageReceivedCallback">Action to be performed when message is received</param>
+        /// <param name="prefetchCount">Number of unacknowledged messages to receive at once. Defaults to 0.</param>
+        void Subscribe(string[] topics, string queue, Action<MessageReceivedEventArgs> messageReceivedCallback, ushort prefetchCount = 0);
+
+        /// <summary>
+        /// Subscribe to a message topic & queue and executes <c>messageReceivedCallback</c> asynchronously for every message that is received.
+        /// Either provide a topic, a queue or both.
+        /// A queue is generated if the name of the queue is not provided.
+        /// </summary>
+        /// <param name="topic">Topic/routing key to bind to</param>
+        /// <param name="queue">Name of the queue to consume</param>
+        /// <param name="messageReceivedCallback"><see cref="Func{MessageReceivedEventArgs, Task}"/> to be performed when message is received</param>
+        /// <param name="prefetchCount">Number of unacknowledged messages to receive at once. Defaults to 0.</param>
+        void SubscribeAsync(string topic, string queue, Func<MessageReceivedEventArgs, Task> messageReceivedCallback, ushort prefetchCount = 0);
+
+        /// <summary>
+        /// Subscribe to a message topic & queue and executes <c>messageReceivedCallback</c> asynchronously for every message that is received.
+        /// Either provide a topic, a queue or both.
+        /// A queue is generated if the name of the queue is not provided.
+        /// </summary>
+        /// <param name="topics">Topics/routing keys to bind to</param>
+        /// <param name="queue">Name of the queue to consume</param>
+        /// <param name="messageReceivedCallback"><see cref="Func{MessageReceivedEventArgs, Task}"/> to be performed when message is received</param>
+        /// <param name="prefetchCount">Number of unacknowledged messages to receive at once. Defaults to 0.</param>
+        void SubscribeAsync(string[] topics, string queue, Func<MessageReceivedEventArgs, Task> messageReceivedCallback, ushort prefetchCount = 0);
+
         /// <summary>
         /// Acknowledge receiving of a message with the given token.
         /// </summary>
         /// <param name="message">Message to be acknowledged.</param>
         void Acknowledge(MessageBase message);
 
         /// <summary>
-        /// Rejects a messags.
+        /// Rejects a message.
         /// </summary>
         /// <param name="message">Message to be rejected.</param>
-        void Reject(MessageBase message);
+        /// <param name="requeue">Determines if the message should be requeued.</param>
+        void Reject(MessageBase message, bool requeue = true);
     }
 }

src/Messaging/Common/Credentials.cs

@@ -0,0 +1,38 @@
+// SPDX-FileCopyrightText: © 2022 MONAI Consortium
+// SPDX-License-Identifier: Apache License 2.0
+
+using System.ComponentModel.DataAnnotations;
+using Newtonsoft.Json;
+
+namespace Monai.Deploy.Messaging.Common
+{
+    public class Credentials
+    {
+        /// <summary>
+        /// Gets or sets the access key or user name of the credentials pair.
+        /// </summary>
+        [JsonProperty(PropertyName = "access_key")]
+        [Required]
+        public string AccessKey { get; set; }
+
+        /// <summary>
+        /// Gets or sets the access token or password of the credentials pair.
+        /// </summary>
+        [JsonProperty(PropertyName = "access_token")]
+        [Required]
+        public string AccessToken { get; set; }
+
+        /// <summary>
+        /// Gets or sets the session token of the credentials pair.
+        /// </summary>
+        [JsonProperty(PropertyName = "session_token")]
+        public string SessionToken { get; set; }
+
+        public Credentials()
+        {
+            AccessKey = string.Empty;
+            AccessToken = string.Empty;
+            SessionToken = string.Empty;
+        }
+    }
+}

src/Messaging/Common/Log.cs

@@ -30,10 +30,16 @@ public static partial class Log
         [LoggerMessage(EventId = 10006, Level = LogLevel.Information, Message = "Sending nack message {messageId} and requeuing.")]
         public static partial void SendingNAcknowledgement(this ILogger logger, string messageId);
 
-        [LoggerMessage(EventId = 10007, Level = LogLevel.Information, Message = "Nack message sent for message {messageId}.")]
-        public static partial void NAcknowledgementSent(this ILogger logger, string messageId);
+        [LoggerMessage(EventId = 10007, Level = LogLevel.Information, Message = "Nack message sent for message {messageId}, requeue={requeue}.")]
+        public static partial void NAcknowledgementSent(this ILogger logger, string messageId, bool requeue);
 
-        [LoggerMessage(EventId = 10008, Level = LogLevel.Information, Message = "Closing connection.")]
-        public static partial void ClosingConnection(this ILogger logger);
+        [LoggerMessage(EventId = 10008, Level = LogLevel.Information, Message = "Closing connections.")]
+        public static partial void ClosingConnections(this ILogger logger);
+
+        [LoggerMessage(EventId = 10009, Level = LogLevel.Error, Message = "Invalid or corrupted message received: Queue={queueName}, Topic={topic}, Message ID={messageId}.")]
+        public static partial void InvalidMessage(this ILogger logger, string queueName, string topic, string messageId, Exception ex);
+
+        [LoggerMessage(EventId = 10010, Level = LogLevel.Error, Message = "Exception not handled by the subscriber's callback function: Queue={queueName}, Topic={topic}, Message ID={messageId}.")]
+        public static partial void ErrorNotHandledByCallback(this ILogger logger, string queueName, string topic, string messageId, Exception ex);
     }
 }

src/Messaging/Common/MessageConversionException.cs

@@ -0,0 +1,27 @@
+// SPDX-FileCopyrightText: © 2022 MONAI Consortium
+// SPDX-License-Identifier: Apache License 2.0
+//
+using System.Runtime.Serialization;
+
+namespace Monai.Deploy.Messaging.Common
+{
+    [Serializable]
+    public class MessageConversionException : Exception
+    {
+        public MessageConversionException()
+        {
+        }
+
+        public MessageConversionException(string message) : base(message)
+        {
+        }
+
+        public MessageConversionException(string message, Exception innerException) : base(message, innerException)
+        {
+        }
+
+        protected MessageConversionException(SerializationInfo info, StreamingContext context) : base(info, context)
+        {
+        }
+    }
+}

src/Messaging/Common/MessageValidationException.cs

@@ -0,0 +1,31 @@
+// SPDX-FileCopyrightText: © 2022 MONAI Consortium
+// SPDX-License-Identifier: Apache License 2.0
+
+using System.ComponentModel.DataAnnotations;
+using System.Runtime.Serialization;
+
+namespace Monai.Deploy.Messaging.Common
+{
+    [Serializable]
+    public class MessageValidationException : Exception
+    {
+        public MessageValidationException(List<ValidationResult> errors)
+            : base(FormatMessage(errors))
+        {
+        }
+
+        protected MessageValidationException(SerializationInfo info, StreamingContext context) : base(info, context)
+        {
+        }
+
+        private static string FormatMessage(List<ValidationResult> errors)
+        {
+            if (errors is null || errors.Count == 0)
+            {
+                return "Invalid message.";
+            }
+
+            return $"Invalid message: {string.Join(',', errors.Select(p => $"{p.ErrorMessage} Path: {string.Join(',', p.MemberNames)}."))}";
+        }
+    }
+}