diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 290cd014ac..346018c17d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -546,6 +546,35 @@ jobs: run: dotnet restore - name: Azure Tests run: dotnet test ./tests/Paramore.Brighter.AzureServiceBus.Tests/Paramore.Brighter.AzureServiceBus.Tests.csproj --filter "Fragile!=CI" --configuration Release --logger "console;verbosity=normal" --logger GitHubActions --blame -v n + + pulsar-ci: + runs-on: ubuntu-latest + timeout-minutes: 5 + needs: [build] + steps: + - uses: actions/checkout@v5 + with: + fetch-depth: 0 # Required to fetch the Git tags + filter: tree:0 + show-progress: false + - name: Setup dotnet + uses: actions/setup-dotnet@v5 + with: + dotnet-version: | + 8.0.x + 9.0.x + - uses: actions/cache@v4 + with: + path: ~/.nuget/packages + key: Linux-nuget-${{ hashFiles('**/Directory.Packages.props') }} + restore-keys: | + Linux-nuget + + - name: Set up Apache Pulsar + uses: reugn/github-action-pulsar@v1 + + - name: Pulsar Tests + run: dotnet test ./tests/Paramore.Brighter.Pulsar.Tests/Paramore.Brighter.Pulsar.Tests.csproj --filter "Fragile!=CI" --configuration Release --logger "console;verbosity=normal" --logger GitHubActions --blame -v n mongodb-ci: runs-on: ubuntu-latest diff --git a/Brighter.sln b/Brighter.sln index 21597c1769..8a70aab87c 100644 --- a/Brighter.sln +++ b/Brighter.sln @@ -442,6 +442,10 @@ Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TaskReceiverConsole", "samp EndProject Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "TaskStatusSender", "samples\TaskQueue\KafkaDynamicEventStream\TaskStatusSender\TaskStatusSender.csproj", "{24360989-A956-45E9-BF07-7FD9E7553C7D}" EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paramore.Brighter.MessagingGateway.Pulsar", "src\Paramore.Brighter.MessagingGateway.Pulsar\Paramore.Brighter.MessagingGateway.Pulsar.csproj", "{BDC154A4-5978-4D39-BEEC-4E2F41DF334A}" +EndProject +Project("{FAE04EC0-301F-11D3-BF4B-00C04F79EFBC}") = "Paramore.Brighter.Pulsar.Tests", "tests\Paramore.Brighter.Pulsar.Tests\Paramore.Brighter.Pulsar.Tests.csproj", "{94C2C616-E5B1-439A-AD78-1146180AEC84}" +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.Transformers.Gcp", "src\Paramore.Brighter.Transformers.Gcp\Paramore.Brighter.Transformers.Gcp.csproj", "{231C1680-1590-458A-B9B2-38A3EDEE5E24}" @@ -2732,6 +2736,30 @@ Global {261E1392-7713-525F-2859-7B40CA416A50}.Release|Mixed Platforms.Build.0 = Release|Any CPU {261E1392-7713-525F-2859-7B40CA416A50}.Release|x86.ActiveCfg = Release|Any CPU {261E1392-7713-525F-2859-7B40CA416A50}.Release|x86.Build.0 = Release|Any CPU + {BDC154A4-5978-4D39-BEEC-4E2F41DF334A}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {BDC154A4-5978-4D39-BEEC-4E2F41DF334A}.Debug|Any CPU.Build.0 = Debug|Any CPU + {BDC154A4-5978-4D39-BEEC-4E2F41DF334A}.Debug|Mixed Platforms.ActiveCfg = Debug|Any CPU + {BDC154A4-5978-4D39-BEEC-4E2F41DF334A}.Debug|Mixed Platforms.Build.0 = Debug|Any CPU + {BDC154A4-5978-4D39-BEEC-4E2F41DF334A}.Debug|x86.ActiveCfg = Debug|Any CPU + {BDC154A4-5978-4D39-BEEC-4E2F41DF334A}.Debug|x86.Build.0 = Debug|Any CPU + {BDC154A4-5978-4D39-BEEC-4E2F41DF334A}.Release|Any CPU.ActiveCfg = Release|Any CPU + {BDC154A4-5978-4D39-BEEC-4E2F41DF334A}.Release|Any CPU.Build.0 = Release|Any CPU + {BDC154A4-5978-4D39-BEEC-4E2F41DF334A}.Release|Mixed Platforms.ActiveCfg = Release|Any CPU + {BDC154A4-5978-4D39-BEEC-4E2F41DF334A}.Release|Mixed Platforms.Build.0 = Release|Any CPU + {BDC154A4-5978-4D39-BEEC-4E2F41DF334A}.Release|x86.ActiveCfg = Release|Any CPU + {BDC154A4-5978-4D39-BEEC-4E2F41DF334A}.Release|x86.Build.0 = Release|Any CPU + {94C2C616-E5B1-439A-AD78-1146180AEC84}.Debug|Any CPU.ActiveCfg = Debug|Any CPU + {94C2C616-E5B1-439A-AD78-1146180AEC84}.Debug|Any CPU.Build.0 = Debug|Any CPU + {94C2C616-E5B1-439A-AD78-1146180AEC84}.Debug|Mixed Platforms.ActiveCfg = Debug|Any CPU + {94C2C616-E5B1-439A-AD78-1146180AEC84}.Debug|Mixed Platforms.Build.0 = Debug|Any CPU + {94C2C616-E5B1-439A-AD78-1146180AEC84}.Debug|x86.ActiveCfg = Debug|Any CPU + {94C2C616-E5B1-439A-AD78-1146180AEC84}.Debug|x86.Build.0 = Debug|Any CPU + {94C2C616-E5B1-439A-AD78-1146180AEC84}.Release|Any CPU.ActiveCfg = Release|Any CPU + {94C2C616-E5B1-439A-AD78-1146180AEC84}.Release|Any CPU.Build.0 = Release|Any CPU + {94C2C616-E5B1-439A-AD78-1146180AEC84}.Release|Mixed Platforms.ActiveCfg = Release|Any CPU + {94C2C616-E5B1-439A-AD78-1146180AEC84}.Release|Mixed Platforms.Build.0 = Release|Any CPU + {94C2C616-E5B1-439A-AD78-1146180AEC84}.Release|x86.ActiveCfg = Release|Any CPU + {94C2C616-E5B1-439A-AD78-1146180AEC84}.Release|x86.Build.0 = Release|Any CPU EndGlobalSection GlobalSection(SolutionProperties) = preSolution HideSolutionNode = FALSE @@ -2871,6 +2899,7 @@ Global {79CA356E-B08C-4D88-88C9-653EC8D8BF4D} = {9347BD22-1E8A-4A79-9D86-D76D076E566F} {43F23C67-6C9D-44BB-B3A8-9313E570879D} = {9347BD22-1E8A-4A79-9D86-D76D076E566F} {24360989-A956-45E9-BF07-7FD9E7553C7D} = {9347BD22-1E8A-4A79-9D86-D76D076E566F} + {94C2C616-E5B1-439A-AD78-1146180AEC84} = {329736D2-BF92-4D06-A7BF-19F4B6B64EDD} {9063F17B-5636-4AD5-999B-C894517DB5FD} = {329736D2-BF92-4D06-A7BF-19F4B6B64EDD} {39B7CFF4-4CA9-4B1F-B9C4-EED3A657D00D} = {329736D2-BF92-4D06-A7BF-19F4B6B64EDD} {261E1392-7713-525F-2859-7B40CA416A50} = {329736D2-BF92-4D06-A7BF-19F4B6B64EDD} diff --git a/Directory.Packages.props b/Directory.Packages.props index 9236e5f601..a77fe6c3e4 100644 --- a/Directory.Packages.props +++ b/Directory.Packages.props @@ -33,6 +33,7 @@ + diff --git a/docker-compose-pulsar.yaml b/docker-compose-pulsar.yaml new file mode 100644 index 0000000000..8cc1aeeec4 --- /dev/null +++ b/docker-compose-pulsar.yaml @@ -0,0 +1,87 @@ +services: + # Start zookeeper + zookeeper: + image: apachepulsar/pulsar:latest + container_name: zookeeper + restart: on-failure + environment: + - metadataStoreUrl=zk:zookeeper:2181 + - PULSAR_MEM=-Xms256m -Xmx256m -XX:MaxDirectMemorySize=256m + command: + - bash + - -c + - | + bin/apply-config-from-env.py conf/zookeeper.conf && \ + bin/generate-zookeeper-config.sh conf/zookeeper.conf && \ + exec bin/pulsar zookeeper + healthcheck: + test: ["CMD", "bin/pulsar-zookeeper-ruok.sh"] + interval: 10s + timeout: 5s + retries: 30 + + # Init cluster metadata + pulsar-init: + container_name: pulsar-init + hostname: pulsar-init + image: apachepulsar/pulsar:latest + command: + - bash + - -c + - | + bin/pulsar initialize-cluster-metadata \ + --cluster cluster-a \ + --zookeeper zookeeper:2181 \ + --configuration-store zookeeper:2181 \ + --web-service-url http://broker:8080 \ + --broker-service-url pulsar://broker:6650 + depends_on: + zookeeper: + condition: service_healthy + + # Start bookie + bookie: + image: apachepulsar/pulsar:latest + container_name: bookie + restart: on-failure + environment: + - clusterName=cluster-a + - zkServers=zookeeper:2181 + - metadataServiceUri=metadata-store:zk:zookeeper:2181 + # otherwise every time we run docker compose uo or down we fail to start due to Cookie + # See: https://github.com/apache/bookkeeper/blob/405e72acf42bb1104296447ea8840d805094c787/bookkeeper-server/src/main/java/org/apache/bookkeeper/bookie/Cookie.java#L57-68 + - advertisedAddress=bookie + - BOOKIE_MEM=-Xms512m -Xmx512m -XX:MaxDirectMemorySize=256m + depends_on: + zookeeper: + condition: service_healthy + pulsar-init: + condition: service_completed_successfully + # Map the local directory to the container to avoid bookie startup failure due to insufficient container disks. + command: bash -c "bin/apply-config-from-env.py conf/bookkeeper.conf && exec bin/pulsar bookie" + + # Start broker + broker: + image: apachepulsar/pulsar:latest + container_name: broker + hostname: broker + restart: on-failure + environment: + - metadataStoreUrl=zk:zookeeper:2181 + - zookeeperServers=zookeeper:2181 + - clusterName=cluster-a + - managedLedgerDefaultEnsembleSize=1 + - managedLedgerDefaultWriteQuorum=1 + - managedLedgerDefaultAckQuorum=1 + - advertisedAddress=broker + - advertisedListeners=external:pulsar://127.0.0.1:6650 + - PULSAR_MEM=-Xms512m -Xmx512m -XX:MaxDirectMemorySize=256m + depends_on: + zookeeper: + condition: service_healthy + bookie: + condition: service_started + ports: + - "6650:6650" + - "8080:8080" + command: bash -c "bin/apply-config-from-env.py conf/broker.conf && exec bin/pulsar broker" \ No newline at end of file diff --git a/src/Paramore.Brighter.MessagingGateway.Pulsar/HeaderNames.cs b/src/Paramore.Brighter.MessagingGateway.Pulsar/HeaderNames.cs new file mode 100644 index 0000000000..ff194b9330 --- /dev/null +++ b/src/Paramore.Brighter.MessagingGateway.Pulsar/HeaderNames.cs @@ -0,0 +1,63 @@ +namespace Paramore.Brighter.MessagingGateway.Pulsar; + +/// +/// Contains constant definitions for header names used in messaging systems, +/// particularly for Apache Pulsar integration with Brighter. +/// Includes standard headers, CloudEvents (CE) headers, and custom Brighter-Pulsar headers. +/// +public static class HeaderNames +{ + /// Content type of the message payload (e.g., application/json) + public const string ContentType = "ContentType"; + + /// Correlation ID for tracing related messages + public const string CorrelationId = "CorrelationId"; + + /// CloudEvents-formatted unique message identifier + public const string MessageId = "CE-EventId"; + + /// Number of times a message has been processed/requeued + public const string HandledCount = "HandledCount"; + + /// Brighter message type classification (e.g., MT_COMMAND, MT_EVENT) + public const string MessageType = "MessageType"; + + /// Reply destination for request-reply patterns + public const string ReplyTo = "ReplyTo"; + + /// CloudEvents specification version (e.g., "1.0") + public const string SpecVersion = "CE-SpecVersion"; + + /// CloudEvents event type descriptor + public const string Type = "CE-EventType"; + + /// Timestamp of event occurrence in RFC3339 format + public const string Time = "CE-EventTime"; + + /// CloudEvents subject describing event content + public const string Subject = "CE-Subject"; + + /// CloudEvents schema URL for payload validation + public const string DataSchema = "CE-DataSchema"; + + /// CloudEvents source URI identifying event origin + public const string Source = "CE-Source"; + + /// W3C Trace Context traceparent value + public const string TraceParent = "CE-X-TraceParent"; + + /// W3C Trace Context tracestate value + public const string TraceState = "CE-X-TraceState"; + + /// OpenTelemetry baggage items (key-value pairs) + public const string Baggage = "CE-X-Baggage"; + + /// Original message topic/routing key + public const string Topic = "Topic"; + + /// Pulsar schema version identifier + public const string SchemaVersion = "Brighter-Pulsar-SchemaVersion"; + + /// Pulsar message sequence identifier + public const string SequenceId = "Brighter-Pulsar-SequenceId"; +} 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..f4a7cdd4ac --- /dev/null +++ b/src/Paramore.Brighter.MessagingGateway.Pulsar/Paramore.Brighter.MessagingGateway.Pulsar.csproj @@ -0,0 +1,23 @@ + + + + $(BrighterTargetFrameworks) + Provides an implementation of the messaging gateway for decoupled invocation in the Paramore.Brighter pipeline, using Apache Pulsar + Apache Pulsar;Pub/Sub;Command;Event;Service Activator;Decoupled;Invocation;Messaging;Remote;Command Dispatcher;Command Processor;Request;Service;Task Queue;Work Queue;Retry;Circuit Breaker;Availability + Rafael Andrade + enable + false + + + + + + + + + + + + + + diff --git a/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarBackgroundMessageConsumer.cs b/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarBackgroundMessageConsumer.cs new file mode 100644 index 0000000000..c79a265378 --- /dev/null +++ b/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarBackgroundMessageConsumer.cs @@ -0,0 +1,119 @@ +using System.Buffers; +using System.Threading; +using System.Threading.Channels; +using System.Threading.Tasks; +using DotPulsar.Abstractions; + +namespace Paramore.Brighter.MessagingGateway.Pulsar; + +/// +/// Background message consumer for Apache Pulsar that buffers messages in a bounded channel. +/// +/// +/// This class manages a background message consumption loop that: +/// +/// Receives messages from Pulsar using an +/// Writes messages to a bounded channel for consumption by other components +/// Implements reference counting for safe start/stop operations +/// +/// +/// The consumer uses a fire-and-forget pattern where the consumption loop runs independently once started. +/// +/// Maximum number of messages to buffer in the channel +/// Pulsar message consumer implementation +public sealed class PulsarBackgroundMessageConsumer(int maxLenght, IConsumer> consumer) +{ + private int _total; + private CancellationTokenSource? _cancellationTokenSource; + private readonly Channel>> _channel = System.Threading.Channels.Channel.CreateBounded>>(new BoundedChannelOptions(maxLenght) + { + SingleReader = false, SingleWriter = true + }); + + /// + /// Provides read access to the message channel + /// + public ChannelReader>> Reader => _channel.Reader; + + /// + /// Gets the underlying Pulsar consumer instance + /// + public IConsumer> Consumer => consumer; + + /// + /// Starts the background message consumption loop + /// + /// + /// Implements reference counting: + /// + /// First call starts the background loop + /// Subsequent calls increment the reference count but don't start additional loops + /// + /// + public void Start() + { + var total = Interlocked.Increment(ref _total); + if (total == 1) + { + _cancellationTokenSource = new CancellationTokenSource(); + _ = ExecuteAsync(_cancellationTokenSource.Token); + } + } + + /// + /// Background message consumption loop + /// + /// Cancellation token to stop the loop + /// + /// Continuously performs: + /// + /// Receive message from Pulsar + /// Write message to output channel + /// Wait for channel write availability + /// + /// + /// Errors during message reception are silently ignored to maintain loop continuity. + /// + private async Task ExecuteAsync(CancellationToken cancellationToken) + { + while (!cancellationToken.IsCancellationRequested) + { + try + { + var message = await consumer.Receive(cancellationToken); + if (message is null) + { + continue; + } + + await _channel.Writer.WriteAsync(message, cancellationToken); + await _channel.Writer.WaitToWriteAsync(cancellationToken); + } + catch + { + // Ignoring any errors + } + } + } + + /// + /// Stops the background message consumption loop + /// + /// + /// Implements reference counting: + /// + /// Decrements the reference count + /// Stops the loop when reference count reaches zero + /// + /// + /// Safe to call multiple times - only the last call that brings the count to zero will stop the loop. + /// + public void Stop() + { + var total = Interlocked.Decrement(ref _total); + if (total == 0 && _cancellationTokenSource != null) + { + _cancellationTokenSource.Cancel(); + } + } +} diff --git a/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarChannelFactory.cs b/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarChannelFactory.cs new file mode 100644 index 0000000000..7d298f0e24 --- /dev/null +++ b/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarChannelFactory.cs @@ -0,0 +1,69 @@ +using System.Threading; +using System.Threading.Tasks; + +namespace Paramore.Brighter.MessagingGateway.Pulsar; + +/// +/// Factory for creating Pulsar message channels (synchronous and asynchronous) using a Pulsar consumer factory. +/// +/// +/// This factory creates channels that integrate with Apache Pulsar messaging systems. +/// Channels are configured using subscription details including channel names, routing keys, and buffer sizes. +/// +/// The factory responsible for creating Pulsar message consumers +public class PulsarChannelFactory(PulsarMessageConsumerFactory factory) : IAmAChannelFactory +{ + /// + /// Creates a synchronous Pulsar message channel + /// + /// Subscription configuration containing channel parameters + /// Synchronous channel instance bound to Pulsar + /// + /// Uses the following subscription properties: + /// + /// ChannelName: Logical name for the channel + /// RoutingKey: Pulsar topic routing information + /// BufferSize: Internal message buffer capacity + /// + /// + public IAmAChannelSync CreateSyncChannel(Subscription subscription) + { + return new Channel( + subscription.ChannelName, + subscription.RoutingKey, + factory.Create(subscription), + subscription.BufferSize); + } + + /// + /// Creates an asynchronous Pulsar message channel + /// + /// Subscription configuration containing channel parameters + /// Asynchronous channel instance bound to Pulsar + /// + /// Uses the same subscription configuration parameters as . + /// + public IAmAChannelAsync CreateAsyncChannel(Subscription subscription) + { + return new ChannelAsync( + subscription.ChannelName, + subscription.RoutingKey, + factory.CreateAsync(subscription), + subscription.BufferSize); + } + + /// + /// Creates an asynchronous Pulsar channel wrapped in a completed Task + /// + /// Subscription configuration containing channel parameters + /// Cancellation token (not used in this implementation) + /// Completed task containing an asynchronous channel instance + /// + /// This implementation is synchronous and does not perform asynchronous operations. + /// The cancellation token parameter is provided for interface compliance but is not utilized. + /// + public Task CreateAsyncChannelAsync(Subscription subscription, CancellationToken ct = default) + { + return Task.FromResult(CreateAsyncChannel(subscription)); + } +} diff --git a/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarMessageConsumer.cs b/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarMessageConsumer.cs new file mode 100644 index 0000000000..1cc351ba53 --- /dev/null +++ b/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarMessageConsumer.cs @@ -0,0 +1,479 @@ +using System; +using System.Buffers; +using System.Collections.Generic; +using System.Net.Mime; +using System.Threading; +using System.Threading.Tasks; +using DotPulsar; +using Microsoft.Extensions.Logging; +using Paramore.Brighter.Logging; +using Paramore.Brighter.Observability; +using Paramore.Brighter.Tasks; + +namespace Paramore.Brighter.MessagingGateway.Pulsar; + +/// +/// A message consumer implementation for Apache Pulsar that supports both synchronous and asynchronous operations. +/// +/// +/// This class handles message consumption from Pulsar topics including: +/// - Message acknowledgment +/// - Message rejection +/// - Queue purging +/// - Message receiving +/// - Message requeuing +/// +/// It wraps a background Pulsar consumer and provides adapters to convert Pulsar-specific message formats +/// to Brighter's internal message format. +/// +/// The background consumer responsible for low-level Pulsar interactions +public partial class PulsarMessageConsumer(PulsarBackgroundMessageConsumer backgroundPulsarConsumer) : IAmAMessageConsumerAsync, IAmAMessageConsumerSync +{ + private static readonly ILogger s_logger = ApplicationLogging.CreateLogger(); + + /// + /// Acknowledge a message so it won't be redelivered + /// + /// The message to acknowledge + /// + /// Uses the "ReceiptHandle" stored in the message's header bag to identify + /// the Pulsar MessageId for acknowledgment + /// + public void Acknowledge(Message message) + => BrighterAsyncContext.Run(async () => await AcknowledgeAsync(message)); + + /// + /// Asynchronously acknowledge a message so it won't be redelivered + /// + /// The message to acknowledge + /// Optional cancellation token + /// Throws if acknowledgment fails + public async Task AcknowledgeAsync(Message message, CancellationToken cancellationToken = default) + { + if (!message.Header.Bag.TryGetValue("ReceiptHandle", out var receiptHandle)) + { + return; + } + + if (receiptHandle is not MessageId messageId) + { + return; + } + + try + { + await backgroundPulsarConsumer.Consumer.Acknowledge(messageId, cancellationToken); + Log.AcknowledgedMessage(s_logger, message.Id, messageId.ToString()); + } + catch (Exception ex) + { + Log.ErrorAcknowledgingMessage(s_logger, ex, message.Id, messageId.ToString()); + throw; + } + } + + /// + /// Reject a message so it can be redelivered + /// + /// The message to reject + /// True if message was rejected successfully + public bool Reject(Message message) + => BrighterAsyncContext.Run(async () => await RejectAsync(message)); + + /// + /// Asynchronously reject a message so it can be redelivered + /// + /// The message to reject + /// Optional cancellation token + /// True if message was rejected successfully + /// + /// In Pulsar, rejection is implemented as an acknowledgment since Pulsar + /// doesn't have a native reject mechanism. This might cause the message to be + /// permanently removed rather than redelivered depending on configuration. + /// + public async Task RejectAsync(Message message, CancellationToken cancellationToken = default) + { + if (!message.Header.Bag.TryGetValue("ReceiptHandle", out var receiptHandle)) + { + return false; + } + + if (receiptHandle is not MessageId messageId) + { + return false; + } + + try + { + Log.RejectingMessage(s_logger, message.Id, messageId.ToString()); + await backgroundPulsarConsumer.Consumer.Acknowledge(messageId, cancellationToken); + return true; + } + catch (Exception exception) + { + Log.ErrorRejectingMessage(s_logger, exception, message.Id, messageId.ToString()); + throw; + } + } + + /// + /// Purge all messages in the queue + /// + /// + /// Implemented by seeking to the latest message position, effectively + /// skipping all pending messages + /// + public void Purge() => BrighterAsyncContext.Run(async () => await PurgeAsync()); + + /// + /// Asynchronously purge all messages in the queue + /// + /// Optional cancellation token + public async Task PurgeAsync(CancellationToken cancellationToken = default) + { + try + { + Log.PurgingQueue(s_logger, backgroundPulsarConsumer.Consumer.SubscriptionName); + await backgroundPulsarConsumer.Consumer.Seek(MessageId.Latest, cancellationToken); + Log.PurgedQueue(s_logger, backgroundPulsarConsumer.Consumer.SubscriptionName); + } + catch (Exception exception) + { + Log.ErrorPurgingQueue(s_logger, exception, backgroundPulsarConsumer.Consumer.SubscriptionName); + throw; + } + } + + /// + /// Receive messages from the queue (synchronous wrapper) + /// + /// Maximum time to wait for messages + /// Array of received messages + public Message[] Receive(TimeSpan? timeOut = null) + => BrighterAsyncContext.Run(async () => await ReceiveAsync(timeOut)); + + /// + /// + /// Asynchronously receive messages from Pulsar + /// + /// Optional timeout for receiving messages + /// Optional cancellation token + /// Array of received messages + /// + /// Converts Pulsar messages to Brighter's internal message format: + /// 1. Extracts Pulsar properties into message header bag + /// 2. Maps Pulsar metadata to Brighter header fields + /// 3. Handles timeouts by returning empty message array + /// + /// Important mappings: + /// - Pulsar MessageId → "ReceiptHandle" in header bag + /// - Pulsar Properties → Custom header attributes + /// - EventTime → Timestamp header + /// - RedeliveryCount → HandledCount header + /// + public async Task ReceiveAsync(TimeSpan? timeOut = null, CancellationToken cancellationToken = default) + { + using var cts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken); + if (timeOut.HasValue && timeOut.Value != TimeSpan.Zero) + { + cts.CancelAfter(timeOut.Value); + } + + try + { + var pulsarMessage = await backgroundPulsarConsumer.Reader.ReadAsync(cts.Token); + var bag = new Dictionary(); + + foreach (var pair in pulsarMessage.Properties) + { + bag[pair.Key] = pair.Value; + } + + bag["ReceiptHandle"] = pulsarMessage.MessageId; + bag[HeaderNames.SequenceId] = pulsarMessage.SequenceId; + if (pulsarMessage.SchemaVersion != null) + { + bag[HeaderNames.SchemaVersion] = pulsarMessage.SchemaVersion; + } + + var header = new MessageHeader( + messageId: GetMessageId(pulsarMessage.Properties), + topic: GetTopic(pulsarMessage.Properties, backgroundPulsarConsumer.Consumer.Topic), + messageType: GetMessageType(pulsarMessage.Properties), + contentType: GetContentType(pulsarMessage.Properties), + partitionKey: GetPartitionKey(pulsarMessage.Key), + timeStamp: pulsarMessage.EventTimeAsDateTimeOffset, + correlationId: GetCorrelationId(pulsarMessage.Properties), + handledCount: (int)pulsarMessage.RedeliveryCount, + type: GetCloudEventType(pulsarMessage.Properties), + source: GetSource(pulsarMessage.Properties), + replyTo: GetReplyTo(pulsarMessage.Properties), + subject: GetSubject(pulsarMessage.Properties), + dataSchema: GetDataSchema(pulsarMessage.Properties), + baggage: GetBaggage(pulsarMessage.Properties), + traceParent: GetTraceParent(pulsarMessage.Properties), + traceState: GetTraceState(pulsarMessage.Properties)) + { + SpecVersion = GetSpecVersion(pulsarMessage.Properties), + Bag = bag + }; + + return [new Message(header, new MessageBody(pulsarMessage.Data.ToArray()))]; + + } + catch (OperationCanceledException) + { + return [new Message()]; + } + + static Id GetMessageId(IReadOnlyDictionary properties) + { + if (!properties.TryGetValue(HeaderNames.MessageId, out var id) || string.IsNullOrEmpty(id)) + { + return Id.Random(); + } + + return Id.Create(id); + } + + static RoutingKey GetTopic(IReadOnlyDictionary properties, string defaultTopic) + { + if (!properties.TryGetValue(HeaderNames.Topic, out var topic) || string.IsNullOrEmpty(topic)) + { + return new RoutingKey(defaultTopic); + } + + return new RoutingKey(topic); + } + + static MessageType GetMessageType(IReadOnlyDictionary properties) + { + if (!properties.TryGetValue(HeaderNames.MessageType, out var messageType) || string.IsNullOrEmpty(messageType)) + { + return MessageType.MT_EVENT; + } + +#if NETSTANDARD + return (MessageType)Enum.Parse(typeof(MessageType), messageType); +#else + return Enum.Parse(messageType); +#endif + } + + static ContentType GetContentType(IReadOnlyDictionary properties) + { + if (!properties.TryGetValue(HeaderNames.ContentType, out var contentType) || string.IsNullOrEmpty(contentType)) + { + return new ContentType(MediaTypeNames.Text.Plain); + } + + return new ContentType(contentType); + } + + static PartitionKey? GetPartitionKey(string? partitionKey) + => string.IsNullOrEmpty(partitionKey) ? null : new PartitionKey(partitionKey!); + + static Id GetCorrelationId(IReadOnlyDictionary properties) + { + if (!properties.TryGetValue(HeaderNames.CorrelationId, out var id) || string.IsNullOrEmpty(id)) + { + return Id.Empty; + } + + return Id.Create(id); + } + + static string GetSpecVersion(IReadOnlyDictionary properties) + { + if (!properties.TryGetValue(HeaderNames.SpecVersion, out var specVersion) + || string.IsNullOrEmpty(specVersion)) + { + return MessageHeader.DefaultSpecVersion; + } + + return specVersion; + } + + static CloudEventsType GetCloudEventType(IReadOnlyDictionary properties) + { + if (!properties.TryGetValue(HeaderNames.Type, out var type) + || string.IsNullOrEmpty(type)) + { + return CloudEventsType.Empty; + } + + return new CloudEventsType(type); + } + + static Uri GetSource(IReadOnlyDictionary properties) + { + if (!properties.TryGetValue(HeaderNames.Source, out var source) || !Uri.TryCreate(source, UriKind.RelativeOrAbsolute, out var sourceUri)) + { + return new Uri(MessageHeader.DefaultSource); + } + + return sourceUri; + } + + static RoutingKey? GetReplyTo(IReadOnlyDictionary properties) + { + if (!properties.TryGetValue(HeaderNames.ReplyTo, out var replyTo) || string.IsNullOrEmpty(replyTo)) + { + return null; + } + + return new RoutingKey(replyTo); + } + + static string? GetSubject(IReadOnlyDictionary properties) + { + properties.TryGetValue(HeaderNames.Subject, out var subject); + return subject; + } + + static Uri? GetDataSchema(IReadOnlyDictionary properties) + { + if (!properties.TryGetValue(HeaderNames.DataSchema, out var source) || !Uri.TryCreate(source, UriKind.RelativeOrAbsolute, out var sourceUri)) + { + return null; + } + + return sourceUri; + } + + static Baggage GetBaggage(IReadOnlyDictionary properties) + { + if (!properties.TryGetValue(HeaderNames.Baggage, out var baggage) || string.IsNullOrEmpty(baggage)) + { + return new Baggage(); + } + + return Baggage.FromString(baggage); + } + + static TraceParent? GetTraceParent(IReadOnlyDictionary properties) + { + if (!properties.TryGetValue(HeaderNames.TraceParent, out var traceParent) || string.IsNullOrEmpty(traceParent)) + { + return null; + } + + return new TraceParent(traceParent); + } + + static TraceState? GetTraceState(IReadOnlyDictionary properties) + { + if (!properties.TryGetValue(HeaderNames.TraceState, out var traceState) || string.IsNullOrEmpty(traceState)) + { + return null; + } + + return new TraceState(traceState); + } + } + + /// + /// Requeue a message with optional delay + /// + /// The message to requeue + /// Optional delay before redelivery (not supported in Pulsar) + /// True if message was requeued successfully + /// + /// In Pulsar, this triggers redelivery of the message through + /// the redeliverUnacknowledgedMessages API + /// + public bool Requeue(Message message, TimeSpan? delay = null) + => BrighterAsyncContext.Run(async () => await RequeueAsync(message, delay)); + + /// + /// + /// Asynchronously requeue a message with optional delay + /// + /// The message to requeue + /// Optional delay before redelivery (ignored in Pulsar implementation) + /// Optional cancellation token + /// True if message was requeued successfully + public async Task RequeueAsync(Message message, TimeSpan? delay = null, + CancellationToken cancellationToken = default) + { + if (!message.Header.Bag.TryGetValue("ReceiptHandle", out var receiptHandle)) + { + return false; + } + + var messageId = receiptHandle as MessageId; + if (messageId! == null!) + { + return false; + } + + try + { + + Log.RejectingMessage(s_logger, message.Id, messageId.ToString()); + + await backgroundPulsarConsumer.Consumer.RedeliverUnacknowledgedMessages([messageId], cancellationToken); + + Log.RequeuedMessage(s_logger, message.Id); + return true; + } + catch (Exception ex) + { + Log.ErrorRejectingMessage(s_logger, ex, message.Id, messageId.ToString()); + throw; + } + } + + /// + /// Asynchronously release consumer resources + /// + public ValueTask DisposeAsync() + { + backgroundPulsarConsumer.Stop(); + return new ValueTask(); + } + + /// + /// Release consumer resources + /// + public void Dispose() + { + backgroundPulsarConsumer.Stop(); + } + + private static partial class Log + { + [LoggerMessage(LogLevel.Information, "PulsarMessageConsumer: Acknowledged message {MessageId} with receipt handle {ReceiptHandle} ")] + public static partial void AcknowledgedMessage(ILogger logger, string messageId, string receiptHandle); + + [LoggerMessage(LogLevel.Error, "PulsarMessageConsumer: Error during acknowledged message {Id} with receipt handle {ReceiptHandle}")] + public static partial void ErrorAcknowledgingMessage(ILogger logger, Exception exception, string id, string receiptHandle); + + + [LoggerMessage(LogLevel.Information, "PulsarMessageConsumer: Rejecting the message {Id} with receipt handle {ReceiptHandle}")] + public static partial void RejectingMessage(ILogger logger, string id, string? receiptHandle); + + [LoggerMessage(LogLevel.Error, "PulsarMessageConsumer: Error during rejecting the message {Id} with receipt handle {ReceiptHandle}")] + public static partial void ErrorRejectingMessage(ILogger logger, Exception exception, string id, string? receiptHandle); + + + [LoggerMessage(LogLevel.Information, "PulsarMessageConsumer: Purging the queue {ChannelName}")] + public static partial void PurgingQueue(ILogger logger, string channelName); + + [LoggerMessage(LogLevel.Information, "PulsarMessageConsumer: Purged the queue {ChannelName}")] + public static partial void PurgedQueue(ILogger logger, string channelName); + + [LoggerMessage(LogLevel.Error, "PulsarMessageConsumer: Error purging queue {ChannelName}")] + public static partial void ErrorPurgingQueue(ILogger logger, Exception exception, string channelName); + + + [LoggerMessage(LogLevel.Information, "PulsarMessageConsumer: re-queueing the message {Id}")] + public static partial void RequeueingMessage(ILogger logger, string id); + + [LoggerMessage(LogLevel.Information, "PulsarMessageConsumer: re-queued the message {Id}")] + public static partial void RequeuedMessage(ILogger logger, string id); + + [LoggerMessage(LogLevel.Error, "SqsMessageConsumer: Error during re-queueing the message {Id} with receipt handle {ReceiptHandle} on the queue {ChannelName}")] + public static partial void ErrorRequeueingMessage(ILogger logger, Exception exception, string id, string? receiptHandle, string channelName); + } +} diff --git a/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarMessageConsumerFactory.cs b/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarMessageConsumerFactory.cs new file mode 100644 index 0000000000..bd2779a98a --- /dev/null +++ b/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarMessageConsumerFactory.cs @@ -0,0 +1,62 @@ +using System.Collections.Concurrent; +using System.Collections.Generic; +using DotPulsar.Extensions; + +namespace Paramore.Brighter.MessagingGateway.Pulsar; + +/// +/// Factory for creating Pulsar message consumers (synchronous and asynchronous) for Brighter's message processing pipeline. +/// +/// The connection gateway to Apache Pulsar used for creating consumers. +public class PulsarMessageConsumerFactory(PulsarMessagingGatewayConnection connection) : IAmAMessageConsumerFactory +{ + private static readonly ConcurrentDictionary s_backgroundConsumers = new(); + + /// + /// Creates a consumer for the specified queue. + /// + /// The queue to connect to + /// IAmAMessageConsumerSync CreatePulsarConsumer(subscription); + + /// + /// Creates a consumer for the specified queue. + /// + /// The queue to connect to + /// IAmAMessageConsumerSync + public IAmAMessageConsumerAsync CreateAsync(Subscription subscription) + => CreatePulsarConsumer(subscription); + + private PulsarMessageConsumer CreatePulsarConsumer(Subscription subscription) + { + if (subscription is not PulsarSubscription pulsarSubscription) + { + throw new ConfigurationException("We expect PulsarSubscription or PulsarSubscription as a parameter"); + } + + var background = s_backgroundConsumers.GetOrAdd(pulsarSubscription, CreateConsumerBackground); + background.Start(); + return new PulsarMessageConsumer(background); + } + + private PulsarBackgroundMessageConsumer CreateConsumerBackground(PulsarSubscription pulsarSubscription) + { + var client = connection.Create(); + var builder = client.NewConsumer(pulsarSubscription.Schema) + .AllowOutOfOrderDeliver(pulsarSubscription.AllowOutOfOrderDeliver) + .InitialPosition(pulsarSubscription.InitialPosition) + .MessagePrefetchCount((uint)pulsarSubscription.BufferSize) + .PriorityLevel(pulsarSubscription.PriorityLevel) + .ReadCompacted(pulsarSubscription.ReadCompacted) + .SubscriptionName(pulsarSubscription.ChannelName.Value) + .SubscriptionType(pulsarSubscription.SubscriptionType) + .Topic(pulsarSubscription.RoutingKey); + + pulsarSubscription.Configuration?.Invoke(builder); + + var consumer = builder.Create(); + + return new PulsarBackgroundMessageConsumer(pulsarSubscription.NoOfPerformers, consumer); + } +} diff --git a/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarMessageProducer.cs b/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarMessageProducer.cs new file mode 100644 index 0000000000..7a7f631b62 --- /dev/null +++ b/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarMessageProducer.cs @@ -0,0 +1,153 @@ +using System; +using System.Buffers; +using System.Diagnostics; +using System.Threading; +using System.Threading.Tasks; +using DotPulsar; +using DotPulsar.Abstractions; +using Paramore.Brighter.Extensions; +using Paramore.Brighter.Observability; +using Paramore.Brighter.Tasks; + +namespace Paramore.Brighter.MessagingGateway.Pulsar; + +/// +/// Implements a message producer for Apache Pulsar within the Brighter message processing framework. +/// Handles message publication with optional delay and metadata management. +/// +/// The underlying Pulsar producer instance +/// Configuration for Pulsar message publication +/// Time provider for delayed message scheduling +/// Options controlling instrumentation behavior +/// +/// This producer bridges Brighter's message publication model with Pulsar's producer API: +/// - Converts Brighter to Pulsar's message format +/// - Supports delayed message delivery +/// - Manages cloud events metadata and standard headers +/// - Implements proper resource disposal patterns +/// +public class PulsarMessageProducer(IProducer> producer, + PulsarPublication publication, + TimeProvider time, + InstrumentationOptions instrumentation) : IAmAMessageProducerAsync, IAmAMessageProducerSync +{ + private const string Pulsar = "pulsar"; + + /// + public async ValueTask DisposeAsync() => await producer.DisposeAsync(); + + /// + /// The that this Producer is for. + /// + public Publication Publication => publication; + + /// + /// Allows us to set a to let a Producer participate in our telemetry + /// + public Activity? Span { get; set; } + + /// + /// The external message scheduler + /// + public IAmAMessageScheduler? Scheduler { get; set; } + + /// + /// Sends the specified message. + /// + /// The message. + public void Send(Message message) + => SendWithDelay(message, TimeSpan.Zero); + + /// + /// Sends the specified message. + /// + /// The message. + /// A cancellation token to end the operation + public Task SendAsync(Message message, CancellationToken cancellationToken = default) + => SendWithDelayAsync(message, TimeSpan.Zero, cancellationToken); + + /// + /// Send the specified message with specified delay + /// + /// The message. + /// Delay delivery of the message. 0 is no delay. Defaults to 0 + public void SendWithDelay(Message message, TimeSpan? delay) + => BrighterAsyncContext.Run(async() => await SendWithDelayAsync(message, delay)); + + + /// + /// Send the specified message with specified delay + /// + /// The message. + /// Delay to the delivery of the message. 0 is no delay. Defaults to 0 + /// A cancellation token to end the operation + public async Task SendWithDelayAsync(Message message, TimeSpan? delay, CancellationToken cancellationToken = default) + { + BrighterTracer.WriteProducerEvent(Span, Pulsar, message, instrumentation); + await producer.Send(CreateMessageMetadata(message, delay), + new ReadOnlySequence(message.Body.Bytes), + cancellationToken); + } + + private MessageMetadata CreateMessageMetadata(Message message, TimeSpan? delay) + { + var metadata = new MessageMetadata + { + Key = PartitionKey.IsNullOrEmpty(message.Header.PartitionKey) ? null : message.Header.PartitionKey.Value, + EventTimeAsDateTimeOffset = message.Header.TimeStamp, + SchemaVersion = publication.SchemaVersion, + SequenceId = publication.GenerateSequenceId(message), + }; + + if (delay.HasValue && delay.Value != TimeSpan.Zero) + { + metadata.DeliverAtTimeAsDateTimeOffset = time.GetUtcNow() + delay.Value; + } + + metadata[HeaderNames.ContentType] = message.Header.ContentType.ToString(); + metadata[HeaderNames.CorrelationId] = message.Header.CorrelationId; + metadata[HeaderNames.MessageType] = message.Header.MessageType.ToString(); + metadata[HeaderNames.MessageId] = message.Header.MessageId; + metadata[HeaderNames.SpecVersion] = message.Header.SpecVersion; + metadata[HeaderNames.Type] = message.Header.Type; + metadata[HeaderNames.Time] = message.Header.TimeStamp.ToRfc3339(); + metadata[HeaderNames.Topic] = message.Header.Topic; + metadata[HeaderNames.Source] = message.Header.Source.ToString(); + metadata[HeaderNames.Baggage] = message.Header.Baggage.ToString(); + + if (!RoutingKey.IsNullOrEmpty(message.Header.ReplyTo)) + { + metadata[HeaderNames.ReplyTo] = message.Header.ReplyTo; + } + + if (!string.IsNullOrEmpty(message.Header.Subject)) + { + metadata[HeaderNames.Subject] = message.Header.Subject; + } + + if (message.Header.DataSchema != null) + { + metadata[HeaderNames.DataSchema] = message.Header.DataSchema.ToString(); + } + + if (!TraceParent.IsNullOrEmpty(message.Header.TraceParent)) + { + metadata[HeaderNames.TraceParent] = message.Header.TraceParent; + } + + if (!TraceState.IsNullOrEmpty(message.Header.TraceState)) + { + metadata[HeaderNames.TraceState] = message.Header.TraceState; + } + + foreach (var pair in message.Header.Bag) + { + metadata[pair.Key] = pair.Value.ToString(); + } + + return metadata; + } + + /// + public void Dispose() => DisposeAsync().GetAwaiter().GetResult(); +} diff --git a/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarMessagingGatewayConnection.cs b/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarMessagingGatewayConnection.cs new file mode 100644 index 0000000000..469d5af878 --- /dev/null +++ b/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarMessagingGatewayConnection.cs @@ -0,0 +1,72 @@ +using System; +using DotPulsar; +using DotPulsar.Abstractions; +using Paramore.Brighter.Observability; + +namespace Paramore.Brighter.MessagingGateway.Pulsar; + +/// +/// Represents a connection gateway to Apache Pulsar, providing configuration and client creation capabilities. +/// Manages the lifecycle of the underlying Pulsar client using the Singleton pattern. +/// +/// +/// This class is thread-safe and intended to be shared across consumers/producers in an application. +/// The created Pulsar client is cached after initial creation. +/// +public class PulsarMessagingGatewayConnection +{ + /// + /// Gets or sets the optional producer name identifier (used for diagnostics and monitoring) + /// + public string? ProducerName { get; set; } + + /// + /// Gets or sets the Apache Pulsar service URL (e.g., "pulsar://localhost:6650") + /// + /// + /// Required if not provided through the Configuration callback + /// + public Uri? ServiceUrl { get; set; } + + /// + /// Gets or sets an optional configuration callback for advanced Pulsar client settings + /// + /// + /// Allows customization of the Pulsar client builder beyond basic properties. + /// Invoked during client creation before the client is built. + /// + public Action? Configuration { get; set; } + + /// + /// Gets or sets the instrumentation options controlling what metrics are collected + /// + /// + /// Default: InstrumentationOptions.All (collect all available metrics) + /// + public InstrumentationOptions Instrumentation { get; set; } = InstrumentationOptions.All; + + private IPulsarClient? _pulsarClient; + + /// + /// Creates a singleton instance of the Pulsar client using a thread-safe initialization pattern. + /// + /// The singleton instance + public IPulsarClient Create() + { + if (_pulsarClient != null) + { + return _pulsarClient; + } + + var builder = PulsarClient.Builder(); + + if (ServiceUrl != null) + { + builder.ServiceUrl(ServiceUrl); + } + + Configuration?.Invoke(builder); + + return _pulsarClient = builder.Build(); + } +} diff --git a/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarProducerFactory.cs b/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarProducerFactory.cs new file mode 100644 index 0000000000..0f6875cef9 --- /dev/null +++ b/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarProducerFactory.cs @@ -0,0 +1,60 @@ +using System.Collections.Generic; +using System.Threading.Tasks; +using DotPulsar.Extensions; + +namespace Paramore.Brighter.MessagingGateway.Pulsar; + +/// +/// Factory for creating Pulsar message producers based on publication configurations +/// +/// Shared Pulsar connection configuration +/// Collection of publication definitions +public class PulsarProducerFactory(PulsarMessagingGatewayConnection connection, IEnumerable publications) : IAmAMessageProducerFactory +{ + /// + /// Creates message producers. + /// + /// A dictionary of middleware clients by topic/routing key, for sending messages to the middleware + public Dictionary Create() + { + var client = connection.Create(); + var producers = new Dictionary(); + foreach (var publication in publications) + { + if (publication.Topic is null) + { + throw new ConfigurationException("Missing topic on Publication"); + } + + var builder = client.NewProducer(publication.Schema) + .CompressionType(publication.CompressionType) + .InitialSequenceId(publication.InitialSequenceId) + .ProducerAccessMode(publication.AccessMode) + .Topic(publication.Topic); + + var producerName = publication.Name ?? connection.ProducerName; + if (!string.IsNullOrWhiteSpace(producerName)) + { + builder = builder.ProducerName(producerName!); + } + + publication.Configure?.Invoke(builder); + + var producer = builder.Create(); + + producers[new ProducerKey(publication.Topic, publication.Type)] = new PulsarMessageProducer(producer, + publication, + publication.TimeProvider, + publication.Instrumentation ?? connection.Instrumentation); + } + + return producers; + } + + /// + /// Creates message producers. + /// + /// A dictionary of middleware clients by topic/routing key, for sending messages to the middleware + public Task> CreateAsync() + => Task.FromResult(Create()); +} diff --git a/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarPublication.cs b/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarPublication.cs new file mode 100644 index 0000000000..c9c493e4bb --- /dev/null +++ b/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarPublication.cs @@ -0,0 +1,110 @@ +using System; +using System.Buffers; +using DotPulsar; +using DotPulsar.Abstractions; +using Paramore.Brighter.Observability; + +namespace Paramore.Brighter.MessagingGateway.Pulsar; + +/// +/// Configuration for publishing messages to Apache Pulsar within the Brighter framework +/// +/// +/// Extends Brighter's base Publication with Pulsar-specific producer configuration options. +/// Used to define producer behavior, schema handling, and message formatting. +/// +public class PulsarPublication : Publication +{ + /// + /// Gets or sets the compression type for published messages + /// + /// + /// Default: CompressionType.None + /// + public CompressionType CompressionType { get; set; } = CompressionType.None; + + /// + /// Gets or sets the optional producer name (used for diagnostics) + /// + public string? Name { get; set; } + + /// + /// Gets or sets the schema version identifier + /// + public byte[]? SchemaVersion { get; set; } + + /// + /// Gets or sets the initial sequence ID for messages + /// + public ulong InitialSequenceId { get; set; } + + /// + /// Gets or sets the producer access mode + /// + /// + /// Default: ProducerAccessMode.Shared + /// + public ProducerAccessMode AccessMode { get; set; } = ProducerAccessMode.Shared; + + /// + /// Gets or sets the schema for message serialization + /// + /// + /// Default: ByteSequence schema (raw byte array) + /// + public ISchema> Schema { get; set; } = DotPulsar.Schema.ByteSequence; + + /// + /// Gets or sets the function to generate message sequence IDs + /// + /// + /// Default behavior returns 0 (no sequence tracking). + /// Override to implement custom sequence generation. + /// + public Func GenerateSequenceId { get; set; } = _ => 0; + + /// + /// Gets or sets the time provider for delayed message scheduling + /// + /// + /// Default: System time provider + /// + public TimeProvider TimeProvider { get; set; } = TimeProvider.System; + + /// + /// Gets or sets instrumentation options for monitoring + /// + public InstrumentationOptions? Instrumentation { get; set; } + + /// + /// Gets or sets an optional configuration callback for advanced producer setup + /// + /// + /// Allows direct access to the Pulsar producer builder for custom configuration + /// not exposed through standard properties. + /// + public Action>>? Configure { get; set; } +} + +/// +/// Typed publication configuration for specific message types +/// +/// The request type being published +/// +/// Specializes for specific message types. +/// Automatically sets the RequestType property to the specified generic type. +/// +public class PulsarPublication : PulsarPublication + where T : IRequest +{ + /// + /// Initializes a new instance of the typed publication + /// + /// + /// Sets the RequestType property to the generic type argument + /// + public PulsarPublication() + { + RequestType = typeof(T); + } +} diff --git a/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarSubscription.cs b/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarSubscription.cs new file mode 100644 index 0000000000..46304ef51c --- /dev/null +++ b/src/Paramore.Brighter.MessagingGateway.Pulsar/PulsarSubscription.cs @@ -0,0 +1,183 @@ +using System; +using System.Buffers; +using DotPulsar; +using DotPulsar.Abstractions; + +namespace Paramore.Brighter.MessagingGateway.Pulsar; + +/// +/// Configuration for consuming messages from Apache Pulsar within the Brighter framework +/// +/// +/// Extends Brighter's base Subscription with Pulsar-specific consumer configuration options. +/// Defines consumer behavior, message ordering, and subscription semantics. +/// +public class PulsarSubscription : Subscription +{ + /// + /// Initializes a new Pulsar subscription configuration + /// + /// Unique name for this subscription + /// Name of the channel associated with this subscription + /// Topic routing key + /// The of the data that this subscription handles. + /// The that determines how we map a message to a type. Defaults to returning the if null + /// Number of messages to prefetch + /// Number of concurrent consumers + /// Timeout for receive operations + /// Number of times to requeue failed messages (-1 = infinite) + /// Delay before requeuing failed messages + /// Limit for consecutive message failures before stopping + /// Type of message pump to use + /// Factory for creating channels + /// Behavior when channels are missing + /// Delay when no messages are available + /// Delay after channel failures + /// Schema for message deserialization + /// Initial position in the topic + /// Consumer priority level + /// Whether to read from compacted topics + /// Pulsar subscription type + /// Allow out-of-order message delivery + /// Custom consumer configuration callback + public PulsarSubscription(SubscriptionName subscriptionName, ChannelName channelName, RoutingKey routingKey, + Type? requestType = null, Func? getRequestType = null, int bufferSize = 1, int noOfPerformers = 1, + TimeSpan? timeOut = null, int requeueCount = -1, TimeSpan? requeueDelay = null, int unacceptableMessageLimit = 0, + MessagePumpType messagePumpType = MessagePumpType.Unknown, IAmAChannelFactory? channelFactory = null, + OnMissingChannel makeChannels = OnMissingChannel.Create, TimeSpan? emptyChannelDelay = null, + TimeSpan? channelFailureDelay = null, + ISchema>? schema = null, + SubscriptionInitialPosition initialPosition = SubscriptionInitialPosition.Earliest, + int priorityLevel = 0, + bool readCompacted = false, + SubscriptionType subscriptionType = SubscriptionType.Exclusive, + bool allowOutOfOrderDeliver = false, + Action>>? configuration = null) + : base(subscriptionName, channelName, routingKey, requestType, getRequestType, bufferSize, noOfPerformers, timeOut, requeueCount, + requeueDelay, unacceptableMessageLimit, messagePumpType, channelFactory, makeChannels, emptyChannelDelay, + channelFailureDelay) + { + Schema = schema ?? DotPulsar.Schema.ByteSequence; + InitialPosition = initialPosition; + PriorityLevel = priorityLevel; + ReadCompacted = readCompacted; + SubscriptionType = subscriptionType; + AllowOutOfOrderDeliver = allowOutOfOrderDeliver; + Configuration = configuration; + } + + /// + /// Gets the schema for message deserialization + /// + /// + /// Default: ByteSequence schema (raw byte array) + /// + public ISchema> Schema { get; } + + /// + /// Gets the initial position in the topic + /// + /// + /// Default: SubscriptionInitialPosition.Earliest + /// + public SubscriptionInitialPosition InitialPosition { get; } + + /// + /// Gets the consumer priority level + /// + /// + /// Default: 0 + /// + public int PriorityLevel { get; } + + /// + /// Gets whether to read from compacted topics + /// + /// + /// Default: false + /// + public bool ReadCompacted { get; } + + /// + /// Gets the Pulsar subscription type + /// + /// + /// Default: SubscriptionType.Exclusive + /// + public SubscriptionType SubscriptionType { get; } + + /// + /// Gets whether to allow out-of-order message delivery + /// + /// + /// Default: false + /// + public bool AllowOutOfOrderDeliver { get; } + + /// + /// Gets an optional custom configuration callback + /// + /// + /// Allows direct access to the Pulsar consumer builder for advanced configuration + /// not exposed through standard properties. + /// + public Action>>? Configuration { get; } +} + +/// +/// Typed subscription configuration for specific message types +/// +/// The request type being consumed +/// +/// Specializes for specific message types. +/// Automatically sets the DataType property to the specified generic type. +/// +public class PulsarSubscription : PulsarSubscription + where T : IRequest +{ + /// + /// Initializes a new typed Pulsar subscription + /// + /// Unique name for this subscription + /// Name of the channel associated with this subscription + /// Topic routing key + /// The that determines how we map a message to a type. Defaults to returning the if null + /// Number of messages to prefetch + /// Number of concurrent consumers + /// Timeout for receive operations + /// Number of times to requeue failed messages (-1 = infinite) + /// Delay before requeuing failed messages + /// Limit for consecutive message failures before stopping + /// Type of message pump to use + /// Factory for creating channels + /// Behavior when channels are missing + /// Delay when no messages are available + /// Delay after channel failures + /// Schema for message deserialization + /// Initial position in the topic + /// Consumer priority level + /// Whether to read from compacted topics + /// Pulsar subscription type + /// Allow out-of-order message delivery + /// Custom consumer configuration callback + public PulsarSubscription(SubscriptionName subscriptionName, ChannelName channelName, RoutingKey routingKey, + Func? getRequestType = null, int bufferSize = 1, int noOfPerformers = 1, TimeSpan? timeOut = null, + int requeueCount = -1, TimeSpan? requeueDelay = null, int unacceptableMessageLimit = 0, + MessagePumpType messagePumpType = MessagePumpType.Unknown, IAmAChannelFactory? channelFactory = null, + OnMissingChannel makeChannels = OnMissingChannel.Create, TimeSpan? emptyChannelDelay = null, + TimeSpan? channelFailureDelay = null, + ISchema>? schema = null, + SubscriptionInitialPosition initialPosition = SubscriptionInitialPosition.Earliest, + int priorityLevel = 0, + bool readCompacted = false, + SubscriptionType subscriptionType = SubscriptionType.Exclusive, + bool allowOutOfOrderDeliver = false, + Action>>? configuration = null) + : base(subscriptionName, channelName, routingKey, typeof(T), getRequestType, bufferSize, noOfPerformers, timeOut, requeueCount, + requeueDelay, unacceptableMessageLimit, messagePumpType, channelFactory, makeChannels, emptyChannelDelay, + channelFailureDelay, schema, initialPosition, priorityLevel, readCompacted, subscriptionType, + allowOutOfOrderDeliver, configuration) + { + + } +} diff --git a/tests/Paramore.Brighter.Pulsar.Tests/MessageDispatch/When_building_a_dispatcher.cs b/tests/Paramore.Brighter.Pulsar.Tests/MessageDispatch/When_building_a_dispatcher.cs new file mode 100644 index 0000000000..083174f421 --- /dev/null +++ b/tests/Paramore.Brighter.Pulsar.Tests/MessageDispatch/When_building_a_dispatcher.cs @@ -0,0 +1,98 @@ +using Microsoft.Extensions.DependencyInjection; +using Paramore.Brighter.Extensions.DependencyInjection; +using Paramore.Brighter.MessagingGateway.Pulsar; +using Paramore.Brighter.Observability; +using Paramore.Brighter.Pulsar.Tests.TestDoubles; +using Paramore.Brighter.Pulsar.Tests.Utils; +using Paramore.Brighter.ServiceActivator; +using Polly; +using Polly.Registry; +using Xunit; + +namespace Paramore.Brighter.Pulsar.Tests.MessageDispatch; + +[Collection("CommandProcessor")] +public class DispatchBuilderTests : IDisposable +{ + private readonly IAmADispatchBuilder _builder; + private Dispatcher? _dispatcher; + + public DispatchBuilderTests() + { + var messageMapperRegistry = new MessageMapperRegistry( + new SimpleMessageMapperFactory(_ => new MyEventMessageMapper()), + null); + messageMapperRegistry.Register(); + + var connection = GatewayFactory.CreateConnection(); + var consumerFactory = new PulsarMessageConsumerFactory(connection); + var container = new ServiceCollection(); + + var tracer = new BrighterTracer(TimeProvider.System); + const InstrumentationOptions instrumentationOptions = InstrumentationOptions.All; + + var commandProcessor = CommandProcessorBuilder.StartNew() + .Handlers(new HandlerConfiguration(new SubscriberRegistry(), new ServiceProviderHandlerFactory(container.BuildServiceProvider()))) + .DefaultResilience() + .NoExternalBus() + .ConfigureInstrumentation(tracer, instrumentationOptions) + .RequestContextFactory(new InMemoryRequestContextFactory()) + .RequestSchedulerFactory(new InMemorySchedulerFactory()) + .Build(); + + _builder = DispatchBuilder.StartNew() + .CommandProcessor(commandProcessor, + new InMemoryRequestContextFactory() + ) + .MessageMappers(messageMapperRegistry, null, null, null) + .ChannelFactory(new PulsarChannelFactory(consumerFactory)) + .Subscriptions([ + new PulsarSubscription( + new SubscriptionName("foo"), + new ChannelName("mary"), + new RoutingKey("bt_building_dispatch"), + messagePumpType: MessagePumpType.Reactor, + timeOut: TimeSpan.FromMilliseconds(200)), + new PulsarSubscription( + new SubscriptionName("bar"), + new ChannelName("alice"), + new RoutingKey("bt_building_dispatch"), + messagePumpType: MessagePumpType.Reactor, + timeOut: TimeSpan.FromMilliseconds(200)) + ]) + .ConfigureInstrumentation(tracer); + } + + [Fact] + public void When_Building_A_Dispatcher() + { + _dispatcher = _builder.Build(); + + Assert.NotNull(_dispatcher); + Assert.NotNull(GetConnection("foo")); + Assert.NotNull(GetConnection("bar")); + Assert.Equal(DispatcherState.DS_AWAITING, _dispatcher.State); + + Thread.Sleep(1000); + + _dispatcher.Receive(); + + Thread.Sleep(1000); + + Assert.Equal(DispatcherState.DS_RUNNING, _dispatcher.State); + + _dispatcher.End().GetAwaiter().GetResult(); + + Assert.Equal(DispatcherState.DS_STOPPED, _dispatcher.State); + } + + public void Dispose() + { + CommandProcessor.ClearServiceBus(); + } + + private Subscription? GetConnection(string name) + { + return _dispatcher!.Subscriptions.SingleOrDefault(conn => conn.Name == name); + } +} diff --git a/tests/Paramore.Brighter.Pulsar.Tests/MessageDispatch/When_building_a_dispatcher_async.cs b/tests/Paramore.Brighter.Pulsar.Tests/MessageDispatch/When_building_a_dispatcher_async.cs new file mode 100644 index 0000000000..65237a537b --- /dev/null +++ b/tests/Paramore.Brighter.Pulsar.Tests/MessageDispatch/When_building_a_dispatcher_async.cs @@ -0,0 +1,98 @@ +using Microsoft.Extensions.DependencyInjection; +using Paramore.Brighter.Extensions.DependencyInjection; +using Paramore.Brighter.MessagingGateway.Pulsar; +using Paramore.Brighter.Observability; +using Paramore.Brighter.Pulsar.Tests.TestDoubles; +using Paramore.Brighter.Pulsar.Tests.Utils; +using Paramore.Brighter.ServiceActivator; +using Polly; +using Polly.Registry; +using Xunit; + +namespace Paramore.Brighter.Pulsar.Tests.MessageDispatch; + +[Collection("CommandProcessor")] +public class DispatchBuilderTestsAsync : IDisposable +{ + private readonly IAmADispatchBuilder _builder; + private Dispatcher? _dispatcher; + + public DispatchBuilderTestsAsync() + { + var messageMapperRegistry = new MessageMapperRegistry( + null, + new SimpleMessageMapperFactoryAsync(_ => new MyEventMessageMapperAsync())); + messageMapperRegistry.RegisterAsync(); + + var connection = GatewayFactory.CreateConnection(); + + var consumerFactory = new PulsarMessageConsumerFactory(connection); + var container = new ServiceCollection(); + + var tracer = new BrighterTracer(TimeProvider.System); + var instrumentationOptions = InstrumentationOptions.All; + + var commandProcessor = CommandProcessorBuilder.StartNew() + .Handlers(new HandlerConfiguration(new SubscriberRegistry(), new ServiceProviderHandlerFactory(container.BuildServiceProvider()))) + .DefaultResilience() + .NoExternalBus() + .ConfigureInstrumentation(tracer, instrumentationOptions) + .RequestContextFactory(new InMemoryRequestContextFactory()) + .RequestSchedulerFactory(new InMemorySchedulerFactory()) + .Build(); + + _builder = DispatchBuilder.StartNew() + .CommandProcessor(commandProcessor, + new InMemoryRequestContextFactory() + ) + .MessageMappers(null!, messageMapperRegistry, null, new EmptyMessageTransformerFactoryAsync()) + .ChannelFactory(new PulsarChannelFactory(consumerFactory)) + .Subscriptions([ + new PulsarSubscription( + new SubscriptionName("foo"), + new ChannelName("mary"), + new RoutingKey("bt_building_dispatch_async"), + messagePumpType: MessagePumpType.Proactor, + timeOut: TimeSpan.FromMilliseconds(200)), + new PulsarSubscription( + new SubscriptionName("bar"), + new ChannelName("alice"), + new RoutingKey("bt_building_dispatch_async"), + messagePumpType: MessagePumpType.Proactor, + timeOut: TimeSpan.FromMilliseconds(200)) + ]) + .ConfigureInstrumentation(tracer, instrumentationOptions); + } + + [Fact] + public async Task When_Building_A_Dispatcher_With_Async() + { + _dispatcher = _builder.Build(); + + Assert.NotNull(_dispatcher); + Assert.NotNull(GetConnection("foo")); + Assert.NotNull(GetConnection("bar")); + + Assert.Equal(DispatcherState.DS_AWAITING, _dispatcher.State); + + await Task.Delay(1000); + + _dispatcher.Receive(); + + await Task.Delay(1000); + + Assert.Equal(DispatcherState.DS_RUNNING, _dispatcher.State); + + await _dispatcher.End(); + } + + public void Dispose() + { + CommandProcessor.ClearServiceBus(); + } + + private Subscription? GetConnection(string name) + { + return _dispatcher!.Subscriptions.SingleOrDefault(conn => conn.Name == name); + } +} diff --git a/tests/Paramore.Brighter.Pulsar.Tests/MessageDispatch/When_building_a_dispatcher_with_named_gateway.cs b/tests/Paramore.Brighter.Pulsar.Tests/MessageDispatch/When_building_a_dispatcher_with_named_gateway.cs new file mode 100644 index 0000000000..d73adc253c --- /dev/null +++ b/tests/Paramore.Brighter.Pulsar.Tests/MessageDispatch/When_building_a_dispatcher_with_named_gateway.cs @@ -0,0 +1,78 @@ +using Microsoft.Extensions.DependencyInjection; +using Paramore.Brighter.Extensions.DependencyInjection; +using Paramore.Brighter.MessagingGateway.Pulsar; +using Paramore.Brighter.Observability; +using Paramore.Brighter.Pulsar.Tests.TestDoubles; +using Paramore.Brighter.Pulsar.Tests.Utils; +using Paramore.Brighter.ServiceActivator; +using Polly; +using Polly.Registry; +using Xunit; + +namespace Paramore.Brighter.Pulsar.Tests.MessageDispatch; + +[Collection("CommandProcessor")] +public class DispatchBuilderWithNamedGateway : IDisposable +{ + private readonly IAmADispatchBuilder _builder; + private Dispatcher? _dispatcher; + + public DispatchBuilderWithNamedGateway() + { + var messageMapperRegistry = new MessageMapperRegistry( + new SimpleMessageMapperFactory(_ => new MyEventMessageMapper()), + null + ); + messageMapperRegistry.Register(); + + var connection = GatewayFactory.CreateConnection(); + var consumerFactory = new PulsarMessageConsumerFactory(connection); + + var container = new ServiceCollection(); + var tracer = new BrighterTracer(TimeProvider.System); + var instrumentationOptions = InstrumentationOptions.All; + + var commandProcessor = CommandProcessorBuilder.StartNew() + .Handlers(new HandlerConfiguration(new SubscriberRegistry(), new ServiceProviderHandlerFactory(container.BuildServiceProvider()))) + .DefaultResilience() + .NoExternalBus() + .ConfigureInstrumentation(tracer, instrumentationOptions) + .RequestContextFactory(new InMemoryRequestContextFactory()) + .RequestSchedulerFactory(new InMemorySchedulerFactory()) + .Build(); + + _builder = DispatchBuilder.StartNew() + .CommandProcessor(commandProcessor, + new InMemoryRequestContextFactory() + ) + .MessageMappers(messageMapperRegistry, null, new EmptyMessageTransformerFactory(), null) + .ChannelFactory(new PulsarChannelFactory(consumerFactory)) + .Subscriptions([ + new PulsarSubscription( + new SubscriptionName("foo"), + new ChannelName("mary"), + new RoutingKey("bt_named_gateway_dispatch"), + messagePumpType: MessagePumpType.Reactor, + timeOut: TimeSpan.FromMilliseconds(200)), + new PulsarSubscription( + new SubscriptionName("bar"), + new ChannelName("alice"), + new RoutingKey("bt_named_gateway_dispatch"), + messagePumpType: MessagePumpType.Reactor, + timeOut: TimeSpan.FromMilliseconds(200)) + ]) + .ConfigureInstrumentation(tracer, instrumentationOptions); + } + + [Fact] + public void When_building_a_dispatcher_with_named_gateway() + { + _dispatcher = _builder.Build(); + Assert.NotNull(_dispatcher); + } + + public void Dispose() + { + CommandProcessor.ClearServiceBus(); + } +} diff --git a/tests/Paramore.Brighter.Pulsar.Tests/MessagingGateway/Proactor/When_a_message_consumer_a_cloud_events_async.cs b/tests/Paramore.Brighter.Pulsar.Tests/MessagingGateway/Proactor/When_a_message_consumer_a_cloud_events_async.cs new file mode 100644 index 0000000000..2283ba2e9d --- /dev/null +++ b/tests/Paramore.Brighter.Pulsar.Tests/MessagingGateway/Proactor/When_a_message_consumer_a_cloud_events_async.cs @@ -0,0 +1,64 @@ +using Paramore.Brighter.MessagingGateway.Pulsar; +using Paramore.Brighter.Observability; +using Paramore.Brighter.Pulsar.Tests.Utils; +using Xunit; + +namespace Paramore.Brighter.Pulsar.Tests.MessagingGateway.Proactor; + +[Trait("Category", "Pulsar")] +public class BufferedConsumerCloudEventsTestsAsync : IAsyncDisposable +{ + private readonly RoutingKey _routingKey = new(Guid.NewGuid().ToString()); + private readonly IAmAMessageConsumerAsync _consumer; + private readonly IAmAMessageProducerAsync _producer; + private const int BatchSize = 3; + + public BufferedConsumerCloudEventsTestsAsync() + { + var connection = GatewayFactory.CreateConnection(); + var publication = new PulsarPublication { Topic = Guid.NewGuid().ToString() }; + var consumer = GatewayFactory.CreateConsumer(connection, publication); + var producer = GatewayFactory.CreateProducer(connection, publication); + + _consumer = new PulsarMessageConsumer(consumer); + _producer = new PulsarMessageProducer(producer, publication, TimeProvider.System, InstrumentationOptions.None); + } + + [Fact] + public async Task When_uses_cloud_events_async() + { + //Post one more than batch size messages + var messageOne = new Message( + new MessageHeader(Guid.NewGuid().ToString(), _routingKey, MessageType.MT_COMMAND) + { + Type = new CloudEventsType($"Type{Guid.NewGuid():N}"), + Subject = $"Subject{Guid.NewGuid():N}", + Source = new Uri($"/component/{Guid.NewGuid()}", UriKind.RelativeOrAbsolute), + DataSchema = new Uri("https://example.com/storage/tenant/container", UriKind.RelativeOrAbsolute) + }, new MessageBody("test content One")); + + await _producer.SendAsync(messageOne); + + //let them arrive + await Task.Delay(5000); + + //Now retrieve messages from the consumer + var messages = await _consumer.ReceiveAsync(TimeSpan.FromSeconds(10)); + + //We should only have three messages + Assert.Single(messages); + + Assert.Equal(messageOne.Header.MessageId, messages[0].Header.MessageId); + Assert.Equal(messageOne.Header.Subject, messages[0].Header.Subject); + Assert.Equal(messageOne.Header.Type, messages[0].Header.Type); + Assert.Equal(messageOne.Header.Source, messages[0].Header.Source); + Assert.Equal(messageOne.Header.DataSchema, messages[0].Header.DataSchema); + } + + public async ValueTask DisposeAsync() + { + await _consumer.DisposeAsync(); + await _producer.DisposeAsync(); + } + +} diff --git a/tests/Paramore.Brighter.Pulsar.Tests/MessagingGateway/Proactor/When_a_message_consumer_reads_multiple_messages_async.cs b/tests/Paramore.Brighter.Pulsar.Tests/MessagingGateway/Proactor/When_a_message_consumer_reads_multiple_messages_async.cs new file mode 100644 index 0000000000..976a8f8f63 --- /dev/null +++ b/tests/Paramore.Brighter.Pulsar.Tests/MessagingGateway/Proactor/When_a_message_consumer_reads_multiple_messages_async.cs @@ -0,0 +1,62 @@ +using Paramore.Brighter.MessagingGateway.Pulsar; +using Paramore.Brighter.Observability; +using Paramore.Brighter.Pulsar.Tests.Utils; +using Xunit; + +namespace Paramore.Brighter.Pulsar.Tests.MessagingGateway.Proactor; + +[Trait("Category", "Pulsar")] +public class BufferedConsumerTestsAsync : IAsyncDisposable +{ + private readonly IAmAMessageProducerAsync _messageProducer; + private readonly IAmAMessageConsumerAsync _messageConsumer; + private readonly RoutingKey _routingKey = new(Guid.NewGuid().ToString()); + + public BufferedConsumerTestsAsync() + { + var connection = GatewayFactory.CreateConnection(); + var publication = new PulsarPublication { Topic = Guid.NewGuid().ToString() }; + var consumer = GatewayFactory.CreateConsumer(connection, publication); + var producer = GatewayFactory.CreateProducer(connection, publication); + + _messageConsumer = new PulsarMessageConsumer(consumer); + _messageProducer = new PulsarMessageProducer(producer, publication, TimeProvider.System, InstrumentationOptions.None); + } + + [Fact] + public async Task When_a_message_consumer_reads_multiple_messages_async() + { + //Post one more than batch size messages + var messageOne = new Message(new MessageHeader(Guid.NewGuid().ToString(), _routingKey, MessageType.MT_COMMAND), new MessageBody("test content One")); + await _messageProducer.SendAsync(messageOne); + var messageTwo = new Message(new MessageHeader(Guid.NewGuid().ToString(), _routingKey, MessageType.MT_COMMAND), new MessageBody("test content Two")); + await _messageProducer.SendAsync(messageTwo); + + //let them arrive + await Task.Delay(5000); + + //Now retrieve messages from the consumer + var messages = await _messageConsumer.ReceiveAsync(TimeSpan.FromMilliseconds(1000)); + + // We should get only one message + Assert.Single(messages); + + //ack those to remove from the queue + await _messageConsumer.AcknowledgeAsync(messages[0]); + + //Allow ack to register + await Task.Delay(1000); + + //Now retrieve again + messages = await _messageConsumer.ReceiveAsync(TimeSpan.FromMilliseconds(500)); + + //This time, just the one message + Assert.Single(messages); + } + + public async ValueTask DisposeAsync() + { + await _messageConsumer.DisposeAsync(); + await _messageProducer.DisposeAsync(); + } +} diff --git a/tests/Paramore.Brighter.Pulsar.Tests/MessagingGateway/Proactor/When_posting_a_message_via_the_messaging_gateway_async.cs b/tests/Paramore.Brighter.Pulsar.Tests/MessagingGateway/Proactor/When_posting_a_message_via_the_messaging_gateway_async.cs new file mode 100644 index 0000000000..56eeead590 --- /dev/null +++ b/tests/Paramore.Brighter.Pulsar.Tests/MessagingGateway/Proactor/When_posting_a_message_via_the_messaging_gateway_async.cs @@ -0,0 +1,94 @@ +using System.Net.Mime; +using System.Text.Json; +using Paramore.Brighter.JsonConverters; +using Paramore.Brighter.MessagingGateway.Pulsar; +using Paramore.Brighter.Observability; +using Paramore.Brighter.Pulsar.Tests.TestDoubles; +using Paramore.Brighter.Pulsar.Tests.Utils; +using Xunit; + +namespace Paramore.Brighter.Pulsar.Tests.MessagingGateway.Proactor; + +[Trait("Category", "Pulsar")] +public class MessageProducerSendAsyncTests : IAsyncDisposable +{ + private readonly Message _message; + private readonly IAmAChannelAsync _channel; + private readonly IAmAMessageProducerAsync _messageProducer; + private readonly MyCommand _myCommand; + private readonly Id _correlationId; + private readonly RoutingKey _replyTo; + + public MessageProducerSendAsyncTests() + { + _myCommand = new MyCommand { Value = "Test" }; + _correlationId = Id.Random(); + _replyTo = new RoutingKey("http:\\queueUrl"); + var contentType = new ContentType (MediaTypeNames.Text.Plain); + var channelName = Guid.NewGuid().ToString(); + var publication = new PulsarPublication{ Topic = Guid.NewGuid().ToString() }; + + var mqSubscription = new PulsarSubscription( + subscriptionName: new SubscriptionName(channelName), + channelName: new ChannelName(channelName), + routingKey: publication.Topic!, + messagePumpType: MessagePumpType.Proactor + ); + + _message = new Message( + new MessageHeader(_myCommand.Id, publication.Topic!, MessageType.MT_COMMAND, correlationId: _correlationId, + replyTo: new RoutingKey(_replyTo), contentType: contentType), + new MessageBody(JsonSerializer.Serialize((object)_myCommand, JsonSerialisationOptions.Options)) + ); + + var connection = GatewayFactory.CreateConnection(); + var channelFactory = new PulsarChannelFactory(new PulsarMessageConsumerFactory(connection)); + _channel = channelFactory.CreateAsyncChannel(mqSubscription); + _messageProducer = new PulsarMessageProducer( + GatewayFactory.CreateProducer(connection, publication), + publication, TimeProvider.System, InstrumentationOptions.None); + } + + [Fact] + public async Task When_posting_a_message_via_the_producer_async() + { + // arrange + _message.Header.Subject = "test subject"; + await _messageProducer.SendAsync(_message); + + await Task.Delay(1000); + + var message = await _channel.ReceiveAsync(TimeSpan.FromMilliseconds(5000)); + + // clear the queue + await _channel.AcknowledgeAsync(message); + + // should_send_the_message_to_aws_sqs + Assert.Equal(MessageType.MT_COMMAND, message.Header.MessageType); + + Assert.Equal(_myCommand.Id, message.Id); + Assert.False(message.Redelivered); + Assert.Equal(_myCommand.Id, message.Header.MessageId); + Assert.Contains(_messageProducer.Publication.Topic!.Value, message.Header.Topic.Value); + Assert.Equal(_correlationId, message.Header.CorrelationId); + Assert.Equal(_replyTo, message.Header.ReplyTo); + Assert.Equal(0, message.Header.HandledCount); + Assert.Equal(_message.Header.Subject, message.Header.Subject); + // allow for clock drift in the following test, more important to have a contemporary timestamp than anything + Assert.True((message.Header.TimeStamp) > (RoundToSeconds(DateTime.UtcNow.AddMinutes(-1)))); + Assert.Equal(TimeSpan.Zero, message.Header.Delayed); + // {"Id":"cd581ced-c066-4322-aeaf-d40944de8edd","Value":"Test","WasCancelled":false,"TaskCompleted":false} + Assert.Equal(_message.Body.Value, message.Body.Value); + } + + public async ValueTask DisposeAsync() + { + _channel.Dispose(); + await _messageProducer.DisposeAsync(); + } + + private static DateTime RoundToSeconds(DateTime dateTime) + { + return new DateTime(dateTime.Ticks - (dateTime.Ticks % TimeSpan.TicksPerSecond), dateTime.Kind); + } +} diff --git a/tests/Paramore.Brighter.Pulsar.Tests/MessagingGateway/Proactor/When_requeueing_a_message_async.cs b/tests/Paramore.Brighter.Pulsar.Tests/MessagingGateway/Proactor/When_requeueing_a_message_async.cs new file mode 100644 index 0000000000..c17c39f022 --- /dev/null +++ b/tests/Paramore.Brighter.Pulsar.Tests/MessagingGateway/Proactor/When_requeueing_a_message_async.cs @@ -0,0 +1,72 @@ +using System.Net.Mime; +using System.Text.Json; +using Paramore.Brighter.JsonConverters; +using Paramore.Brighter.MessagingGateway.Pulsar; +using Paramore.Brighter.Observability; +using Paramore.Brighter.Pulsar.Tests.TestDoubles; +using Paramore.Brighter.Pulsar.Tests.Utils; +using Xunit; + +namespace Paramore.Brighter.Pulsar.Tests.MessagingGateway.Proactor; + +[Trait("Category", "Pulsar")] +public class MessageProducerRequeueTestsAsync : IAsyncDisposable +{ + private readonly IAmAMessageProducerAsync _sender; + private Message? _requeuedMessage; + private Message? _receivedMessage; + private readonly IAmAChannelAsync _channel; + private readonly Message _message; + + public MessageProducerRequeueTestsAsync() + { + const string replyTo = "http:\\queueUrl"; + var myCommand = new MyCommand () { Value = "Test" }; + string correlationId = Guid.NewGuid().ToString(); + var contentType = new ContentType(MediaTypeNames.Text.Plain); + var channelName = Guid.NewGuid().ToString(); + var routingKey = new RoutingKey(Guid.NewGuid().ToString()); + + var subscription = new PulsarSubscription( + subscriptionName: new SubscriptionName(channelName), + channelName: new ChannelName(channelName), + routingKey: routingKey, + messagePumpType: MessagePumpType.Proactor, + makeChannels: OnMissingChannel.Create + ); + + _message = new Message( + new MessageHeader(myCommand.Id, routingKey, MessageType.MT_COMMAND, correlationId: correlationId, + replyTo: new RoutingKey(replyTo), contentType: contentType), + new MessageBody(JsonSerializer.Serialize((object)myCommand, JsonSerialisationOptions.Options)) + ); + + var connection = GatewayFactory.CreateConnection(); + + var channelFactory = new PulsarChannelFactory(new PulsarMessageConsumerFactory(connection)); + var publication = new PulsarPublication { Topic = routingKey }; + _sender = new PulsarMessageProducer(GatewayFactory.CreateProducer(connection, publication), publication, + TimeProvider.System, InstrumentationOptions.None); + _channel = channelFactory.CreateAsyncChannel(subscription); + } + + [Fact] + public async Task When_requeueing_a_message_async() + { + await _sender.SendAsync(_message); + _receivedMessage = await _channel.ReceiveAsync(TimeSpan.FromMilliseconds(5000)); + await _channel.RequeueAsync(_receivedMessage); + + _requeuedMessage = await _channel.ReceiveAsync(TimeSpan.FromMilliseconds(5000)); + + await _channel.AcknowledgeAsync(_requeuedMessage); + + Assert.Equal(_receivedMessage.Body.Value, _requeuedMessage.Body.Value); + } + + public async ValueTask DisposeAsync() + { + await _sender.DisposeAsync(); + _channel.Dispose(); + } +} diff --git a/tests/Paramore.Brighter.Pulsar.Tests/MessagingGateway/Reactor/When_a_message_consumer_a_cloud_events.cs b/tests/Paramore.Brighter.Pulsar.Tests/MessagingGateway/Reactor/When_a_message_consumer_a_cloud_events.cs new file mode 100644 index 0000000000..63981a2a0f --- /dev/null +++ b/tests/Paramore.Brighter.Pulsar.Tests/MessagingGateway/Reactor/When_a_message_consumer_a_cloud_events.cs @@ -0,0 +1,62 @@ +using Paramore.Brighter.MessagingGateway.Pulsar; +using Paramore.Brighter.Observability; +using Paramore.Brighter.Pulsar.Tests.Utils; +using Xunit; + +namespace Paramore.Brighter.Pulsar.Tests.MessagingGateway.Reactor; + +[Trait("Category", "Pulsar")] +public class BufferedConsumerCloudEventsTests : IDisposable +{ + private readonly RoutingKey _routingKey = new(Guid.NewGuid().ToString()); + private readonly IAmAMessageConsumerSync _consumer; + private readonly IAmAMessageProducerSync _producer; + + public BufferedConsumerCloudEventsTests() + { + var connection = GatewayFactory.CreateConnection(); + var publication = new PulsarPublication { Topic = Guid.NewGuid().ToString() }; + var consumer = GatewayFactory.CreateConsumer(connection, publication); + var producer = GatewayFactory.CreateProducer(connection, publication); + + _consumer = new PulsarMessageConsumer(consumer); + _producer = new PulsarMessageProducer(producer, publication, TimeProvider.System, InstrumentationOptions.None); + } + + [Fact] + public void When_uses_cloud_events() + { + //Post one more than batch size messages + var messageOne = new Message( + new MessageHeader(Guid.NewGuid().ToString(), _routingKey, MessageType.MT_COMMAND) + { + Type = new CloudEventsType($"Type{Guid.NewGuid():N}"), + Subject = $"Subject{Guid.NewGuid():N}", + Source = new Uri($"/component/{Guid.NewGuid()}", UriKind.RelativeOrAbsolute), + DataSchema = new Uri("https://example.com/storage/tenant/container", UriKind.RelativeOrAbsolute) + }, new MessageBody("test content One")); + + _producer.Send(messageOne); + + //let them arrive + Thread.Sleep(5000); + + //Now retrieve messages from the consumer + var messages = _consumer.Receive(TimeSpan.FromMilliseconds(1000)); + + //We should only have three messages + Assert.Single(messages); + + Assert.Equal(messageOne.Header.MessageId, messages[0].Header.MessageId); + Assert.Equal(messageOne.Header.Subject, messages[0].Header.Subject); + Assert.Equal(messageOne.Header.Type, messages[0].Header.Type); + Assert.Equal(messageOne.Header.Source, messages[0].Header.Source); + Assert.Equal(messageOne.Header.DataSchema, messages[0].Header.DataSchema); + } + + public void Dispose() + { + _consumer.Dispose(); + _producer.Dispose(); + } +} diff --git a/tests/Paramore.Brighter.Pulsar.Tests/MessagingGateway/Reactor/When_a_message_consumer_reads_multiple_messages.cs b/tests/Paramore.Brighter.Pulsar.Tests/MessagingGateway/Reactor/When_a_message_consumer_reads_multiple_messages.cs new file mode 100644 index 0000000000..c7ed84dbaf --- /dev/null +++ b/tests/Paramore.Brighter.Pulsar.Tests/MessagingGateway/Reactor/When_a_message_consumer_reads_multiple_messages.cs @@ -0,0 +1,63 @@ +using Paramore.Brighter.MessagingGateway.Pulsar; +using Paramore.Brighter.Observability; +using Paramore.Brighter.Pulsar.Tests.Utils; +using Xunit; + +namespace Paramore.Brighter.Pulsar.Tests.MessagingGateway.Reactor; + +[Trait("Category", "Pulsar")] +public class BufferedConsumerTests : IDisposable +{ + private readonly IAmAMessageProducerSync _messageProducer; + private readonly IAmAMessageConsumerSync _messageConsumer; + private readonly RoutingKey _routingKey = new(Guid.NewGuid().ToString()); + private const int BatchSize = 3; + + public BufferedConsumerTests() + { + var connection = GatewayFactory.CreateConnection(); + var publication = new PulsarPublication { Topic = Guid.NewGuid().ToString() }; + var consumer = GatewayFactory.CreateConsumer(connection, publication); + var producer = GatewayFactory.CreateProducer(connection, publication); + + _messageConsumer = new PulsarMessageConsumer(consumer); + _messageProducer = new PulsarMessageProducer(producer, publication, TimeProvider.System, InstrumentationOptions.None); + } + + [Fact] + public void When_a_message_consumer_reads_multiple_messages() + { + //Post one more than batch size messages + var messageOne = new Message(new MessageHeader(Guid.NewGuid().ToString(), _routingKey, MessageType.MT_COMMAND), new MessageBody("test content One")); + _messageProducer.Send(messageOne); + var messageTwo = new Message(new MessageHeader(Guid.NewGuid().ToString(), _routingKey, MessageType.MT_COMMAND), new MessageBody("test content Two")); + _messageProducer.Send(messageTwo); + + //let them arrive + Thread.Sleep(5000); + + //Now retrieve messages from the consumer + var messages = _messageConsumer.Receive(TimeSpan.FromMilliseconds(1000)); + + // We should get only one message + Assert.Single(messages); + + //ack those to remove from the queue + _messageConsumer.Acknowledge(messages[0]); + + //Allow ack to register + Thread.Sleep(1000); + + //Now retrieve again + messages = _messageConsumer.Receive(TimeSpan.FromMilliseconds(500)); + + //This time, just the one message + Assert.Single(messages); + } + + public void Dispose() + { + _messageProducer.Dispose(); + _messageConsumer.Dispose(); + } +} diff --git a/tests/Paramore.Brighter.Pulsar.Tests/MessagingGateway/Reactor/When_posting_a_message_via_the_messaging_gateway.cs b/tests/Paramore.Brighter.Pulsar.Tests/MessagingGateway/Reactor/When_posting_a_message_via_the_messaging_gateway.cs new file mode 100644 index 0000000000..3637942a62 --- /dev/null +++ b/tests/Paramore.Brighter.Pulsar.Tests/MessagingGateway/Reactor/When_posting_a_message_via_the_messaging_gateway.cs @@ -0,0 +1,94 @@ +using System.Net.Mime; +using System.Text.Json; +using Paramore.Brighter.JsonConverters; +using Paramore.Brighter.MessagingGateway.Pulsar; +using Paramore.Brighter.Observability; +using Paramore.Brighter.Pulsar.Tests.TestDoubles; +using Paramore.Brighter.Pulsar.Tests.Utils; +using Xunit; + +namespace Paramore.Brighter.Pulsar.Tests.MessagingGateway.Reactor; + +[Trait("Category", "Pulsar")] +public class MessageProducerSendTests : IDisposable +{ + private readonly Message _message; + private readonly IAmAChannelSync _channel; + private readonly IAmAMessageProducerSync _messageProducer; + private readonly MyCommand _myCommand; + private readonly Id _correlationId; + private readonly RoutingKey _replyTo; + + public MessageProducerSendTests() + { + _myCommand = new MyCommand { Value = "Test" }; + _correlationId = Id.Random(); + _replyTo = new RoutingKey("http:\\queueUrl"); + ContentType contentType = new(MediaTypeNames.Text.Plain); + var channelName = Guid.NewGuid().ToString(); + var publication = new PulsarPublication { Topic = Guid.NewGuid().ToString() }; + + var mqSubscription = new PulsarSubscription( + subscriptionName: new SubscriptionName(channelName), + channelName: new ChannelName(channelName), + routingKey: publication.Topic!, + messagePumpType: MessagePumpType.Proactor + ); + + _message = new Message( + new MessageHeader(_myCommand.Id, publication.Topic!, MessageType.MT_COMMAND, correlationId: _correlationId, + replyTo: new RoutingKey(_replyTo), contentType: contentType), + new MessageBody(JsonSerializer.Serialize((object)_myCommand, JsonSerialisationOptions.Options)) + ); + + var connection = GatewayFactory.CreateConnection(); + var channelFactory = new PulsarChannelFactory(new PulsarMessageConsumerFactory(connection)); + _channel = channelFactory.CreateSyncChannel(mqSubscription); + _messageProducer = new PulsarMessageProducer( + GatewayFactory.CreateProducer(connection, publication), publication, + TimeProvider.System, InstrumentationOptions.None); + } + + [Fact] + public void When_posting_a_message_via_the_producer_async() + { + // arrange + _message.Header.Subject = "test subject"; + _messageProducer.Send(_message); + + Thread.Sleep(1000); + + var message = _channel.Receive(TimeSpan.FromMilliseconds(5000)); + + // clear the queue + _channel.Acknowledge(message); + + // should_send_the_message_to_aws_sqs + Assert.Equal(MessageType.MT_COMMAND, message.Header.MessageType); + + Assert.Equal(_myCommand.Id, message.Id); + Assert.False(message.Redelivered); + Assert.Equal(_myCommand.Id, message.Header.MessageId); + Assert.Contains(_messageProducer.Publication.Topic!.Value, message.Header.Topic.Value); + Assert.Equal(_correlationId, message.Header.CorrelationId); + Assert.Equal(_replyTo, message.Header.ReplyTo); + Assert.Equal(0, message.Header.HandledCount); + Assert.Equal(_message.Header.Subject, message.Header.Subject); + // allow for clock drift in the following test, more important to have a contemporary timestamp than anything + Assert.True((message.Header.TimeStamp) > (RoundToSeconds(DateTime.UtcNow.AddMinutes(-1)))); + Assert.Equal(TimeSpan.Zero, message.Header.Delayed); + // {"Id":"cd581ced-c066-4322-aeaf-d40944de8edd","Value":"Test","WasCancelled":false,"TaskCompleted":false} + Assert.Equal(_message.Body.Value, message.Body.Value); + } + + public void Dispose() + { + _channel.Dispose(); + _messageProducer.Dispose(); + } + + private static DateTime RoundToSeconds(DateTime dateTime) + { + return new DateTime(dateTime.Ticks - (dateTime.Ticks % TimeSpan.TicksPerSecond), dateTime.Kind); + } +} diff --git a/tests/Paramore.Brighter.Pulsar.Tests/MessagingGateway/Reactor/When_requeueing_a_message.cs b/tests/Paramore.Brighter.Pulsar.Tests/MessagingGateway/Reactor/When_requeueing_a_message.cs new file mode 100644 index 0000000000..2d5a9a47ba --- /dev/null +++ b/tests/Paramore.Brighter.Pulsar.Tests/MessagingGateway/Reactor/When_requeueing_a_message.cs @@ -0,0 +1,72 @@ +using System.Net.Mime; +using System.Text.Json; +using Paramore.Brighter.JsonConverters; +using Paramore.Brighter.MessagingGateway.Pulsar; +using Paramore.Brighter.Observability; +using Paramore.Brighter.Pulsar.Tests.TestDoubles; +using Paramore.Brighter.Pulsar.Tests.Utils; +using Xunit; + +namespace Paramore.Brighter.Pulsar.Tests.MessagingGateway.Reactor; + +[Trait("Category", "Pulsar")] +public class MessageProducerRequeueTests : IDisposable +{ + private readonly IAmAMessageProducerSync _sender; + private Message? _requeuedMessage; + private Message? _receivedMessage; + private readonly IAmAChannelSync _channel; + private readonly Message _message; + + public MessageProducerRequeueTests() + { + const string replyTo = "http:\\queueUrl"; + var myCommand = new MyCommand { Value = "Test" }; + string correlationId = Guid.NewGuid().ToString(); + var contentType = new ContentType(MediaTypeNames.Text.Plain); + var channelName = Guid.NewGuid().ToString(); + var routingKey = new RoutingKey(Uuid.NewAsString()); + + var subscription = new PulsarSubscription( + subscriptionName: new SubscriptionName(channelName), + channelName: new ChannelName(channelName), + routingKey: routingKey, + messagePumpType: MessagePumpType.Proactor, + makeChannels: OnMissingChannel.Create + ); + + _message = new Message( + new MessageHeader(myCommand.Id, routingKey, MessageType.MT_COMMAND, correlationId: correlationId, + replyTo: new RoutingKey(replyTo), contentType: contentType), + new MessageBody(JsonSerializer.Serialize((object)myCommand, JsonSerialisationOptions.Options)) + ); + + var connection = GatewayFactory.CreateConnection(); + + var channelFactory = new PulsarChannelFactory(new PulsarMessageConsumerFactory(connection)); + var publication = new PulsarPublication { Topic = routingKey }; + _sender = new PulsarMessageProducer(GatewayFactory.CreateProducer(connection, publication), publication, + TimeProvider.System, InstrumentationOptions.None); + _channel = channelFactory.CreateSyncChannel(subscription); + } + + [Fact] + public void When_requeueing_a_message_async() + { + _sender.Send(_message); + _receivedMessage = _channel.Receive(TimeSpan.FromMilliseconds(5000)); + _channel.Requeue(_receivedMessage); + + _requeuedMessage = _channel.Receive(TimeSpan.FromMilliseconds(5000)); + + _channel.Acknowledge(_requeuedMessage); + + Assert.Equal(_receivedMessage.Body.Value, _requeuedMessage.Body.Value); + } + + public void Dispose() + { + _sender.Dispose(); + _channel.Dispose(); + } +} diff --git a/tests/Paramore.Brighter.Pulsar.Tests/Paramore.Brighter.Pulsar.Tests.csproj b/tests/Paramore.Brighter.Pulsar.Tests/Paramore.Brighter.Pulsar.Tests.csproj new file mode 100644 index 0000000000..d8cd435425 --- /dev/null +++ b/tests/Paramore.Brighter.Pulsar.Tests/Paramore.Brighter.Pulsar.Tests.csproj @@ -0,0 +1,36 @@ + + + + net9.0 + enable + enable + false + false + + + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + + + all + runtime; build; native; contentfiles; analyzers; buildtransitive + + + + + + + + + + + diff --git a/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/MyCommand.cs b/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/MyCommand.cs new file mode 100644 index 0000000000..0f18039f6e --- /dev/null +++ b/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/MyCommand.cs @@ -0,0 +1,30 @@ +#region Licence +/* The MIT License (MIT) +Copyright © 2014 Ian Cooper + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the “Software”), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in +all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +THE SOFTWARE. */ + +#endregion + +namespace Paramore.Brighter.Pulsar.Tests.TestDoubles; + +internal class MyCommand() : Command(Guid.NewGuid()) +{ + public string Value { get; set; } +} diff --git a/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/MyDeferredCommand.cs b/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/MyDeferredCommand.cs new file mode 100644 index 0000000000..fd625b85e5 --- /dev/null +++ b/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/MyDeferredCommand.cs @@ -0,0 +1,8 @@ +namespace Paramore.Brighter.Pulsar.Tests.TestDoubles; + +internal class MyDeferredCommand : Command +{ + public string Value { get; set; } + public MyDeferredCommand() : base(Guid.NewGuid()) { } + +} diff --git a/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/MyDeferredCommandHandler.cs b/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/MyDeferredCommandHandler.cs new file mode 100644 index 0000000000..487e2ffa39 --- /dev/null +++ b/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/MyDeferredCommandHandler.cs @@ -0,0 +1,13 @@ +using Paramore.Brighter.Actions; + +namespace Paramore.Brighter.Pulsar.Tests.TestDoubles; + +internal class MyDeferredCommandHandler : RequestHandler +{ + public int HandledCount { get; set; } = 0; + public override MyDeferredCommand Handle(MyDeferredCommand command) + { + //Just defer for ever + throw new DeferMessageAction(); + } +} diff --git a/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/MyDeferredCommandHandlerAsync.cs b/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/MyDeferredCommandHandlerAsync.cs new file mode 100644 index 0000000000..c22820f6a2 --- /dev/null +++ b/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/MyDeferredCommandHandlerAsync.cs @@ -0,0 +1,14 @@ +using Paramore.Brighter.Actions; + +namespace Paramore.Brighter.Pulsar.Tests.TestDoubles; + +internal class MyDeferredCommandHandlerAsync : RequestHandlerAsync +{ + public int HandledCount { get; set; } = 0; + + public override Task HandleAsync(MyDeferredCommand command, CancellationToken cancellationToken = default) + { + // Just defer forever + throw new DeferMessageAction(); + } +} diff --git a/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/MyDeferredCommandMessageMapper.cs b/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/MyDeferredCommandMessageMapper.cs new file mode 100644 index 0000000000..e709c56adb --- /dev/null +++ b/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/MyDeferredCommandMessageMapper.cs @@ -0,0 +1,23 @@ +using System.Text.Json; +using Paramore.Brighter.Extensions; +using Paramore.Brighter.JsonConverters; + +namespace Paramore.Brighter.Pulsar.Tests.TestDoubles; + +internal class MyDeferredCommandMessageMapper : IAmAMessageMapper +{ + public IRequestContext Context { get; set; } + + public Message MapToMessage(MyDeferredCommand request, Publication publication) + { + var header = new MessageHeader(messageId: request.Id, topic: publication.Topic, messageType: request.RequestToMessageType()); + var body = new MessageBody(System.Text.Json.JsonSerializer.Serialize(request, new JsonSerializerOptions(JsonSerializerDefaults.General))); + var message = new Message(header, body); + return message; + } + + public MyDeferredCommand MapToRequest(Message message) + { + return JsonSerializer.Deserialize(message.Body.Value, JsonSerialisationOptions.Options); + } +} diff --git a/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/MyDeferredCommandMessageMapperAsync.cs b/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/MyDeferredCommandMessageMapperAsync.cs new file mode 100644 index 0000000000..b8a24346b0 --- /dev/null +++ b/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/MyDeferredCommandMessageMapperAsync.cs @@ -0,0 +1,26 @@ +using System.Text.Json; +using Paramore.Brighter.Extensions; +using Paramore.Brighter.JsonConverters; + +namespace Paramore.Brighter.Pulsar.Tests.TestDoubles; + +internal class MyDeferredCommandMessageMapperAsync : IAmAMessageMapperAsync +{ + public IRequestContext Context { get; set; } + + public async Task MapToMessageAsync(MyDeferredCommand request, Publication publication, CancellationToken cancellationToken = default) + { + if (publication.Topic is null) throw new InvalidOperationException("Missing publication topic"); + + var header = new MessageHeader(messageId: request.Id, topic: publication.Topic, messageType: request.RequestToMessageType()); + var body = new MessageBody(await Task.Run(() => JsonSerializer.Serialize(request, new JsonSerializerOptions(JsonSerializerDefaults.General)))); + var message = new Message(header, body); + return message; + } + + public async Task MapToRequestAsync(Message message, CancellationToken cancellationToken = default) + { + var command = await Task.Run(() => JsonSerializer.Deserialize(message.Body.Value, JsonSerialisationOptions.Options), cancellationToken); + return command ?? new MyDeferredCommand(); + } +} diff --git a/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/MyEvent.cs b/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/MyEvent.cs new file mode 100644 index 0000000000..edd34866a8 --- /dev/null +++ b/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/MyEvent.cs @@ -0,0 +1,65 @@ +#region Licence +/* The MIT License (MIT) +Copyright © 2014 Ian Cooper + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the “Software”), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in +all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +THE SOFTWARE. */ + +#endregion + +namespace Paramore.Brighter.Pulsar.Tests.TestDoubles; + +internal class MyEvent : Event, IEquatable +{ + public int Data { get; private set; } + + public MyEvent() : base(Guid.NewGuid()) + { + Data = 7; + } + + public bool Equals(MyEvent other) + { + if (ReferenceEquals(null, other)) return false; + if (ReferenceEquals(this, other)) return true; + return Data == other.Data; + } + + public override bool Equals(object obj) + { + if (ReferenceEquals(null, obj)) return false; + if (ReferenceEquals(this, obj)) return true; + if (obj.GetType() != this.GetType()) return false; + return Equals((MyEvent)obj); + } + + public override int GetHashCode() + { + return Data; + } + + public static bool operator ==(MyEvent left, MyEvent right) + { + return Equals(left, right); + } + + public static bool operator !=(MyEvent left, MyEvent right) + { + return !Equals(left, right); + } +} diff --git a/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/MyEventMessageMapper.cs b/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/MyEventMessageMapper.cs new file mode 100644 index 0000000000..0a726dbbdf --- /dev/null +++ b/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/MyEventMessageMapper.cs @@ -0,0 +1,47 @@ +#region Licence +/* The MIT License (MIT) +Copyright © 2014 Ian Cooper + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the “Software”), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in +all copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED “AS IS”, WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN +THE SOFTWARE. */ + +#endregion + +using System.Text.Json; +using Paramore.Brighter.Extensions; +using Paramore.Brighter.JsonConverters; + +namespace Paramore.Brighter.Pulsar.Tests.TestDoubles; + +internal class MyEventMessageMapper : IAmAMessageMapper +{ + public IRequestContext Context { get; set; } + + public Message MapToMessage(MyEvent request, Publication publication) + { + var header = new MessageHeader(request.Id, topic:publication.Topic, request.RequestToMessageType()); + var body = new MessageBody(JsonSerializer.Serialize(request, JsonSerialisationOptions.Options)); + var message = new Message(header, body); + return message; + } + + public MyEvent MapToRequest(Message message) + { + return JsonSerializer.Deserialize(message.Body.Value, JsonSerialisationOptions.Options); + } +} diff --git a/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/MyEventMessageMapperAsync.cs b/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/MyEventMessageMapperAsync.cs new file mode 100644 index 0000000000..a491eafc5c --- /dev/null +++ b/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/MyEventMessageMapperAsync.cs @@ -0,0 +1,20 @@ +namespace Paramore.Brighter.Pulsar.Tests.TestDoubles; + +internal class MyEventMessageMapperAsync : IAmAMessageMapperAsync +{ + public IRequestContext Context { get; set; } + + public Task MapToMessageAsync(MyEvent request, Publication publication, CancellationToken cancellationToken = default) + { + var header = new MessageHeader(messageId: request.Id, topic: publication.Topic, messageType: MessageType.MT_EVENT); + var body = new MessageBody(request.ToString()); + var message = new Message(header, body); + return Task.FromResult(message); + } + + public Task MapToRequestAsync(Message message, CancellationToken cancellationToken = default) + { + var myEvent = new MyEvent { Id = message.Id }; + return Task.FromResult(myEvent); + } +} diff --git a/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/QuickHandlerFactory.cs b/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/QuickHandlerFactory.cs new file mode 100644 index 0000000000..02b36b09a5 --- /dev/null +++ b/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/QuickHandlerFactory.cs @@ -0,0 +1,11 @@ +namespace Paramore.Brighter.Pulsar.Tests.TestDoubles; + +internal class QuickHandlerFactory(Func handlerAction) : IAmAHandlerFactorySync +{ + public IHandleRequests Create(Type handlerType, IAmALifetime lifetime) + { + return handlerAction(); + } + + public void Release(IHandleRequests handler, IAmALifetime lifetime) { } +} diff --git a/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/QuickHandlerFactoryAsync.cs b/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/QuickHandlerFactoryAsync.cs new file mode 100644 index 0000000000..a9f0ff5a3c --- /dev/null +++ b/tests/Paramore.Brighter.Pulsar.Tests/TestDoubles/QuickHandlerFactoryAsync.cs @@ -0,0 +1,11 @@ +namespace Paramore.Brighter.Pulsar.Tests.TestDoubles; + +internal class QuickHandlerFactoryAsync(Func handlerAction) : IAmAHandlerFactoryAsync +{ + public IHandleRequestsAsync Create(Type handlerType, IAmALifetime lifetime) + { + return handlerAction(); + } + + public void Release(IHandleRequestsAsync handler, IAmALifetime lifetime) { } +} diff --git a/tests/Paramore.Brighter.Pulsar.Tests/Utils/GatewayFactory.cs b/tests/Paramore.Brighter.Pulsar.Tests/Utils/GatewayFactory.cs new file mode 100644 index 0000000000..493599d931 --- /dev/null +++ b/tests/Paramore.Brighter.Pulsar.Tests/Utils/GatewayFactory.cs @@ -0,0 +1,37 @@ +using System.Buffers; +using DotPulsar; +using DotPulsar.Abstractions; +using DotPulsar.Extensions; +using Paramore.Brighter.MessagingGateway.Pulsar; + +namespace Paramore.Brighter.Pulsar.Tests.Utils; + +public static class GatewayFactory +{ + public static PulsarMessagingGatewayConnection CreateConnection() + { + return new PulsarMessagingGatewayConnection + { + ServiceUrl = new Uri("pulsar://localhost:6650"), + }; + } + + public static PulsarBackgroundMessageConsumer CreateConsumer(PulsarMessagingGatewayConnection connection, Publication publication) + { + var background = new PulsarBackgroundMessageConsumer(1, connection.Create().NewConsumer(Schema.ByteSequence) + .Topic(publication.Topic!.Value) + .SubscriptionName(Guid.NewGuid().ToString()) + .Create()); + + background.Start(); + + return background; + } + + public static IProducer> CreateProducer(PulsarMessagingGatewayConnection connection, Publication publication) + { + return connection.Create().NewProducer(Schema.ByteSequence) + .Topic(publication.Topic!.Value) + .Create(); + } +}