Add unit tests for Router configuration and transport layers
- Implemented tests for RouterConfig, RoutingOptions, StaticInstanceConfig, and RouterConfigOptions to ensure default values are set correctly. - Added tests for RouterConfigProvider to validate configurations and ensure defaults are returned when no file is specified. - Created tests for ConfigValidationResult to check success and error scenarios. - Developed tests for ServiceCollectionExtensions to verify service registration for RouterConfig. - Introduced UdpTransportTests to validate serialization, connection, request-response, and error handling in UDP transport. - Added scripts for signing authority gaps and hashing DevPortal SDK snippets.
This commit is contained in:
StellaOps Bot
@@ -0,0 +1,111 @@
|
||||
using System.Text;
|
||||
using RabbitMQ.Client;
|
||||
using StellaOps.Router.Common.Enums;
|
||||
using StellaOps.Router.Common.Models;
|
||||
|
||||
namespace StellaOps.Router.Transport.RabbitMq;
|
||||
|
||||
/// <summary>
|
||||
/// Handles serialization and deserialization of frames for RabbitMQ transport.
|
||||
/// </summary>
|
||||
public static class RabbitMqFrameProtocol
|
||||
{
|
||||
/// <summary>
|
||||
/// Parses a frame from a RabbitMQ message.
|
||||
/// </summary>
|
||||
/// <param name="body">The message body.</param>
|
||||
/// <param name="properties">The message properties.</param>
|
||||
/// <returns>The parsed frame.</returns>
|
||||
public static Frame ParseFrame(ReadOnlyMemory<byte> body, IReadOnlyBasicProperties properties)
|
||||
{
|
||||
var frameType = ParseFrameType(properties.Type);
|
||||
var correlationId = properties.CorrelationId;
|
||||
|
||||
return new Frame
|
||||
{
|
||||
Type = frameType,
|
||||
CorrelationId = correlationId,
|
||||
Payload = body
|
||||
};
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Creates BasicProperties for a frame.
|
||||
/// </summary>
|
||||
/// <param name="frame">The frame to serialize.</param>
|
||||
/// <param name="replyTo">The reply queue name.</param>
|
||||
/// <param name="timeout">Optional timeout for the message.</param>
|
||||
/// <returns>The basic properties.</returns>
|
||||
public static BasicProperties CreateProperties(Frame frame, string? replyTo, TimeSpan? timeout = null)
|
||||
{
|
||||
var props = new BasicProperties
|
||||
{
|
||||
Type = frame.Type.ToString(),
|
||||
Timestamp = new AmqpTimestamp(DateTimeOffset.UtcNow.ToUnixTimeSeconds()),
|
||||
DeliveryMode = DeliveryModes.Transient // Non-persistent (1)
|
||||
};
|
||||
|
||||
if (!string.IsNullOrEmpty(frame.CorrelationId))
|
||||
{
|
||||
props.CorrelationId = frame.CorrelationId;
|
||||
}
|
||||
|
||||
if (!string.IsNullOrEmpty(replyTo))
|
||||
{
|
||||
props.ReplyTo = replyTo;
|
||||
}
|
||||
|
||||
if (timeout.HasValue)
|
||||
{
|
||||
props.Expiration = ((int)timeout.Value.TotalMilliseconds).ToString();
|
||||
}
|
||||
|
||||
return props;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Parses a FrameType from the message Type property.
|
||||
/// </summary>
|
||||
private static FrameType ParseFrameType(string? type)
|
||||
{
|
||||
if (string.IsNullOrEmpty(type))
|
||||
{
|
||||
return FrameType.Request;
|
||||
}
|
||||
|
||||
if (Enum.TryParse<FrameType>(type, ignoreCase: true, out var result))
|
||||
{
|
||||
return result;
|
||||
}
|
||||
|
||||
return FrameType.Request;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Extracts the connection ID from message properties.
|
||||
/// </summary>
|
||||
/// <param name="properties">The message properties.</param>
|
||||
/// <returns>The connection ID.</returns>
|
||||
public static string ExtractConnectionId(IReadOnlyBasicProperties properties)
|
||||
{
|
||||
// Use ReplyTo as the basis for connection ID (identifies the instance)
|
||||
if (!string.IsNullOrEmpty(properties.ReplyTo))
|
||||
{
|
||||
// Extract instance ID from queue name like "stella.svc.{instanceId}"
|
||||
var parts = properties.ReplyTo.Split('.');
|
||||
if (parts.Length >= 3)
|
||||
{
|
||||
return $"rmq-{parts[^1]}";
|
||||
}
|
||||
return $"rmq-{properties.ReplyTo}";
|
||||
}
|
||||
|
||||
// Fallback to correlation ID
|
||||
if (!string.IsNullOrEmpty(properties.CorrelationId))
|
||||
{
|
||||
return $"rmq-{properties.CorrelationId[..Math.Min(16, properties.CorrelationId.Length)]}";
|
||||
}
|
||||
|
||||
return $"rmq-{Guid.NewGuid():N}"[..32];
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,449 @@
|
||||
using System.Collections.Concurrent;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Options;
|
||||
using RabbitMQ.Client;
|
||||
using RabbitMQ.Client.Events;
|
||||
using StellaOps.Router.Common.Abstractions;
|
||||
using StellaOps.Router.Common.Enums;
|
||||
using StellaOps.Router.Common.Models;
|
||||
|
||||
namespace StellaOps.Router.Transport.RabbitMq;
|
||||
|
||||
/// <summary>
|
||||
/// RabbitMQ transport client implementation for microservices.
|
||||
/// </summary>
|
||||
public sealed class RabbitMqTransportClient : ITransportClient, IMicroserviceTransport, IAsyncDisposable
|
||||
{
|
||||
private readonly RabbitMqTransportOptions _options;
|
||||
private readonly ILogger<RabbitMqTransportClient> _logger;
|
||||
private readonly ConcurrentDictionary<Guid, TaskCompletionSource<Frame>> _pendingRequests = new();
|
||||
private readonly ConcurrentDictionary<string, CancellationTokenSource> _inflightHandlers = new();
|
||||
private readonly CancellationTokenSource _clientCts = new();
|
||||
private IConnection? _connection;
|
||||
private IChannel? _channel;
|
||||
private string? _responseQueueName;
|
||||
private string? _instanceId;
|
||||
private string? _gatewayNodeId;
|
||||
private bool _disposed;
|
||||
|
||||
/// <summary>
|
||||
/// Event raised when a REQUEST frame is received.
|
||||
/// </summary>
|
||||
public event Func<Frame, CancellationToken, Task<Frame>>? OnRequestReceived;
|
||||
|
||||
/// <summary>
|
||||
/// Event raised when a CANCEL frame is received.
|
||||
/// </summary>
|
||||
public event Func<Guid, string?, Task>? OnCancelReceived;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="RabbitMqTransportClient"/> class.
|
||||
/// </summary>
|
||||
public RabbitMqTransportClient(
|
||||
IOptions<RabbitMqTransportOptions> options,
|
||||
ILogger<RabbitMqTransportClient> logger)
|
||||
{
|
||||
_options = options.Value;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Connects to the gateway via RabbitMQ.
|
||||
/// </summary>
|
||||
/// <param name="instance">The instance descriptor.</param>
|
||||
/// <param name="endpoints">The endpoints to register.</param>
|
||||
/// <param name="cancellationToken">Cancellation token.</param>
|
||||
public async Task ConnectAsync(
|
||||
InstanceDescriptor instance,
|
||||
IReadOnlyList<EndpointDescriptor> endpoints,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
ObjectDisposedException.ThrowIf(_disposed, this);
|
||||
|
||||
_instanceId = _options.InstanceId ?? instance.InstanceId;
|
||||
_gatewayNodeId = _options.NodeId ?? "default";
|
||||
|
||||
var factory = new ConnectionFactory
|
||||
{
|
||||
HostName = _options.HostName,
|
||||
Port = _options.Port,
|
||||
VirtualHost = _options.VirtualHost,
|
||||
UserName = _options.UserName,
|
||||
Password = _options.Password,
|
||||
AutomaticRecoveryEnabled = _options.AutomaticRecoveryEnabled,
|
||||
NetworkRecoveryInterval = _options.NetworkRecoveryInterval
|
||||
};
|
||||
|
||||
if (_options.UseSsl)
|
||||
{
|
||||
factory.Ssl = new SslOption
|
||||
{
|
||||
Enabled = true,
|
||||
ServerName = _options.HostName,
|
||||
CertPath = _options.SslCertPath
|
||||
};
|
||||
}
|
||||
|
||||
_connection = await factory.CreateConnectionAsync(cancellationToken);
|
||||
_channel = await _connection.CreateChannelAsync(cancellationToken: cancellationToken);
|
||||
|
||||
// Set QoS
|
||||
await _channel.BasicQosAsync(
|
||||
prefetchSize: 0,
|
||||
prefetchCount: _options.PrefetchCount,
|
||||
global: false,
|
||||
cancellationToken: cancellationToken);
|
||||
|
||||
// Declare exchanges (should already exist from server, but declare for safety)
|
||||
await _channel.ExchangeDeclareAsync(
|
||||
exchange: _options.RequestExchange,
|
||||
type: ExchangeType.Direct,
|
||||
durable: true,
|
||||
autoDelete: false,
|
||||
cancellationToken: cancellationToken);
|
||||
|
||||
await _channel.ExchangeDeclareAsync(
|
||||
exchange: _options.ResponseExchange,
|
||||
type: ExchangeType.Topic,
|
||||
durable: true,
|
||||
autoDelete: false,
|
||||
cancellationToken: cancellationToken);
|
||||
|
||||
// Declare response queue for this instance
|
||||
_responseQueueName = $"{_options.QueuePrefix}.svc.{_instanceId}";
|
||||
await _channel.QueueDeclareAsync(
|
||||
queue: _responseQueueName,
|
||||
durable: _options.DurableQueues,
|
||||
exclusive: false,
|
||||
autoDelete: _options.AutoDeleteQueues,
|
||||
cancellationToken: cancellationToken);
|
||||
|
||||
// Bind to response exchange with instance ID as routing key
|
||||
await _channel.QueueBindAsync(
|
||||
queue: _responseQueueName,
|
||||
exchange: _options.ResponseExchange,
|
||||
routingKey: _instanceId,
|
||||
cancellationToken: cancellationToken);
|
||||
|
||||
// Start consuming responses
|
||||
var consumer = new AsyncEventingBasicConsumer(_channel);
|
||||
consumer.ReceivedAsync += OnMessageReceivedAsync;
|
||||
|
||||
await _channel.BasicConsumeAsync(
|
||||
queue: _responseQueueName,
|
||||
autoAck: true,
|
||||
consumer: consumer,
|
||||
cancellationToken: cancellationToken);
|
||||
|
||||
// Send HELLO frame
|
||||
var helloFrame = new Frame
|
||||
{
|
||||
Type = FrameType.Hello,
|
||||
CorrelationId = Guid.NewGuid().ToString("N"),
|
||||
Payload = ReadOnlyMemory<byte>.Empty
|
||||
};
|
||||
await SendToGatewayAsync(helloFrame, cancellationToken);
|
||||
|
||||
_logger.LogInformation(
|
||||
"Connected to RabbitMQ gateway at {Host}:{Port} as {ServiceName}/{Version}",
|
||||
_options.HostName,
|
||||
_options.Port,
|
||||
instance.ServiceName,
|
||||
instance.Version);
|
||||
}
|
||||
|
||||
private async Task OnMessageReceivedAsync(object sender, BasicDeliverEventArgs e)
|
||||
{
|
||||
try
|
||||
{
|
||||
var frame = RabbitMqFrameProtocol.ParseFrame(e.Body, e.BasicProperties);
|
||||
|
||||
switch (frame.Type)
|
||||
{
|
||||
case FrameType.Request:
|
||||
await HandleRequestFrameAsync(frame, _clientCts.Token);
|
||||
break;
|
||||
|
||||
case FrameType.Cancel:
|
||||
HandleCancelFrame(frame);
|
||||
break;
|
||||
|
||||
case FrameType.Response:
|
||||
if (frame.CorrelationId is not null &&
|
||||
Guid.TryParse(frame.CorrelationId, out var correlationId))
|
||||
{
|
||||
if (_pendingRequests.TryRemove(correlationId, out var tcs))
|
||||
{
|
||||
tcs.TrySetResult(frame);
|
||||
}
|
||||
}
|
||||
break;
|
||||
|
||||
default:
|
||||
_logger.LogWarning("Unexpected frame type {FrameType}", frame.Type);
|
||||
break;
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "Error processing RabbitMQ message");
|
||||
}
|
||||
|
||||
await Task.CompletedTask;
|
||||
}
|
||||
|
||||
private async Task HandleRequestFrameAsync(Frame frame, CancellationToken cancellationToken)
|
||||
{
|
||||
if (OnRequestReceived is null)
|
||||
{
|
||||
_logger.LogWarning("No request handler registered");
|
||||
return;
|
||||
}
|
||||
|
||||
var correlationId = frame.CorrelationId ?? Guid.NewGuid().ToString("N");
|
||||
|
||||
using var handlerCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
|
||||
_inflightHandlers[correlationId] = handlerCts;
|
||||
|
||||
try
|
||||
{
|
||||
var response = await OnRequestReceived(frame, handlerCts.Token);
|
||||
var responseFrame = response with { CorrelationId = correlationId };
|
||||
|
||||
if (!handlerCts.Token.IsCancellationRequested)
|
||||
{
|
||||
await SendToGatewayAsync(responseFrame, cancellationToken);
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
_logger.LogDebug("Request {CorrelationId} was cancelled", correlationId);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "Error handling request {CorrelationId}", correlationId);
|
||||
}
|
||||
finally
|
||||
{
|
||||
_inflightHandlers.TryRemove(correlationId, out _);
|
||||
}
|
||||
}
|
||||
|
||||
private void HandleCancelFrame(Frame frame)
|
||||
{
|
||||
if (frame.CorrelationId is null) return;
|
||||
|
||||
_logger.LogDebug("Received CANCEL for {CorrelationId}", frame.CorrelationId);
|
||||
|
||||
if (_inflightHandlers.TryGetValue(frame.CorrelationId, out var cts))
|
||||
{
|
||||
try
|
||||
{
|
||||
cts.Cancel();
|
||||
}
|
||||
catch (ObjectDisposedException)
|
||||
{
|
||||
// Already completed
|
||||
}
|
||||
}
|
||||
|
||||
if (Guid.TryParse(frame.CorrelationId, out var guid))
|
||||
{
|
||||
if (_pendingRequests.TryRemove(guid, out var tcs))
|
||||
{
|
||||
tcs.TrySetCanceled();
|
||||
}
|
||||
OnCancelReceived?.Invoke(guid, null);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task SendToGatewayAsync(Frame frame, CancellationToken cancellationToken)
|
||||
{
|
||||
ObjectDisposedException.ThrowIf(_disposed, this);
|
||||
|
||||
var properties = RabbitMqFrameProtocol.CreateProperties(
|
||||
frame,
|
||||
_responseQueueName,
|
||||
_options.DefaultTimeout);
|
||||
|
||||
await _channel!.BasicPublishAsync(
|
||||
exchange: _options.RequestExchange,
|
||||
routingKey: _gatewayNodeId!,
|
||||
mandatory: false,
|
||||
basicProperties: properties,
|
||||
body: frame.Payload,
|
||||
cancellationToken: cancellationToken);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<Frame> SendRequestAsync(
|
||||
ConnectionState connection,
|
||||
Frame requestFrame,
|
||||
TimeSpan timeout,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
ObjectDisposedException.ThrowIf(_disposed, this);
|
||||
|
||||
var correlationId = requestFrame.CorrelationId is not null &&
|
||||
Guid.TryParse(requestFrame.CorrelationId, out var parsed)
|
||||
? parsed
|
||||
: Guid.NewGuid();
|
||||
|
||||
var framedRequest = requestFrame with { CorrelationId = correlationId.ToString("N") };
|
||||
|
||||
using var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
|
||||
timeoutCts.CancelAfter(timeout);
|
||||
|
||||
var tcs = new TaskCompletionSource<Frame>(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
var registration = timeoutCts.Token.Register(() =>
|
||||
{
|
||||
if (_pendingRequests.TryRemove(correlationId, out var pendingTcs))
|
||||
{
|
||||
pendingTcs.TrySetCanceled(timeoutCts.Token);
|
||||
}
|
||||
});
|
||||
|
||||
_pendingRequests[correlationId] = tcs;
|
||||
|
||||
try
|
||||
{
|
||||
await SendToGatewayAsync(framedRequest, timeoutCts.Token);
|
||||
return await tcs.Task;
|
||||
}
|
||||
catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
throw new TimeoutException($"Request {correlationId} timed out after {timeout}");
|
||||
}
|
||||
finally
|
||||
{
|
||||
await registration.DisposeAsync();
|
||||
_pendingRequests.TryRemove(correlationId, out _);
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task SendCancelAsync(
|
||||
ConnectionState connection,
|
||||
Guid correlationId,
|
||||
string? reason = null)
|
||||
{
|
||||
ObjectDisposedException.ThrowIf(_disposed, this);
|
||||
|
||||
var cancelFrame = new Frame
|
||||
{
|
||||
Type = FrameType.Cancel,
|
||||
CorrelationId = correlationId.ToString("N"),
|
||||
Payload = ReadOnlyMemory<byte>.Empty
|
||||
};
|
||||
|
||||
await SendToGatewayAsync(cancelFrame, CancellationToken.None);
|
||||
_logger.LogDebug("Sent CANCEL for {CorrelationId}", correlationId);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task SendStreamingAsync(
|
||||
ConnectionState connection,
|
||||
Frame requestHeader,
|
||||
Stream requestBody,
|
||||
Func<Stream, Task> readResponseBody,
|
||||
PayloadLimits limits,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
// Streaming could be implemented by chunking messages, but for now we don't support it
|
||||
// This keeps RabbitMQ transport simple
|
||||
throw new NotSupportedException(
|
||||
"RabbitMQ transport does not currently support streaming. Use TCP or TLS transport for streaming.");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Sends a heartbeat.
|
||||
/// </summary>
|
||||
public async Task SendHeartbeatAsync(HeartbeatPayload heartbeat, CancellationToken cancellationToken)
|
||||
{
|
||||
var frame = new Frame
|
||||
{
|
||||
Type = FrameType.Heartbeat,
|
||||
CorrelationId = null,
|
||||
Payload = ReadOnlyMemory<byte>.Empty
|
||||
};
|
||||
|
||||
await SendToGatewayAsync(frame, cancellationToken);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Cancels all in-flight handlers.
|
||||
/// </summary>
|
||||
public void CancelAllInflight(string reason)
|
||||
{
|
||||
var count = 0;
|
||||
foreach (var cts in _inflightHandlers.Values)
|
||||
{
|
||||
try
|
||||
{
|
||||
cts.Cancel();
|
||||
count++;
|
||||
}
|
||||
catch (ObjectDisposedException)
|
||||
{
|
||||
// Already completed
|
||||
}
|
||||
}
|
||||
|
||||
if (count > 0)
|
||||
{
|
||||
_logger.LogInformation("Cancelled {Count} in-flight handlers: {Reason}", count, reason);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Disconnects from the gateway.
|
||||
/// </summary>
|
||||
public async Task DisconnectAsync()
|
||||
{
|
||||
CancelAllInflight("Shutdown");
|
||||
|
||||
// Cancel all pending requests
|
||||
foreach (var kvp in _pendingRequests)
|
||||
{
|
||||
if (_pendingRequests.TryRemove(kvp.Key, out var tcs))
|
||||
{
|
||||
tcs.TrySetCanceled();
|
||||
}
|
||||
}
|
||||
|
||||
await _clientCts.CancelAsync();
|
||||
|
||||
if (_channel is not null)
|
||||
{
|
||||
await _channel.CloseAsync();
|
||||
}
|
||||
|
||||
if (_connection is not null)
|
||||
{
|
||||
await _connection.CloseAsync();
|
||||
}
|
||||
|
||||
_logger.LogInformation("Disconnected from RabbitMQ gateway");
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
if (_disposed) return;
|
||||
_disposed = true;
|
||||
|
||||
await DisconnectAsync();
|
||||
|
||||
if (_channel is not null)
|
||||
{
|
||||
await _channel.DisposeAsync();
|
||||
}
|
||||
|
||||
if (_connection is not null)
|
||||
{
|
||||
await _connection.DisposeAsync();
|
||||
}
|
||||
|
||||
_clientCts.Dispose();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,102 @@
|
||||
namespace StellaOps.Router.Transport.RabbitMq;
|
||||
|
||||
/// <summary>
|
||||
/// Options for RabbitMQ transport configuration.
|
||||
/// </summary>
|
||||
public sealed class RabbitMqTransportOptions
|
||||
{
|
||||
/// <summary>
|
||||
/// Gets or sets the RabbitMQ host name.
|
||||
/// </summary>
|
||||
public string HostName { get; set; } = "localhost";
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the RabbitMQ port.
|
||||
/// </summary>
|
||||
public int Port { get; set; } = 5672;
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the RabbitMQ virtual host.
|
||||
/// </summary>
|
||||
public string VirtualHost { get; set; } = "/";
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the RabbitMQ username.
|
||||
/// </summary>
|
||||
public string UserName { get; set; } = "guest";
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the RabbitMQ password.
|
||||
/// </summary>
|
||||
public string Password { get; set; } = "guest";
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets whether to use SSL/TLS.
|
||||
/// </summary>
|
||||
public bool UseSsl { get; set; } = false;
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the SSL certificate path.
|
||||
/// </summary>
|
||||
public string? SslCertPath { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets whether queues should be durable.
|
||||
/// </summary>
|
||||
public bool DurableQueues { get; set; } = false;
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets whether queues should auto-delete on disconnect.
|
||||
/// </summary>
|
||||
public bool AutoDeleteQueues { get; set; } = true;
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the prefetch count (concurrent messages).
|
||||
/// </summary>
|
||||
public ushort PrefetchCount { get; set; } = 10;
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the exchange prefix.
|
||||
/// </summary>
|
||||
public string ExchangePrefix { get; set; } = "stella.router";
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the queue prefix.
|
||||
/// </summary>
|
||||
public string QueuePrefix { get; set; } = "stella";
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the request exchange name.
|
||||
/// </summary>
|
||||
public string RequestExchange => $"{ExchangePrefix}.requests";
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the response exchange name.
|
||||
/// </summary>
|
||||
public string ResponseExchange => $"{ExchangePrefix}.responses";
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the node ID for this gateway instance.
|
||||
/// </summary>
|
||||
public string? NodeId { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the instance ID for this microservice instance.
|
||||
/// </summary>
|
||||
public string? InstanceId { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets whether to use automatic recovery.
|
||||
/// </summary>
|
||||
public bool AutomaticRecoveryEnabled { get; set; } = true;
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the network recovery interval.
|
||||
/// </summary>
|
||||
public TimeSpan NetworkRecoveryInterval { get; set; } = TimeSpan.FromSeconds(5);
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the default request timeout.
|
||||
/// </summary>
|
||||
public TimeSpan DefaultTimeout { get; set; } = TimeSpan.FromSeconds(30);
|
||||
}
|
||||
@@ -0,0 +1,289 @@
|
||||
using System.Collections.Concurrent;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Options;
|
||||
using RabbitMQ.Client;
|
||||
using RabbitMQ.Client.Events;
|
||||
using StellaOps.Router.Common.Abstractions;
|
||||
using StellaOps.Router.Common.Enums;
|
||||
using StellaOps.Router.Common.Models;
|
||||
|
||||
namespace StellaOps.Router.Transport.RabbitMq;
|
||||
|
||||
/// <summary>
|
||||
/// RabbitMQ transport server implementation for the gateway.
|
||||
/// </summary>
|
||||
public sealed class RabbitMqTransportServer : ITransportServer, IAsyncDisposable
|
||||
{
|
||||
private readonly RabbitMqTransportOptions _options;
|
||||
private readonly ILogger<RabbitMqTransportServer> _logger;
|
||||
private readonly ConcurrentDictionary<string, (string ReplyTo, ConnectionState State)> _connections = new();
|
||||
private readonly string _nodeId;
|
||||
private IConnection? _connection;
|
||||
private IChannel? _channel;
|
||||
private string? _requestQueueName;
|
||||
private bool _disposed;
|
||||
|
||||
/// <summary>
|
||||
/// Event raised when a connection is established (on first HELLO).
|
||||
/// </summary>
|
||||
public event Action<string, ConnectionState>? OnConnection;
|
||||
|
||||
/// <summary>
|
||||
/// Event raised when a connection is lost.
|
||||
/// </summary>
|
||||
public event Action<string>? OnDisconnection;
|
||||
|
||||
/// <summary>
|
||||
/// Event raised when a frame is received.
|
||||
/// </summary>
|
||||
public event Action<string, Frame>? OnFrame;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="RabbitMqTransportServer"/> class.
|
||||
/// </summary>
|
||||
public RabbitMqTransportServer(
|
||||
IOptions<RabbitMqTransportOptions> options,
|
||||
ILogger<RabbitMqTransportServer> logger)
|
||||
{
|
||||
_options = options.Value;
|
||||
_logger = logger;
|
||||
_nodeId = _options.NodeId ?? Guid.NewGuid().ToString("N")[..8];
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
ObjectDisposedException.ThrowIf(_disposed, this);
|
||||
|
||||
var factory = new ConnectionFactory
|
||||
{
|
||||
HostName = _options.HostName,
|
||||
Port = _options.Port,
|
||||
VirtualHost = _options.VirtualHost,
|
||||
UserName = _options.UserName,
|
||||
Password = _options.Password,
|
||||
AutomaticRecoveryEnabled = _options.AutomaticRecoveryEnabled,
|
||||
NetworkRecoveryInterval = _options.NetworkRecoveryInterval
|
||||
};
|
||||
|
||||
if (_options.UseSsl)
|
||||
{
|
||||
factory.Ssl = new SslOption
|
||||
{
|
||||
Enabled = true,
|
||||
ServerName = _options.HostName,
|
||||
CertPath = _options.SslCertPath
|
||||
};
|
||||
}
|
||||
|
||||
_connection = await factory.CreateConnectionAsync(cancellationToken);
|
||||
_channel = await _connection.CreateChannelAsync(cancellationToken: cancellationToken);
|
||||
|
||||
// Set QoS (prefetch count)
|
||||
await _channel.BasicQosAsync(
|
||||
prefetchSize: 0,
|
||||
prefetchCount: _options.PrefetchCount,
|
||||
global: false,
|
||||
cancellationToken: cancellationToken);
|
||||
|
||||
// Declare exchanges
|
||||
await _channel.ExchangeDeclareAsync(
|
||||
exchange: _options.RequestExchange,
|
||||
type: ExchangeType.Direct,
|
||||
durable: true,
|
||||
autoDelete: false,
|
||||
cancellationToken: cancellationToken);
|
||||
|
||||
await _channel.ExchangeDeclareAsync(
|
||||
exchange: _options.ResponseExchange,
|
||||
type: ExchangeType.Topic,
|
||||
durable: true,
|
||||
autoDelete: false,
|
||||
cancellationToken: cancellationToken);
|
||||
|
||||
// Declare and bind request queue
|
||||
_requestQueueName = $"{_options.QueuePrefix}.gw.{_nodeId}.in";
|
||||
await _channel.QueueDeclareAsync(
|
||||
queue: _requestQueueName,
|
||||
durable: _options.DurableQueues,
|
||||
exclusive: false,
|
||||
autoDelete: _options.AutoDeleteQueues,
|
||||
cancellationToken: cancellationToken);
|
||||
|
||||
await _channel.QueueBindAsync(
|
||||
queue: _requestQueueName,
|
||||
exchange: _options.RequestExchange,
|
||||
routingKey: _nodeId,
|
||||
cancellationToken: cancellationToken);
|
||||
|
||||
// Start consuming
|
||||
var consumer = new AsyncEventingBasicConsumer(_channel);
|
||||
consumer.ReceivedAsync += OnMessageReceivedAsync;
|
||||
|
||||
await _channel.BasicConsumeAsync(
|
||||
queue: _requestQueueName,
|
||||
autoAck: true, // At-most-once delivery
|
||||
consumer: consumer,
|
||||
cancellationToken: cancellationToken);
|
||||
|
||||
_logger.LogInformation(
|
||||
"RabbitMQ transport server started, consuming from {Queue}",
|
||||
_requestQueueName);
|
||||
}
|
||||
|
||||
private async Task OnMessageReceivedAsync(object sender, BasicDeliverEventArgs e)
|
||||
{
|
||||
try
|
||||
{
|
||||
var frame = RabbitMqFrameProtocol.ParseFrame(e.Body, e.BasicProperties);
|
||||
var connectionId = RabbitMqFrameProtocol.ExtractConnectionId(e.BasicProperties);
|
||||
var replyTo = e.BasicProperties.ReplyTo ?? string.Empty;
|
||||
|
||||
// Handle HELLO specially to register connection
|
||||
if (frame.Type == FrameType.Hello && !_connections.ContainsKey(connectionId))
|
||||
{
|
||||
var state = new ConnectionState
|
||||
{
|
||||
ConnectionId = connectionId,
|
||||
Instance = new InstanceDescriptor
|
||||
{
|
||||
InstanceId = connectionId,
|
||||
ServiceName = "unknown",
|
||||
Version = "1.0.0",
|
||||
Region = "default"
|
||||
},
|
||||
Status = InstanceHealthStatus.Healthy,
|
||||
LastHeartbeatUtc = DateTime.UtcNow,
|
||||
TransportType = TransportType.RabbitMq
|
||||
};
|
||||
|
||||
_connections[connectionId] = (replyTo, state);
|
||||
_logger.LogInformation(
|
||||
"RabbitMQ connection established: {ConnectionId} with replyTo {ReplyTo}",
|
||||
connectionId,
|
||||
replyTo);
|
||||
OnConnection?.Invoke(connectionId, state);
|
||||
}
|
||||
|
||||
// Update heartbeat timestamp on HEARTBEAT frames
|
||||
if (frame.Type == FrameType.Heartbeat &&
|
||||
_connections.TryGetValue(connectionId, out var conn))
|
||||
{
|
||||
conn.State.LastHeartbeatUtc = DateTime.UtcNow;
|
||||
}
|
||||
|
||||
OnFrame?.Invoke(connectionId, frame);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "Error processing RabbitMQ message");
|
||||
}
|
||||
|
||||
await Task.CompletedTask;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Sends a frame to a connection.
|
||||
/// </summary>
|
||||
/// <param name="connectionId">The connection ID.</param>
|
||||
/// <param name="frame">The frame to send.</param>
|
||||
/// <param name="cancellationToken">Cancellation token.</param>
|
||||
public async Task SendFrameAsync(
|
||||
string connectionId,
|
||||
Frame frame,
|
||||
CancellationToken cancellationToken = default)
|
||||
{
|
||||
ObjectDisposedException.ThrowIf(_disposed, this);
|
||||
|
||||
if (!_connections.TryGetValue(connectionId, out var conn))
|
||||
{
|
||||
throw new InvalidOperationException($"Connection {connectionId} not found");
|
||||
}
|
||||
|
||||
var properties = RabbitMqFrameProtocol.CreateProperties(frame, null, _options.DefaultTimeout);
|
||||
|
||||
// Send to response exchange with instance ID as routing key
|
||||
var routingKey = conn.ReplyTo.Split('.')[^1]; // Extract instance ID from queue name
|
||||
|
||||
await _channel!.BasicPublishAsync(
|
||||
exchange: _options.ResponseExchange,
|
||||
routingKey: routingKey,
|
||||
mandatory: false,
|
||||
basicProperties: properties,
|
||||
body: frame.Payload,
|
||||
cancellationToken: cancellationToken);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Gets the connection state by ID.
|
||||
/// </summary>
|
||||
/// <param name="connectionId">The connection ID.</param>
|
||||
/// <returns>The connection state, or null if not found.</returns>
|
||||
public ConnectionState? GetConnectionState(string connectionId)
|
||||
{
|
||||
return _connections.TryGetValue(connectionId, out var conn) ? conn.State : null;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Gets all active connections.
|
||||
/// </summary>
|
||||
public IEnumerable<ConnectionState> GetConnections() =>
|
||||
_connections.Values.Select(c => c.State);
|
||||
|
||||
/// <summary>
|
||||
/// Gets the number of active connections.
|
||||
/// </summary>
|
||||
public int ConnectionCount => _connections.Count;
|
||||
|
||||
/// <summary>
|
||||
/// Removes a connection.
|
||||
/// </summary>
|
||||
/// <param name="connectionId">The connection ID.</param>
|
||||
public void RemoveConnection(string connectionId)
|
||||
{
|
||||
if (_connections.TryRemove(connectionId, out _))
|
||||
{
|
||||
_logger.LogInformation("RabbitMQ connection removed: {ConnectionId}", connectionId);
|
||||
OnDisconnection?.Invoke(connectionId);
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task StopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
_logger.LogInformation("Stopping RabbitMQ transport server");
|
||||
|
||||
if (_channel is not null)
|
||||
{
|
||||
await _channel.CloseAsync(cancellationToken);
|
||||
}
|
||||
|
||||
if (_connection is not null)
|
||||
{
|
||||
await _connection.CloseAsync(cancellationToken);
|
||||
}
|
||||
|
||||
_connections.Clear();
|
||||
|
||||
_logger.LogInformation("RabbitMQ transport server stopped");
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async ValueTask DisposeAsync()
|
||||
{
|
||||
if (_disposed) return;
|
||||
_disposed = true;
|
||||
|
||||
await StopAsync(CancellationToken.None);
|
||||
|
||||
if (_channel is not null)
|
||||
{
|
||||
await _channel.DisposeAsync();
|
||||
}
|
||||
|
||||
if (_connection is not null)
|
||||
{
|
||||
await _connection.DisposeAsync();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,53 @@
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using StellaOps.Router.Common.Abstractions;
|
||||
|
||||
namespace StellaOps.Router.Transport.RabbitMq;
|
||||
|
||||
/// <summary>
|
||||
/// Extension methods for registering RabbitMQ transport services.
|
||||
/// </summary>
|
||||
public static class ServiceCollectionExtensions
|
||||
{
|
||||
/// <summary>
|
||||
/// Adds RabbitMQ transport server services to the service collection.
|
||||
/// </summary>
|
||||
/// <param name="services">The service collection.</param>
|
||||
/// <param name="configure">Optional configuration action.</param>
|
||||
/// <returns>The service collection.</returns>
|
||||
public static IServiceCollection AddRabbitMqTransportServer(
|
||||
this IServiceCollection services,
|
||||
Action<RabbitMqTransportOptions>? configure = null)
|
||||
{
|
||||
if (configure is not null)
|
||||
{
|
||||
services.Configure(configure);
|
||||
}
|
||||
|
||||
services.AddSingleton<RabbitMqTransportServer>();
|
||||
services.AddSingleton<ITransportServer>(sp => sp.GetRequiredService<RabbitMqTransportServer>());
|
||||
|
||||
return services;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Adds RabbitMQ transport client services to the service collection.
|
||||
/// </summary>
|
||||
/// <param name="services">The service collection.</param>
|
||||
/// <param name="configure">Optional configuration action.</param>
|
||||
/// <returns>The service collection.</returns>
|
||||
public static IServiceCollection AddRabbitMqTransportClient(
|
||||
this IServiceCollection services,
|
||||
Action<RabbitMqTransportOptions>? configure = null)
|
||||
{
|
||||
if (configure is not null)
|
||||
{
|
||||
services.Configure(configure);
|
||||
}
|
||||
|
||||
services.AddSingleton<RabbitMqTransportClient>();
|
||||
services.AddSingleton<ITransportClient>(sp => sp.GetRequiredService<RabbitMqTransportClient>());
|
||||
services.AddSingleton<IMicroserviceTransport>(sp => sp.GetRequiredService<RabbitMqTransportClient>());
|
||||
|
||||
return services;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
<Project Sdk="Microsoft.NET.Sdk">
|
||||
|
||||
<PropertyGroup>
|
||||
<TargetFramework>net10.0</TargetFramework>
|
||||
<ImplicitUsings>enable</ImplicitUsings>
|
||||
<Nullable>enable</Nullable>
|
||||
<LangVersion>preview</LangVersion>
|
||||
<TreatWarningsAsErrors>true</TreatWarningsAsErrors>
|
||||
<RootNamespace>StellaOps.Router.Transport.RabbitMq</RootNamespace>
|
||||
</PropertyGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<ProjectReference Include="..\StellaOps.Router.Common\StellaOps.Router.Common.csproj" />
|
||||
</ItemGroup>
|
||||
|
||||
<ItemGroup>
|
||||
<PackageReference Include="Microsoft.Extensions.DependencyInjection.Abstractions" Version="10.0.0-rc.2.25502.107" />
|
||||
<PackageReference Include="Microsoft.Extensions.Logging.Abstractions" Version="10.0.0-rc.2.25502.107" />
|
||||
<PackageReference Include="Microsoft.Extensions.Options" Version="10.0.0-rc.2.25502.107" />
|
||||
<PackageReference Include="RabbitMQ.Client" Version="7.0.0" />
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
Reference in New Issue
Block a user