diff --git a/Brighter.sln b/Brighter.sln index 40872fc852..9b2a7ef52e 100644 --- a/Brighter.sln +++ b/Brighter.sln @@ -440,6 +440,10 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TaskStatusSender", "samples EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paramore.Brighter.RocketMQ.Tests", "tests\Paramore.Brighter.RocketMQ.Tests\Paramore.Brighter.RocketMQ.Tests\Paramore.Brighter.RocketMQ.Tests.csproj", "{9063F17B-5636-4AD5-999B-C894517DB5FD}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paramore.Brighter.MessagingGateway.Pulsar", "src\Paramore.Brighter.MessagingGateway.Pulsar\Paramore.Brighter.MessagingGateway.Pulsar.csproj", "{525E736D-A262-8E2E-0C85-1A9B1C9BD3F3}" +EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paramore.Brighter.MessagingGateway.Pulsar.Tests", "tests\Paramore.Brighter.MessagingGateway.Pulsar.Tests\Paramore.Brighter.MessagingGateway.Pulsar.Tests.csproj", "{44DBA505-FB28-428F-AD4F-3BBF3BD12DB3}" +EndProject Global GlobalSection(SolutionConfigurationPlatforms) = preSolution Debug|Any CPU = Debug|Any CPU @@ -2046,6 +2050,18 @@ Global {5D57A811-A3C2-42BA-A086-A5BFE8B590C7}.Release|Mixed Platforms.Build.0 = Release|Any CPU {5D57A811-A3C2-42BA-A086-A5BFE8B590C7}.Release|x86.ActiveCfg = Release|Any CPU {5D57A811-A3C2-42BA-A086-A5BFE8B590C7}.Release|x86.Build.0 = Release|Any CPU + {9D7DFB30-1E56-46C7-A56E-2FBC6C58EB96}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {9D7DFB30-1E56-46C7-A56E-2FBC6C58EB96}.Debug|Any CPU.Build.0 = Debug|Any CPU + {9D7DFB30-1E56-46C7-A56E-2FBC6C58EB96}.Debug|Mixed Platforms.ActiveCfg = Debug|Any CPU + {9D7DFB30-1E56-46C7-A56E-2FBC6C58EB96}.Debug|Mixed Platforms.Build.0 = Debug|Any CPU + {9D7DFB30-1E56-46C7-A56E-2FBC6C58EB96}.Debug|x86.ActiveCfg = Debug|Any CPU + {9D7DFB30-1E56-46C7-A56E-2FBC6C58EB96}.Debug|x86.Build.0 = Debug|Any CPU + {9D7DFB30-1E56-46C7-A56E-2FBC6C58EB96}.Release|Any CPU.ActiveCfg = Release|Any CPU + {9D7DFB30-1E56-46C7-A56E-2FBC6C58EB96}.Release|Any CPU.Build.0 = Release|Any CPU + {9D7DFB30-1E56-46C7-A56E-2FBC6C58EB96}.Release|Mixed Platforms.ActiveCfg = Release|Any CPU + {9D7DFB30-1E56-46C7-A56E-2FBC6C58EB96}.Release|Mixed Platforms.Build.0 = Release|Any CPU + {9D7DFB30-1E56-46C7-A56E-2FBC6C58EB96}.Release|x86.ActiveCfg = Release|Any CPU + {9D7DFB30-1E56-46C7-A56E-2FBC6C58EB96}.Release|x86.Build.0 = Release|Any CPU {022AD920-4E8D-4370-9C6D-CA4D8DA3DB6F}.Debug|Any CPU.ActiveCfg = Debug|Any CPU {022AD920-4E8D-4370-9C6D-CA4D8DA3DB6F}.Debug|Any CPU.Build.0 = Debug|Any CPU {022AD920-4E8D-4370-9C6D-CA4D8DA3DB6F}.Debug|Mixed Platforms.ActiveCfg = Debug|Any CPU @@ -2274,6 +2290,18 @@ Global {D530B147-067A-408D-BB1B-A4290324012F}.Release|Mixed Platforms.Build.0 = Release|Any CPU {D530B147-067A-408D-BB1B-A4290324012F}.Release|x86.ActiveCfg = Release|Any CPU {D530B147-067A-408D-BB1B-A4290324012F}.Release|x86.Build.0 = Release|Any CPU + {62D9AF6E-B671-4444-82A4-8416B8049E14}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {62D9AF6E-B671-4444-82A4-8416B8049E14}.Debug|Any CPU.Build.0 = Debug|Any CPU + {62D9AF6E-B671-4444-82A4-8416B8049E14}.Debug|Mixed Platforms.ActiveCfg = Debug|Any CPU + {62D9AF6E-B671-4444-82A4-8416B8049E14}.Debug|Mixed Platforms.Build.0 = Debug|Any CPU + {62D9AF6E-B671-4444-82A4-8416B8049E14}.Debug|x86.ActiveCfg = Debug|Any CPU + {62D9AF6E-B671-4444-82A4-8416B8049E14}.Debug|x86.Build.0 = Debug|Any CPU + {62D9AF6E-B671-4444-82A4-8416B8049E14}.Release|Any CPU.ActiveCfg = Release|Any CPU + {62D9AF6E-B671-4444-82A4-8416B8049E14}.Release|Any CPU.Build.0 = Release|Any CPU + {62D9AF6E-B671-4444-82A4-8416B8049E14}.Release|Mixed Platforms.ActiveCfg = Release|Any CPU + {62D9AF6E-B671-4444-82A4-8416B8049E14}.Release|Mixed Platforms.Build.0 = Release|Any CPU + {62D9AF6E-B671-4444-82A4-8416B8049E14}.Release|x86.ActiveCfg = Release|Any CPU + {62D9AF6E-B671-4444-82A4-8416B8049E14}.Release|x86.Build.0 = Release|Any CPU {7AA5B0BF-3520-45C4-9B8A-7F131EFDA227}.Debug|Any CPU.ActiveCfg = Debug|Any CPU {7AA5B0BF-3520-45C4-9B8A-7F131EFDA227}.Debug|Any CPU.Build.0 = Debug|Any CPU {7AA5B0BF-3520-45C4-9B8A-7F131EFDA227}.Debug|Mixed Platforms.ActiveCfg = Debug|Any CPU @@ -2406,18 +2434,6 @@ Global {3B6D084F-C034-49C6-A8C4-3C23DCC83CF2}.Release|Mixed Platforms.Build.0 = Release|Any CPU {3B6D084F-C034-49C6-A8C4-3C23DCC83CF2}.Release|x86.ActiveCfg = Release|Any CPU {3B6D084F-C034-49C6-A8C4-3C23DCC83CF2}.Release|x86.Build.0 = Release|Any CPU - {62D9AF6E-B671-4444-82A4-8416B8049E14}.Debug|Any CPU.ActiveCfg = Debug|Any CPU - {62D9AF6E-B671-4444-82A4-8416B8049E14}.Debug|Any CPU.Build.0 = Debug|Any CPU - {62D9AF6E-B671-4444-82A4-8416B8049E14}.Debug|Mixed Platforms.ActiveCfg = Debug|Any CPU - {62D9AF6E-B671-4444-82A4-8416B8049E14}.Debug|Mixed Platforms.Build.0 = Debug|Any CPU - {62D9AF6E-B671-4444-82A4-8416B8049E14}.Debug|x86.ActiveCfg = Debug|Any CPU - {62D9AF6E-B671-4444-82A4-8416B8049E14}.Debug|x86.Build.0 = Debug|Any CPU - {62D9AF6E-B671-4444-82A4-8416B8049E14}.Release|Any CPU.ActiveCfg = Release|Any CPU - {62D9AF6E-B671-4444-82A4-8416B8049E14}.Release|Any CPU.Build.0 = Release|Any CPU - {62D9AF6E-B671-4444-82A4-8416B8049E14}.Release|Mixed Platforms.ActiveCfg = Release|Any CPU - {62D9AF6E-B671-4444-82A4-8416B8049E14}.Release|Mixed Platforms.Build.0 = Release|Any CPU - {62D9AF6E-B671-4444-82A4-8416B8049E14}.Release|x86.ActiveCfg = Release|Any CPU - {62D9AF6E-B671-4444-82A4-8416B8049E14}.Release|x86.Build.0 = Release|Any CPU {4EA5F196-DDA8-4941-956B-D413B03051C9}.Debug|Any CPU.ActiveCfg = Debug|Any CPU {4EA5F196-DDA8-4941-956B-D413B03051C9}.Debug|Any CPU.Build.0 = Debug|Any CPU {4EA5F196-DDA8-4941-956B-D413B03051C9}.Debug|Mixed Platforms.ActiveCfg = Debug|Any CPU @@ -2490,8 +2506,6 @@ Global {0E6A0B80-58B7-4AA2-9E40-EE0AA5D4719E}.Release|Mixed Platforms.Build.0 = Release|Any CPU {0E6A0B80-58B7-4AA2-9E40-EE0AA5D4719E}.Release|x86.ActiveCfg = Release|Any CPU {0E6A0B80-58B7-4AA2-9E40-EE0AA5D4719E}.Release|x86.Build.0 = Release|Any CPU - {F00B137A-C187-4C33-A37B-22AD40B71600}.Debug|Any CPU.ActiveCfg = Debug|Any CPU - {F00B137A-C187-4C33-A37B-22AD40B71600}.Debug|Any CPU.Build.0 = Debug|Any CPU {79CA356E-B08C-4D88-88C9-653EC8D8BF4D}.Debug|Any CPU.ActiveCfg = Debug|Any CPU {79CA356E-B08C-4D88-88C9-653EC8D8BF4D}.Debug|Any CPU.Build.0 = Debug|Any CPU {79CA356E-B08C-4D88-88C9-653EC8D8BF4D}.Debug|Mixed Platforms.ActiveCfg = Debug|Any CPU @@ -2528,8 +2542,6 @@ Global {24360989-A956-45E9-BF07-7FD9E7553C7D}.Release|Mixed Platforms.Build.0 = Release|Any CPU {24360989-A956-45E9-BF07-7FD9E7553C7D}.Release|x86.ActiveCfg = Release|Any CPU {24360989-A956-45E9-BF07-7FD9E7553C7D}.Release|x86.Build.0 = Release|Any CPU - {9D7DFB30-1E56-46C7-A56E-2FBC6C58EB96}.Debug|Any CPU.ActiveCfg = Debug|Any CPU - {9D7DFB30-1E56-46C7-A56E-2FBC6C58EB96}.Debug|Any CPU.Build.0 = Debug|Any CPU {9063F17B-5636-4AD5-999B-C894517DB5FD}.Debug|Any CPU.ActiveCfg = Debug|Any CPU {9063F17B-5636-4AD5-999B-C894517DB5FD}.Debug|Any CPU.Build.0 = Debug|Any CPU {9063F17B-5636-4AD5-999B-C894517DB5FD}.Debug|Mixed Platforms.ActiveCfg = Debug|Any CPU @@ -2542,6 +2554,30 @@ Global {9063F17B-5636-4AD5-999B-C894517DB5FD}.Release|Mixed Platforms.Build.0 = Release|Any CPU {9063F17B-5636-4AD5-999B-C894517DB5FD}.Release|x86.ActiveCfg = Release|Any CPU {9063F17B-5636-4AD5-999B-C894517DB5FD}.Release|x86.Build.0 = Release|Any CPU + {525E736D-A262-8E2E-0C85-1A9B1C9BD3F3}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {525E736D-A262-8E2E-0C85-1A9B1C9BD3F3}.Debug|Any CPU.Build.0 = Debug|Any CPU + {525E736D-A262-8E2E-0C85-1A9B1C9BD3F3}.Debug|Mixed Platforms.ActiveCfg = Debug|Any CPU + {525E736D-A262-8E2E-0C85-1A9B1C9BD3F3}.Debug|Mixed Platforms.Build.0 = Debug|Any CPU + {525E736D-A262-8E2E-0C85-1A9B1C9BD3F3}.Debug|x86.ActiveCfg = Debug|Any CPU + {525E736D-A262-8E2E-0C85-1A9B1C9BD3F3}.Debug|x86.Build.0 = Debug|Any CPU + {525E736D-A262-8E2E-0C85-1A9B1C9BD3F3}.Release|Any CPU.ActiveCfg = Release|Any CPU + {525E736D-A262-8E2E-0C85-1A9B1C9BD3F3}.Release|Any CPU.Build.0 = Release|Any CPU + {525E736D-A262-8E2E-0C85-1A9B1C9BD3F3}.Release|Mixed Platforms.ActiveCfg = Release|Any CPU + {525E736D-A262-8E2E-0C85-1A9B1C9BD3F3}.Release|Mixed Platforms.Build.0 = Release|Any CPU + {525E736D-A262-8E2E-0C85-1A9B1C9BD3F3}.Release|x86.ActiveCfg = Release|Any CPU + {525E736D-A262-8E2E-0C85-1A9B1C9BD3F3}.Release|x86.Build.0 = Release|Any CPU + {44DBA505-FB28-428F-AD4F-3BBF3BD12DB3}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {44DBA505-FB28-428F-AD4F-3BBF3BD12DB3}.Debug|Any CPU.Build.0 = Debug|Any CPU + {44DBA505-FB28-428F-AD4F-3BBF3BD12DB3}.Debug|Mixed Platforms.ActiveCfg = Debug|Any CPU + {44DBA505-FB28-428F-AD4F-3BBF3BD12DB3}.Debug|Mixed Platforms.Build.0 = Debug|Any CPU + {44DBA505-FB28-428F-AD4F-3BBF3BD12DB3}.Debug|x86.ActiveCfg = Debug|Any CPU + {44DBA505-FB28-428F-AD4F-3BBF3BD12DB3}.Debug|x86.Build.0 = Debug|Any CPU + {44DBA505-FB28-428F-AD4F-3BBF3BD12DB3}.Release|Any CPU.ActiveCfg = Release|Any CPU + {44DBA505-FB28-428F-AD4F-3BBF3BD12DB3}.Release|Any CPU.Build.0 = Release|Any CPU + {44DBA505-FB28-428F-AD4F-3BBF3BD12DB3}.Release|Mixed Platforms.ActiveCfg = Release|Any CPU + {44DBA505-FB28-428F-AD4F-3BBF3BD12DB3}.Release|Mixed Platforms.Build.0 = Release|Any CPU + {44DBA505-FB28-428F-AD4F-3BBF3BD12DB3}.Release|x86.ActiveCfg = Release|Any CPU + {44DBA505-FB28-428F-AD4F-3BBF3BD12DB3}.Release|x86.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -2681,6 +2717,7 @@ Global {43F23C67-6C9D-44BB-B3A8-9313E570879D} = {9347BD22-1E8A-4A79-9D86-D76D076E566F} {24360989-A956-45E9-BF07-7FD9E7553C7D} = {9347BD22-1E8A-4A79-9D86-D76D076E566F} {9063F17B-5636-4AD5-999B-C894517DB5FD} = {329736D2-BF92-4D06-A7BF-19F4B6B64EDD} + {44DBA505-FB28-428F-AD4F-3BBF3BD12DB3} = {329736D2-BF92-4D06-A7BF-19F4B6B64EDD} EndGlobalSection GlobalSection(ExtensibilityGlobals) = postSolution SolutionGuid = {8B7C7E31-2E32-4E0D-9426-BC9AF22E9F4C} diff --git a/Directory.Packages.props b/Directory.Packages.props index 5f765d12d8..7d58a8fec9 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -13,6 +13,7 @@ + @@ -104,6 +105,7 @@ + @@ -116,6 +118,8 @@ + + all runtime; build; native; contentfiles; analyzers; buildtransitive diff --git a/src/Paramore.Brighter.MessagingGateway.Pulsar/Paramore.Brighter.MessagingGateway.Pulsar.csproj b/src/Paramore.Brighter.MessagingGateway.Pulsar/Paramore.Brighter.MessagingGateway.Pulsar.csproj new file mode 100644 index 0000000000..c19084ffbf --- /dev/null +++ b/src/Paramore.Brighter.MessagingGateway.Pulsar/Paramore.Brighter.MessagingGateway.Pulsar.csproj @@ -0,0 +1,19 @@ + + + + Brighter Messaging Gateway for Apache Pulsar + Your Name + $(BrighterTargetFrameworks) + latest + enable + + + + + + + + + + + \ No newline at end of file diff --git a/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarMessageConsumer.cs b/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarMessageConsumer.cs new file mode 100644 index 0000000000..22fd9e64f0 --- /dev/null +++ b/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarMessageConsumer.cs @@ -0,0 +1,171 @@ +using System; +using System.Buffers; +using System.Collections.Concurrent; +using System.Text; +using System.Threading; +using System.Threading.Tasks; +using DotPulsar; +using DotPulsar.Abstractions; +using DotPulsar.Extensions; +using Paramore.Brighter; + +namespace Paramore.Brighter.MessagingGateway.Pulsar +{ + /// + /// Brighter message consumer implementation using DotPulsar. + /// Responsible for receiving, acknowledging, and requeuing messages from Pulsar. + /// + public sealed class PulsarMessageConsumer : IAmAMessageConsumerAsync + { + private readonly PulsarMessagingGatewayConfiguration _config; + private readonly IPulsarClient _client; + private IConsumer? _consumer; + + // IMPORTANT: Map between Brighter's Message.Id -> Pulsar IMessage + // Needed to ACK or REQUEUE the exact message later. + private readonly ConcurrentDictionary> _inflight = new(); + + public PulsarMessageConsumer(PulsarMessagingGatewayConfiguration config) + : this(config, PulsarClient.Builder().ServiceUrl(new Uri(config.ServiceUrl)).Build()) + { } + + public PulsarMessageConsumer(PulsarMessagingGatewayConfiguration config, IPulsarClient client) + { + _config = config ?? throw new ArgumentNullException(nameof(config)); + _config.Validate(); + _client = client ?? throw new ArgumentNullException(nameof(client)); + } + + /// + /// Ensure that the Pulsar consumer is created only once. + /// + private void EnsureConsumer() + { + if (_consumer != null) + return; + + _consumer = _client + .NewConsumer(Schema.ByteArray) + .Topic(_config.Topic) + .SubscriptionName(_config.SubscriptionName) + .Create(); + } + + /// + /// Receive a single message from Pulsar and return it as a Brighter Message. + /// Note: Brighter's channel.Receive() often works in small batches; returning one message here is sufficient. + /// + public async Task ReceiveAsync(TimeSpan? timeOut = null, CancellationToken cancellationToken = default) + { + EnsureConsumer(); + + using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + cts.CancelAfter(timeOut ?? TimeSpan.FromMilliseconds(1000)); + + try + { + await foreach (var pulsarMsg in _consumer!.Messages(cts.Token)) + { + // Convert byte[] payload -> UTF8 string body + var payload = pulsarMsg.Data.ToArray(); + var bodyText = Encoding.UTF8.GetString(payload); + + // Create a Brighter message with a new unique Id + var brighterId = new Id(Guid.NewGuid().ToString()); + var brighterMsg = new Message( + new MessageHeader(brighterId, _config.Topic, MessageType.MT_EVENT), + new MessageBody(bodyText) + ); + + // Track message for later ACK/REQUEUE + _inflight[brighterId] = pulsarMsg; + + return new[] { brighterMsg }; + } + } + catch (OperationCanceledException) + { + // IMPORTANT: Timeout reached -> return empty array (no messages) + } + catch (Exception ex) + { + // TODO: Replace Console.Error with ILogger + Console.Error.WriteLine($"[Pulsar] Receive error: {ex}"); + } + + return Array.Empty(); + } + + /// + /// Acknowledge a previously received message. + /// + public async Task AcknowledgeAsync(Message message, CancellationToken cancellationToken = default) + { + if (message == null) + throw new ArgumentNullException(nameof(message)); + if (_consumer == null) + return; + + if (_inflight.TryRemove(message.Id, out var pulsarMsg)) + { + // DotPulsar.Extensions – ACK must be called with the exact IMessage received. + await _consumer.Acknowledge(pulsarMsg, cancellationToken); + } + } + + /// + /// Reject is not natively supported by Pulsar; we simulate by requesting re-delivery. + /// + public async Task RejectAsync(Message message, CancellationToken cancellationToken = default) + { + return await RequeueAsync(message, null, cancellationToken); + } + + /// + /// Pulsar does not support purging messages directly for a topic/subscription. + /// + public async Task PurgeAsync(CancellationToken cancellationToken = default) + { + await Task.Yield(); + throw new NotSupportedException("Apache Pulsar does not support purging messages directly."); + } + + /// + /// Request re-delivery for a specific unacknowledged message. + /// NOTE: Pulsar will redeliver; delay parameter is ignored (not supported natively). + /// + public async Task RequeueAsync(Message message, TimeSpan? delay = null, CancellationToken cancellationToken = default) + { + if (message == null) + throw new ArgumentNullException(nameof(message)); + if (_consumer == null) + return false; + + if (_inflight.TryRemove(message.Id, out var pulsarMsg)) + { + // IMPORTANT: Correct way to ask Pulsar to redeliver a specific message. + await _consumer.RedeliverUnacknowledgedMessages(new[] { pulsarMsg.MessageId }, cancellationToken); + return true; + } + + // If message is not in _inflight, we cannot requeue it specifically. + return false; + } + + /// + /// Dispose Pulsar consumer and client gracefully. + /// + public async ValueTask DisposeAsync() + { + try + { + if (_consumer is not null) + await _consumer.DisposeAsync(); + } + finally + { + await _client.DisposeAsync(); + } + } + } +} diff --git a/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarMessageProducer.cs b/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarMessageProducer.cs new file mode 100644 index 0000000000..a733b4342e --- /dev/null +++ b/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarMessageProducer.cs @@ -0,0 +1,88 @@ +using System; +using System.Diagnostics; +using System.Text; +using System.Threading; +using System.Threading.Tasks; +using DotPulsar; +using DotPulsar.Abstractions; +using DotPulsar.Extensions; +using Paramore.Brighter; + +namespace Paramore.Brighter.MessagingGateway.Pulsar +{ + /// + /// Simple Brighter message producer using DotPulsar. + /// Sends messages as byte[] encoded in UTF-8. + /// + public sealed class PulsarMessageProducer : IAmAMessageProducer, IAsyncDisposable + { + private readonly PulsarMessagingGatewayConfiguration _config; + private readonly IPulsarClient _client; + private IProducer? _producer; + + // IAmAMessageProducer contract + public Publication Publication { get; } + public Activity? Span { get; set; } + public IAmAMessageScheduler? Scheduler { get; set; } + + public PulsarMessageProducer(PulsarMessagingGatewayConfiguration config, Publication publication) + : this(config, publication, PulsarClient.Builder().ServiceUrl(new Uri(config.ServiceUrl)).Build()) + { } + + public PulsarMessageProducer(PulsarMessagingGatewayConfiguration config, Publication publication, IPulsarClient client) + { + _config = config ?? throw new ArgumentNullException(nameof(config)); + _config.Validate(); + Publication = publication ?? throw new ArgumentNullException(nameof(publication)); + _client = client ?? throw new ArgumentNullException(nameof(client)); + } + + /// + /// Ensure the Pulsar producer is created only once. + /// + private void EnsureProducer() + { + if (_producer != null) + return; + + _producer = _client + .NewProducer(Schema.ByteArray) + .Topic(_config.Topic) + .Create(); + } + + /// + /// Send a Brighter Message to Pulsar. + /// IMPORTANT: Brighter will wrap this through its own pipeline, + /// so this method focuses purely on encoding and publishing. + /// + public async Task SendAsync(Message message, CancellationToken cancellationToken = default) + { + if (message == null) + throw new ArgumentNullException(nameof(message)); + EnsureProducer(); + + // Convert message body to UTF-8 byte[] + var payload = Encoding.UTF8.GetBytes(message.Body.Value); + + // DotPulsar's Send sends a single message and waits for acknowledgment from Pulsar + await _producer!.Send(payload, cancellationToken); + } + + /// + /// Dispose Pulsar producer and client gracefully. + /// + public async ValueTask DisposeAsync() + { + try + { + if (_producer is not null) + await _producer.DisposeAsync(); + } + finally + { + await _client.DisposeAsync(); + } + } + } +} diff --git a/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarMessagingGatewayConfiguration.cs b/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarMessagingGatewayConfiguration.cs new file mode 100644 index 0000000000..2be6938e09 --- /dev/null +++ b/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarMessagingGatewayConfiguration.cs @@ -0,0 +1,21 @@ +using System; + +namespace Paramore.Brighter.MessagingGateway.Pulsar +{ + public class PulsarMessagingGatewayConfiguration + { + public string ServiceUrl { get; set; } = "pulsar://localhost:6650"; + public string Topic { get; set; } = ""; + public string SubscriptionName { get; set; } = ""; + + public void Validate() + { + if (string.IsNullOrWhiteSpace(ServiceUrl)) + throw new ArgumentException("ServiceUrl is required", nameof(ServiceUrl)); + if (string.IsNullOrWhiteSpace(Topic)) + throw new ArgumentException("Topic is required", nameof(Topic)); + if (string.IsNullOrWhiteSpace(SubscriptionName)) + throw new ArgumentException("SubscriptionName is required", nameof(SubscriptionName)); + } + } +} diff --git a/src/Paramore.Brighter.MessagingGateway.Pulsar/ServiceCollectionExtensions.cs b/src/Paramore.Brighter.MessagingGateway.Pulsar/ServiceCollectionExtensions.cs new file mode 100644 index 0000000000..7404af3970 --- /dev/null +++ b/src/Paramore.Brighter.MessagingGateway.Pulsar/ServiceCollectionExtensions.cs @@ -0,0 +1,37 @@ +using System; +using DotPulsar; +using DotPulsar.Abstractions; +using Microsoft.Extensions.DependencyInjection; + +namespace Paramore.Brighter.MessagingGateway.Pulsar +{ + public static class ServiceCollectionExtensions + { + public static IServiceCollection AddBrighterPulsar( + this IServiceCollection services, + PulsarMessagingGatewayConfiguration config, + Publication publication) + { + if (services == null) + throw new ArgumentNullException(nameof(services)); + if (config == null) + throw new ArgumentNullException(nameof(config)); + if (publication == null) + throw new ArgumentNullException(nameof(publication)); + + config.Validate(); + + services.AddSingleton(config); + services.AddSingleton(_ => + PulsarClient.Builder().ServiceUrl(new Uri(config.ServiceUrl)).Build()); + + services.AddSingleton(sp => + new PulsarMessageProducer(config, publication, sp.GetRequiredService())); + + services.AddSingleton(sp => + new PulsarMessageConsumer(config, sp.GetRequiredService())); + + return services; + } + } +} diff --git a/src/Paramore.Brighter.ServiceActivator.Control.Api/Properties/launchSettings.json b/src/Paramore.Brighter.ServiceActivator.Control.Api/Properties/launchSettings.json new file mode 100644 index 0000000000..9079a8535b --- /dev/null +++ b/src/Paramore.Brighter.ServiceActivator.Control.Api/Properties/launchSettings.json @@ -0,0 +1,12 @@ +{ + "profiles": { + "Paramore.Brighter.ServiceActivator.Control.Api": { + "commandName": "Project", + "launchBrowser": true, + "environmentVariables": { + "ASPNETCORE_ENVIRONMENT": "Development" + }, + "applicationUrl": "https://localhost:57434;http://localhost:57435" + } + } +} \ No newline at end of file