-
Notifications
You must be signed in to change notification settings - Fork 9
/
MultiOutputService.cs
54 lines (47 loc) · 1.85 KB
/
MultiOutputService.cs
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
using System;
using System.Collections.Generic;
using System.Linq;
using System.Runtime.CompilerServices;
using System.Threading;
using System.Threading.Tasks;
using ConsumeAndMultiOutputPublisherWithRabbitMQ.Model;
using Motor.Extensions.Hosting.Abstractions;
using Motor.Extensions.Hosting.CloudEvents;
namespace ConsumeAndMultiOutputPublisherWithRabbitMQ;
public class MultiOutputService : IMultiOutputService<InputMessage, OutputMessage>
{
// Handle incoming messages
public async IAsyncEnumerable<MotorCloudEvent<OutputMessage>> ConvertMessageAsync(
MotorCloudEvent<InputMessage> inputEvent, [EnumeratorCancellation] CancellationToken token = default)
{
// Get the input message from the cloud event
var input = inputEvent.TypedData;
// Do your magic here .....
var output = MagicFuncAsync(input);
// Create a new cloud event from your output message which is automatically published and return a new task.
await foreach (var outputMessage in output.WithCancellation(token))
{
yield return inputEvent.CreateNew(outputMessage);
}
}
private static async IAsyncEnumerable<OutputMessage> MagicFuncAsync(InputMessage input)
{
if (string.IsNullOrEmpty(input.FancyText))
{
// Reject message in RabbitMQ queue (Any ArgumentException can be used to reject to messages.).
throw new ArgumentNullException("FancyText is empty");
}
// Magic async function.
await Task.Delay(10).ConfigureAwait(false);
yield return new()
{
NotSoFancyText = input.FancyText.Reverse().ToString(),
NotSoFancyNumber = input.FancyNumber * -1,
};
yield return new()
{
NotSoFancyText = input.FancyText,
NotSoFancyNumber = input.FancyNumber * -2,
};
}
}