Skip to content

Commit 82cbb73

Browse files
authored
Release/0.1.24 (#209) (#213)
+semver: patch * adding new fields to messages for external app executions * added forgot licence header * added export request to export cmplete message event * Update ExportRequestEvent and update dependencies (#210) * Update dependencies * Remove ExportRequestType enum and replace with a list of plug-in assembly names * Update dependencies decisions * Include taskId in WorkflowRequestEvent to support multiple external app executions --------- Signed-off-by: Neil South <[email protected]> Signed-off-by: Victor Chang <[email protected]>
1 parent 31afb2f commit 82cbb73

10 files changed

+68
-49
lines changed

doc/dependency_decisions.yml

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -789,4 +789,4 @@
789789
:why: MIT ( https://licenses.nuget.org/MIT)
790790
:versions:
791791
- 2.5.0
792-
:when: 2022-08-16 21:40:32.294717110 Z
792+
:when: 2022-08-16 21:40:32.294717110 Z

src/Messaging/API/IMessageBrokerSubscriberService.cs

Lines changed: 0 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -31,30 +31,6 @@ public interface IMessageBrokerSubscriberService : IDisposable
3131
/// </summary>
3232
string Name { get; }
3333

34-
/// <summary>
35-
/// Subscribe to a message topic & queue and executes <c>messageReceivedCallback</c> for every message that is received.
36-
/// Either provide a topic, a queue or both.
37-
/// A queue is generated if the name of the queue is not provided.
38-
/// </summary>
39-
/// <param name="topic">Topic/routing key to bind to</param>
40-
/// <param name="queue">Name of the queue to consume</param>
41-
/// <param name="messageReceivedCallback">Action to be performed when message is received</param>
42-
/// <param name="prefetchCount">Number of unacknowledged messages to receive at once. Defaults to 0.</param>
43-
[Obsolete("This method is obsolete, use SubscribeAsync instead")]
44-
void Subscribe(string topic, string queue, Action<MessageReceivedEventArgs> messageReceivedCallback, ushort prefetchCount = 0);
45-
46-
/// <summary>
47-
/// Subscribe to a message topic & queue and executes <c>messageReceivedCallback</c> for every message that is received.
48-
/// Either provide a topic, a queue or both.
49-
/// A queue is generated if the name of the queue is not provided.
50-
/// </summary>
51-
/// <param name="topics">Topics/routing keys to bind to</param>
52-
/// <param name="queue">Name of the queue to consume</param>
53-
/// <param name="messageReceivedCallback">Action to be performed when message is received</param>
54-
/// <param name="prefetchCount">Number of unacknowledged messages to receive at once. Defaults to 0.</param>
55-
[Obsolete("This method is obsolete, use SubscribeAsync instead")]
56-
void Subscribe(string[] topics, string queue, Action<MessageReceivedEventArgs> messageReceivedCallback, ushort prefetchCount = 0);
57-
5834
/// <summary>
5935
/// Subscribe to a message topic & queue and executes <c>messageReceivedCallback</c> asynchronously for every message that is received.
6036
/// Either provide a topic, a queue or both.
Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
/*
2+
* Copyright 2023 MONAI Consortium
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
using System.Runtime.Serialization;
18+
19+
namespace Monai.Deploy.Messaging.Common
20+
{
21+
public class ServiceException : Exception
22+
{
23+
public ServiceException()
24+
{
25+
}
26+
27+
public ServiceException(string? message) : base(message)
28+
{
29+
}
30+
31+
public ServiceException(string? message, Exception? innerException) : base(message, innerException)
32+
{
33+
}
34+
35+
protected ServiceException(SerializationInfo info, StreamingContext context) : base(info, context)
36+
{
37+
}
38+
}
39+
}

src/Messaging/Events/ExportRequestEvent.cs

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -77,17 +77,17 @@ public class ExportRequestEvent : EventBase
7777
public List<string> ErrorMessages { get; private set; }
7878

7979
/// <summary>
80-
/// Gets or set the ExportRequest type.
81-
/// For standard exports this will be ExportRequestType.None
82-
/// but for exports to external apps this will be ExportRequestType.ExternalProcessing
80+
/// A list of data output plug-in type names to be executed by the export services.
81+
/// Each string must be a fully-qualified type name.
82+
/// E.g. <code>MyCompnay.MyProject.MyNamepsace.MyPlugin, MyCompnay.MyProject.MyNamepsace</code> where
83+
/// <code>MyCompnay.MyProject.MyNamepsace</code> is the name of the assembly (DLL).
8384
/// </summary>
84-
[JsonProperty(PropertyName = "export_request")]
85-
[Required]
86-
public ExportRequestType ExportRequest { get; set; } = default!;
85+
public List<string> PluginAssemblies { get; private set; }
8786

8887
public ExportRequestEvent()
8988
{
9089
ErrorMessages = new List<string>();
90+
PluginAssemblies = new List<string>();
9191
}
9292

9393
public void AddErrorMessages(IList<string> errorMessages)

src/Messaging/Events/WorkflowRequestEvent.cs

Lines changed: 8 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -89,6 +89,14 @@ public class WorkflowRequestEvent : EventBase
8989
[JsonProperty(PropertyName = "workflow_instance_id")]
9090
public string? WorkflowInstanceId { get; set; } = default;
9191

92+
/// <summary>
93+
/// Sets the task ID for the Workflow Manager.
94+
/// This is only applicable to resume events (after external app executions)
95+
/// In standard workflows this will not be set
96+
/// </summary>
97+
[JsonProperty(PropertyName = "task_id")]
98+
public string? TaskId { get; set; } = default;
99+
92100
/// <summary>
93101
/// Gets or sets a list of files and metadata files in this request.
94102
/// </summary>

src/Messaging/Tests/IServiceCollectionExtensionsTests.cs

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -203,7 +203,10 @@ internal class GoodSubscriberService : IMessageBrokerSubscriberService
203203
{
204204
public string Name => throw new NotImplementedException();
205205

206+
#pragma warning disable CS0067 // The event 'GoodSubscriberService.OnConnectionError' is never used
207+
// event used by users of this library
206208
public event ConnectionErrorHandler? OnConnectionError;
209+
#pragma warning restore CS0067 // The event 'GoodSubscriberService.OnConnectionError' is never used
207210

208211
public void Acknowledge(MessageBase message) => throw new NotImplementedException();
209212

src/Messaging/Tests/WorkflowRequestMessageTest.cs

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -36,7 +36,9 @@ public void ConvertsJsonMessageToMessage()
3636
FileCount = 10,
3737
PayloadId = Guid.NewGuid(),
3838
Timestamp = DateTime.Now,
39-
Workflows = new List<string> { Guid.NewGuid().ToString() }
39+
Workflows = new List<string> { Guid.NewGuid().ToString() },
40+
WorkflowInstanceId = Guid.NewGuid().ToString(),
41+
TaskId = Guid.NewGuid().ToString(),
4042
};
4143

4244
var files = new List<BlockStorageInfo>()

src/Plugins/RabbitMQ/Subscriber/RabbitMqMessageSubscriberService.cs

Lines changed: 2 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -109,7 +109,7 @@ private void CreateChannel()
109109
.Execute(() =>
110110
{
111111
_logger.ConnectingToRabbitMQ(Name, _endpoint, _virtualHost);
112-
_channel = _rabbitMqConnectionFactory.CreateChannel(ChannelType.Subscriber, _endpoint, _username, _password, _virtualHost, _useSSL, _portNumber);
112+
_channel = _rabbitMqConnectionFactory.CreateChannel(ChannelType.Subscriber, _endpoint, _username, _password, _virtualHost, _useSSL, _portNumber) ?? throw new ServiceException("Failed to create a new channel to RabbitMQ");
113113
_channel.ExchangeDeclare(_exchange, ExchangeType.Topic, durable: true, autoDelete: false);
114114
_channel.ExchangeDeclare(_deadLetterExchange, ExchangeType.Topic, durable: true, autoDelete: false);
115115
_channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
@@ -162,15 +162,6 @@ internal static void ValidateConfiguration(Dictionary<string, string> configurat
162162
throw new ConfigurationException($"{ConfigurationKeys.SubscriberServiceName} has int values of less than 1");
163163
}
164164
}
165-
[Obsolete("This method is obsolete, use SubscribeAsync instead")]
166-
public void Subscribe(string topic, string queue, Action<MessageReceivedEventArgs> messageReceivedCallback, ushort prefetchCount = 0)
167-
=> Subscribe(new string[] { topic }, queue, messageReceivedCallback, prefetchCount);
168-
169-
[Obsolete("This method is obsolete, use SubscribeAsync instead")]
170-
public void Subscribe(string[] topics, string queue, Action<MessageReceivedEventArgs> messageReceivedCallback, ushort prefetchCount = 0)
171-
{
172-
SubscribeAsync(topics, queue, new Func<MessageReceivedEventArgs, Task>((args) => { messageReceivedCallback.Invoke(args); return Task.FromResult(0); }));
173-
}
174165

175166
public void SubscribeAsync(string topic, string queue, Func<MessageReceivedEventArgs, Task> messageReceivedCallback, ushort prefetchCount = 0)
176167
=> SubscribeAsync(new string[] { topic }, queue, messageReceivedCallback, prefetchCount);
@@ -212,7 +203,7 @@ private EventingBasicConsumer CreateConsumer(Func<MessageReceivedEventArgs, Task
212203
_logger.InvalidMessage(queueDeclareResult.QueueName, eventArgs.RoutingKey, eventArgs.BasicProperties.MessageId, ex);
213204

214205
_logger.SendingNAcknowledgement(eventArgs.BasicProperties.MessageId);
215-
_channel.BasicNack(eventArgs.DeliveryTag, multiple: false, requeue: false);
206+
_channel!.BasicNack(eventArgs.DeliveryTag, multiple: false, requeue: false);
216207
_logger.NAcknowledgementSent(eventArgs.BasicProperties.MessageId, false);
217208
return;
218209
}
@@ -367,7 +358,6 @@ private static MessageReceivedEventArgs CreateMessage(string topic, BasicDeliver
367358
{
368359
Guard.Against.NullOrWhiteSpace(topic, nameof(topic));
369360
Guard.Against.Null(eventArgs, nameof(eventArgs));
370-
371361
Guard.Against.Null(eventArgs.Body, nameof(eventArgs.Body));
372362
Guard.Against.Null(eventArgs.BasicProperties, nameof(eventArgs.BasicProperties));
373363
Guard.Against.Null(eventArgs.BasicProperties.MessageId, nameof(eventArgs.BasicProperties.MessageId));

src/Plugins/RabbitMQ/Tests/Unit/RabbitMqMessageSubscriberServiceTest.cs

Lines changed: 6 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -133,7 +133,7 @@ public void SubscribesToATopic()
133133

134134
var service = new RabbitMQMessageSubscriberService(_options, _logger.Object, _connectionFactory.Object);
135135

136-
service.Subscribe("topic", "queue", (args) =>
136+
service.SubscribeAsync("topic", "queue", async (args) =>
137137
{
138138
Assert.Equal(message.ApplicationId, args.Message.ApplicationId);
139139
Assert.Equal(message.ContentType, args.Message.ContentType);
@@ -143,6 +143,7 @@ public void SubscribesToATopic()
143143
Assert.Equal("topic", args.Message.MessageDescription);
144144
Assert.Equal(message.MessageId, args.Message.MessageId);
145145
Assert.Equal(message.Body, args.Message.Body);
146+
await Task.CompletedTask.ConfigureAwait(false);
146147
});
147148

148149
service.SubscribeAsync("topic", "queue", async (args) =>
@@ -240,7 +241,7 @@ public void SubscribesToATopicAndDeadLetterQueueIsDown()
240241

241242
var service = new RabbitMQMessageSubscriberService(_options, _logger.Object, _connectionFactory.Object);
242243

243-
service.Subscribe("topic", "queue", (args) =>
244+
service.SubscribeAsync("topic", "queue", async (args) =>
244245
{
245246
Assert.Equal(message.ApplicationId, args.Message.ApplicationId);
246247
Assert.Equal(message.ContentType, args.Message.ContentType);
@@ -250,6 +251,7 @@ public void SubscribesToATopicAndDeadLetterQueueIsDown()
250251
Assert.Equal("topic", args.Message.MessageDescription);
251252
Assert.Equal(message.MessageId, args.Message.MessageId);
252253
Assert.Equal(message.Body, args.Message.Body);
254+
await Task.CompletedTask;
253255
});
254256

255257
service.SubscribeAsync("topic", "queue", async (args) =>
@@ -349,7 +351,7 @@ public void SubscribesToATopicAndDeadLetterQueueSubscriptionFailsWithGenericExce
349351

350352
var act = () =>
351353
{
352-
service.Subscribe("topic", "queue", (args) =>
354+
service.SubscribeAsync("topic", "queue", async (args) =>
353355
{
354356
Assert.Equal(message.ApplicationId, args.Message.ApplicationId);
355357
Assert.Equal(message.ContentType, args.Message.ContentType);
@@ -359,7 +361,7 @@ public void SubscribesToATopicAndDeadLetterQueueSubscriptionFailsWithGenericExce
359361
Assert.Equal("topic", args.Message.MessageDescription);
360362
Assert.Equal(message.MessageId, args.Message.MessageId);
361363
Assert.Equal(message.Body, args.Message.Body);
362-
364+
await Task.CompletedTask;
363365
});
364366
};
365367
Assert.Throws<OperationInterruptedException>(act);

third-party-licenses.md

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -17166,4 +17166,3 @@ Data pulled from spdx/license-list-data on February 9, 2023.
1716617166
```
1716717167

1716817168
</details>
17169-

0 commit comments

Comments
 (0)