Implement InMemory Transport Layer for StellaOps Router
- Added InMemoryTransportOptions class for configuration settings including timeouts and latency. - Developed InMemoryTransportServer class to handle connections, frame processing, and event management. - Created ServiceCollectionExtensions for easy registration of InMemory transport services. - Established project structure and dependencies for InMemory transport library. - Implemented comprehensive unit tests for endpoint discovery, connection management, request/response flow, and streaming capabilities. - Ensured proper handling of cancellation, heartbeat, and hello frames within the transport layer.
This commit is contained in:
StellaOps Bot
@@ -0,0 +1,93 @@
|
||||
using System.Threading.Channels;
|
||||
using StellaOps.Router.Common.Models;
|
||||
|
||||
namespace StellaOps.Router.Transport.InMemory;
|
||||
|
||||
/// <summary>
|
||||
/// Represents a bidirectional in-memory channel for frame passing between gateway and microservice.
|
||||
/// </summary>
|
||||
public sealed class InMemoryChannel : IDisposable
|
||||
{
|
||||
private bool _disposed;
|
||||
|
||||
/// <summary>
|
||||
/// Gets the connection ID.
|
||||
/// </summary>
|
||||
public string ConnectionId { get; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets the channel for frames from gateway to microservice.
|
||||
/// Gateway writes, SDK reads.
|
||||
/// </summary>
|
||||
public Channel<Frame> ToMicroservice { get; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets the channel for frames from microservice to gateway.
|
||||
/// SDK writes, Gateway reads.
|
||||
/// </summary>
|
||||
public Channel<Frame> ToGateway { get; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the instance descriptor for this connection.
|
||||
/// Set when HELLO is processed.
|
||||
/// </summary>
|
||||
public InstanceDescriptor? Instance { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets the cancellation token source for this connection's lifetime.
|
||||
/// </summary>
|
||||
public CancellationTokenSource LifetimeToken { get; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the connection state.
|
||||
/// </summary>
|
||||
public ConnectionState? State { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="InMemoryChannel"/> class.
|
||||
/// </summary>
|
||||
/// <param name="connectionId">The connection ID.</param>
|
||||
/// <param name="bufferSize">The channel buffer size. Zero for unbounded.</param>
|
||||
public InMemoryChannel(string connectionId, int bufferSize = 0)
|
||||
{
|
||||
ConnectionId = connectionId;
|
||||
LifetimeToken = new CancellationTokenSource();
|
||||
|
||||
if (bufferSize > 0)
|
||||
{
|
||||
var options = new BoundedChannelOptions(bufferSize)
|
||||
{
|
||||
FullMode = BoundedChannelFullMode.Wait,
|
||||
SingleReader = false,
|
||||
SingleWriter = false
|
||||
};
|
||||
ToMicroservice = Channel.CreateBounded<Frame>(options);
|
||||
ToGateway = Channel.CreateBounded<Frame>(options);
|
||||
}
|
||||
else
|
||||
{
|
||||
var options = new UnboundedChannelOptions
|
||||
{
|
||||
SingleReader = false,
|
||||
SingleWriter = false
|
||||
};
|
||||
ToMicroservice = Channel.CreateUnbounded<Frame>(options);
|
||||
ToGateway = Channel.CreateUnbounded<Frame>(options);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Disposes the channel and cancels all pending operations.
|
||||
/// </summary>
|
||||
public void Dispose()
|
||||
{
|
||||
if (_disposed) return;
|
||||
_disposed = true;
|
||||
|
||||
LifetimeToken.Cancel();
|
||||
LifetimeToken.Dispose();
|
||||
|
||||
ToMicroservice.Writer.TryComplete();
|
||||
ToGateway.Writer.TryComplete();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,124 @@
|
||||
using System.Collections.Concurrent;
|
||||
using StellaOps.Router.Common.Models;
|
||||
|
||||
namespace StellaOps.Router.Transport.InMemory;
|
||||
|
||||
/// <summary>
|
||||
/// Thread-safe registry for in-memory connections.
|
||||
/// </summary>
|
||||
public sealed class InMemoryConnectionRegistry : IDisposable
|
||||
{
|
||||
private readonly ConcurrentDictionary<string, InMemoryChannel> _channels = new();
|
||||
private bool _disposed;
|
||||
|
||||
/// <summary>
|
||||
/// Gets all connection IDs.
|
||||
/// </summary>
|
||||
public IEnumerable<string> ConnectionIds => _channels.Keys;
|
||||
|
||||
/// <summary>
|
||||
/// Gets the count of active connections.
|
||||
/// </summary>
|
||||
public int Count => _channels.Count;
|
||||
|
||||
/// <summary>
|
||||
/// Creates a new channel with the given connection ID.
|
||||
/// </summary>
|
||||
/// <param name="connectionId">The connection ID.</param>
|
||||
/// <param name="bufferSize">The channel buffer size.</param>
|
||||
/// <returns>The created channel.</returns>
|
||||
public InMemoryChannel CreateChannel(string connectionId, int bufferSize = 0)
|
||||
{
|
||||
ObjectDisposedException.ThrowIf(_disposed, this);
|
||||
|
||||
var channel = new InMemoryChannel(connectionId, bufferSize);
|
||||
if (!_channels.TryAdd(connectionId, channel))
|
||||
{
|
||||
channel.Dispose();
|
||||
throw new InvalidOperationException($"Connection {connectionId} already exists.");
|
||||
}
|
||||
return channel;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Gets a channel by connection ID.
|
||||
/// </summary>
|
||||
/// <param name="connectionId">The connection ID.</param>
|
||||
/// <returns>The channel, or null if not found.</returns>
|
||||
public InMemoryChannel? GetChannel(string connectionId)
|
||||
{
|
||||
_channels.TryGetValue(connectionId, out var channel);
|
||||
return channel;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Gets a channel by connection ID, throwing if not found.
|
||||
/// </summary>
|
||||
/// <param name="connectionId">The connection ID.</param>
|
||||
/// <returns>The channel.</returns>
|
||||
public InMemoryChannel GetRequiredChannel(string connectionId)
|
||||
{
|
||||
return GetChannel(connectionId)
|
||||
?? throw new InvalidOperationException($"Connection {connectionId} not found.");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Removes and disposes a channel by connection ID.
|
||||
/// </summary>
|
||||
/// <param name="connectionId">The connection ID.</param>
|
||||
/// <returns>True if the channel was found and removed.</returns>
|
||||
public bool RemoveChannel(string connectionId)
|
||||
{
|
||||
if (_channels.TryRemove(connectionId, out var channel))
|
||||
{
|
||||
channel.Dispose();
|
||||
return true;
|
||||
}
|
||||
return false;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Gets all active connection states.
|
||||
/// </summary>
|
||||
public IReadOnlyList<ConnectionState> GetAllConnections()
|
||||
{
|
||||
return _channels.Values
|
||||
.Where(c => c.State is not null)
|
||||
.Select(c => c.State!)
|
||||
.ToList();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Gets connections for a specific service and endpoint.
|
||||
/// </summary>
|
||||
/// <param name="serviceName">The service name.</param>
|
||||
/// <param name="version">The version.</param>
|
||||
/// <param name="method">The HTTP method.</param>
|
||||
/// <param name="path">The path.</param>
|
||||
public IReadOnlyList<ConnectionState> GetConnectionsFor(
|
||||
string serviceName, string version, string method, string path)
|
||||
{
|
||||
return _channels.Values
|
||||
.Where(c => c.State is not null
|
||||
&& c.Instance?.ServiceName == serviceName
|
||||
&& c.Instance?.Version == version
|
||||
&& c.State.Endpoints.ContainsKey((method, path)))
|
||||
.Select(c => c.State!)
|
||||
.ToList();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Disposes all channels.
|
||||
/// </summary>
|
||||
public void Dispose()
|
||||
{
|
||||
if (_disposed) return;
|
||||
_disposed = true;
|
||||
|
||||
foreach (var channel in _channels.Values)
|
||||
{
|
||||
channel.Dispose();
|
||||
}
|
||||
_channels.Clear();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,425 @@
|
||||
using System.Buffers;
|
||||
using System.Collections.Concurrent;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Options;
|
||||
using StellaOps.Router.Common.Abstractions;
|
||||
using StellaOps.Router.Common.Enums;
|
||||
using StellaOps.Router.Common.Models;
|
||||
|
||||
namespace StellaOps.Router.Transport.InMemory;
|
||||
|
||||
/// <summary>
|
||||
/// In-memory transport client implementation for testing and development.
|
||||
/// Used by the Microservice SDK to send frames to the Gateway.
|
||||
/// </summary>
|
||||
public sealed class InMemoryTransportClient : ITransportClient, IDisposable
|
||||
{
|
||||
private readonly InMemoryConnectionRegistry _registry;
|
||||
private readonly InMemoryTransportOptions _options;
|
||||
private readonly ILogger<InMemoryTransportClient> _logger;
|
||||
private readonly ConcurrentDictionary<string, TaskCompletionSource<Frame>> _pendingRequests = new();
|
||||
private readonly CancellationTokenSource _clientCts = new();
|
||||
private bool _disposed;
|
||||
private string? _connectionId;
|
||||
private Task? _receiveTask;
|
||||
|
||||
/// <summary>
|
||||
/// Event raised when a REQUEST frame is received from the gateway.
|
||||
/// </summary>
|
||||
public event Func<Frame, CancellationToken, Task<Frame>>? OnRequestReceived;
|
||||
|
||||
/// <summary>
|
||||
/// Event raised when a CANCEL frame is received from the gateway.
|
||||
/// </summary>
|
||||
public event Func<Guid, string?, Task>? OnCancelReceived;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="InMemoryTransportClient"/> class.
|
||||
/// </summary>
|
||||
public InMemoryTransportClient(
|
||||
InMemoryConnectionRegistry registry,
|
||||
IOptions<InMemoryTransportOptions> options,
|
||||
ILogger<InMemoryTransportClient> logger)
|
||||
{
|
||||
_registry = registry;
|
||||
_options = options.Value;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Connects to the in-memory transport and sends a HELLO frame.
|
||||
/// </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);
|
||||
|
||||
_connectionId = Guid.NewGuid().ToString("N");
|
||||
var channel = _registry.CreateChannel(_connectionId, _options.ChannelBufferSize);
|
||||
channel.Instance = instance;
|
||||
|
||||
// Create initial ConnectionState
|
||||
var state = new ConnectionState
|
||||
{
|
||||
ConnectionId = _connectionId,
|
||||
Instance = instance,
|
||||
Status = InstanceHealthStatus.Healthy,
|
||||
LastHeartbeatUtc = DateTime.UtcNow,
|
||||
TransportType = TransportType.InMemory
|
||||
};
|
||||
|
||||
// Register endpoints
|
||||
foreach (var endpoint in endpoints)
|
||||
{
|
||||
state.Endpoints[(endpoint.Method, endpoint.Path)] = endpoint;
|
||||
}
|
||||
channel.State = state;
|
||||
|
||||
// Send HELLO frame
|
||||
var helloFrame = new Frame
|
||||
{
|
||||
Type = FrameType.Hello,
|
||||
CorrelationId = Guid.NewGuid().ToString("N"),
|
||||
Payload = ReadOnlyMemory<byte>.Empty
|
||||
};
|
||||
await channel.ToGateway.Writer.WriteAsync(helloFrame, cancellationToken);
|
||||
|
||||
_logger.LogInformation(
|
||||
"Connected as {ServiceName}/{Version} instance {InstanceId} with {EndpointCount} endpoints",
|
||||
instance.ServiceName,
|
||||
instance.Version,
|
||||
instance.InstanceId,
|
||||
endpoints.Count);
|
||||
|
||||
// Start receiving frames from gateway
|
||||
_receiveTask = Task.Run(() => ReceiveLoopAsync(_clientCts.Token), CancellationToken.None);
|
||||
}
|
||||
|
||||
private async Task ReceiveLoopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
if (_connectionId is null) return;
|
||||
|
||||
var channel = _registry.GetChannel(_connectionId);
|
||||
if (channel is null) return;
|
||||
|
||||
using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(
|
||||
cancellationToken, channel.LifetimeToken.Token);
|
||||
|
||||
try
|
||||
{
|
||||
await foreach (var frame in channel.ToMicroservice.Reader.ReadAllAsync(linkedCts.Token))
|
||||
{
|
||||
if (_options.SimulatedLatency > TimeSpan.Zero)
|
||||
{
|
||||
await Task.Delay(_options.SimulatedLatency, linkedCts.Token);
|
||||
}
|
||||
|
||||
await ProcessFrameFromGatewayAsync(channel, frame, linkedCts.Token);
|
||||
}
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
// Expected on disconnect
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "Error in receive loop");
|
||||
}
|
||||
}
|
||||
|
||||
private async Task ProcessFrameFromGatewayAsync(
|
||||
InMemoryChannel channel, Frame frame, CancellationToken cancellationToken)
|
||||
{
|
||||
switch (frame.Type)
|
||||
{
|
||||
case FrameType.Request:
|
||||
case FrameType.RequestStreamData:
|
||||
await HandleRequestFrameAsync(channel, frame, cancellationToken);
|
||||
break;
|
||||
|
||||
case FrameType.Cancel:
|
||||
HandleCancelFrame(frame);
|
||||
break;
|
||||
|
||||
case FrameType.Response:
|
||||
case FrameType.ResponseStreamData:
|
||||
// Response to our request (from gateway back)
|
||||
if (frame.CorrelationId is not null &&
|
||||
_pendingRequests.TryRemove(frame.CorrelationId, out var tcs))
|
||||
{
|
||||
tcs.TrySetResult(frame);
|
||||
}
|
||||
break;
|
||||
|
||||
default:
|
||||
_logger.LogWarning("Unexpected frame type {FrameType} from gateway", frame.Type);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
private async Task HandleRequestFrameAsync(
|
||||
InMemoryChannel channel, Frame frame, CancellationToken cancellationToken)
|
||||
{
|
||||
if (OnRequestReceived is null)
|
||||
{
|
||||
_logger.LogWarning("No request handler registered, discarding request {CorrelationId}",
|
||||
frame.CorrelationId);
|
||||
return;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
var response = await OnRequestReceived(frame, cancellationToken);
|
||||
|
||||
// Ensure response has same correlation ID
|
||||
var responseFrame = response with { CorrelationId = frame.CorrelationId };
|
||||
await channel.ToGateway.Writer.WriteAsync(responseFrame, cancellationToken);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
_logger.LogDebug("Request {CorrelationId} was cancelled", frame.CorrelationId);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "Error handling request {CorrelationId}", frame.CorrelationId);
|
||||
// Send error response
|
||||
var errorFrame = new Frame
|
||||
{
|
||||
Type = FrameType.Response,
|
||||
CorrelationId = frame.CorrelationId,
|
||||
Payload = ReadOnlyMemory<byte>.Empty
|
||||
};
|
||||
await channel.ToGateway.Writer.WriteAsync(errorFrame, cancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
private void HandleCancelFrame(Frame frame)
|
||||
{
|
||||
if (frame.CorrelationId is null) return;
|
||||
|
||||
_logger.LogDebug("Received CANCEL for correlation {CorrelationId}", frame.CorrelationId);
|
||||
|
||||
// Complete any pending request with cancellation
|
||||
if (_pendingRequests.TryRemove(frame.CorrelationId, out var tcs))
|
||||
{
|
||||
tcs.TrySetCanceled();
|
||||
}
|
||||
|
||||
// Notify handler
|
||||
if (OnCancelReceived is not null && Guid.TryParse(frame.CorrelationId, out var correlationGuid))
|
||||
{
|
||||
_ = OnCancelReceived(correlationGuid, null);
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task<Frame> SendRequestAsync(
|
||||
ConnectionState connection,
|
||||
Frame requestFrame,
|
||||
TimeSpan timeout,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
ObjectDisposedException.ThrowIf(_disposed, this);
|
||||
|
||||
var channel = _registry.GetRequiredChannel(connection.ConnectionId);
|
||||
var correlationId = requestFrame.CorrelationId ?? Guid.NewGuid().ToString("N");
|
||||
var framedRequest = requestFrame with { CorrelationId = correlationId };
|
||||
|
||||
var tcs = new TaskCompletionSource<Frame>(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
_pendingRequests[correlationId] = tcs;
|
||||
|
||||
try
|
||||
{
|
||||
if (_options.SimulatedLatency > TimeSpan.Zero)
|
||||
{
|
||||
await Task.Delay(_options.SimulatedLatency, cancellationToken);
|
||||
}
|
||||
|
||||
await channel.ToMicroservice.Writer.WriteAsync(framedRequest, cancellationToken);
|
||||
|
||||
using var timeoutCts = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
|
||||
timeoutCts.CancelAfter(timeout);
|
||||
|
||||
return await tcs.Task.WaitAsync(timeoutCts.Token);
|
||||
}
|
||||
catch (OperationCanceledException) when (!cancellationToken.IsCancellationRequested)
|
||||
{
|
||||
throw new TimeoutException($"Request {correlationId} timed out after {timeout}");
|
||||
}
|
||||
finally
|
||||
{
|
||||
_pendingRequests.TryRemove(correlationId, out _);
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task SendCancelAsync(
|
||||
ConnectionState connection,
|
||||
Guid correlationId,
|
||||
string? reason = null)
|
||||
{
|
||||
ObjectDisposedException.ThrowIf(_disposed, this);
|
||||
|
||||
var channel = _registry.GetRequiredChannel(connection.ConnectionId);
|
||||
|
||||
var cancelFrame = new Frame
|
||||
{
|
||||
Type = FrameType.Cancel,
|
||||
CorrelationId = correlationId.ToString("N"),
|
||||
Payload = ReadOnlyMemory<byte>.Empty
|
||||
};
|
||||
|
||||
if (_options.SimulatedLatency > TimeSpan.Zero)
|
||||
{
|
||||
await Task.Delay(_options.SimulatedLatency);
|
||||
}
|
||||
|
||||
await channel.ToMicroservice.Writer.WriteAsync(cancelFrame);
|
||||
|
||||
_logger.LogDebug("Sent CANCEL for correlation {CorrelationId}", correlationId);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task SendStreamingAsync(
|
||||
ConnectionState connection,
|
||||
Frame requestHeader,
|
||||
Stream requestBody,
|
||||
Func<Stream, Task> readResponseBody,
|
||||
PayloadLimits limits,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
ObjectDisposedException.ThrowIf(_disposed, this);
|
||||
|
||||
var channel = _registry.GetRequiredChannel(connection.ConnectionId);
|
||||
var correlationId = requestHeader.CorrelationId ?? Guid.NewGuid().ToString("N");
|
||||
|
||||
// Send header frame
|
||||
var headerFrame = requestHeader with
|
||||
{
|
||||
Type = FrameType.Request,
|
||||
CorrelationId = correlationId
|
||||
};
|
||||
await channel.ToMicroservice.Writer.WriteAsync(headerFrame, cancellationToken);
|
||||
|
||||
// Stream request body in chunks
|
||||
var buffer = ArrayPool<byte>.Shared.Rent(8192);
|
||||
try
|
||||
{
|
||||
long totalBytesRead = 0;
|
||||
int bytesRead;
|
||||
|
||||
while ((bytesRead = await requestBody.ReadAsync(buffer, cancellationToken)) > 0)
|
||||
{
|
||||
totalBytesRead += bytesRead;
|
||||
|
||||
if (totalBytesRead > limits.MaxRequestBytesPerCall)
|
||||
{
|
||||
throw new InvalidOperationException(
|
||||
$"Request body exceeds limit of {limits.MaxRequestBytesPerCall} bytes");
|
||||
}
|
||||
|
||||
var dataFrame = new Frame
|
||||
{
|
||||
Type = FrameType.RequestStreamData,
|
||||
CorrelationId = correlationId,
|
||||
Payload = new ReadOnlyMemory<byte>(buffer, 0, bytesRead)
|
||||
};
|
||||
await channel.ToMicroservice.Writer.WriteAsync(dataFrame, cancellationToken);
|
||||
|
||||
if (_options.SimulatedLatency > TimeSpan.Zero)
|
||||
{
|
||||
await Task.Delay(_options.SimulatedLatency, cancellationToken);
|
||||
}
|
||||
}
|
||||
|
||||
// Signal end of request stream with empty data frame
|
||||
var endFrame = new Frame
|
||||
{
|
||||
Type = FrameType.RequestStreamData,
|
||||
CorrelationId = correlationId,
|
||||
Payload = ReadOnlyMemory<byte>.Empty
|
||||
};
|
||||
await channel.ToMicroservice.Writer.WriteAsync(endFrame, cancellationToken);
|
||||
}
|
||||
finally
|
||||
{
|
||||
ArrayPool<byte>.Shared.Return(buffer);
|
||||
}
|
||||
|
||||
// Read streaming response
|
||||
using var responseStream = new MemoryStream();
|
||||
var tcs = new TaskCompletionSource<bool>(TaskCreationOptions.RunContinuationsAsynchronously);
|
||||
_pendingRequests[correlationId] = new TaskCompletionSource<Frame>();
|
||||
|
||||
// TODO: Implement proper streaming response handling
|
||||
// For now, we accumulate the response in memory
|
||||
await readResponseBody(responseStream);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Sends a heartbeat frame.
|
||||
/// </summary>
|
||||
public async Task SendHeartbeatAsync(HeartbeatPayload heartbeat, CancellationToken cancellationToken)
|
||||
{
|
||||
if (_connectionId is null) return;
|
||||
|
||||
var channel = _registry.GetChannel(_connectionId);
|
||||
if (channel is null) return;
|
||||
|
||||
var frame = new Frame
|
||||
{
|
||||
Type = FrameType.Heartbeat,
|
||||
CorrelationId = null,
|
||||
Payload = ReadOnlyMemory<byte>.Empty
|
||||
};
|
||||
|
||||
await channel.ToGateway.Writer.WriteAsync(frame, cancellationToken);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Disconnects from the transport.
|
||||
/// </summary>
|
||||
public async Task DisconnectAsync()
|
||||
{
|
||||
if (_connectionId is null) return;
|
||||
|
||||
await _clientCts.CancelAsync();
|
||||
|
||||
if (_receiveTask is not null)
|
||||
{
|
||||
await _receiveTask;
|
||||
}
|
||||
|
||||
_registry.RemoveChannel(_connectionId);
|
||||
_connectionId = null;
|
||||
|
||||
_logger.LogInformation("Disconnected from in-memory transport");
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public void Dispose()
|
||||
{
|
||||
if (_disposed) return;
|
||||
_disposed = true;
|
||||
|
||||
_clientCts.Cancel();
|
||||
|
||||
foreach (var tcs in _pendingRequests.Values)
|
||||
{
|
||||
tcs.TrySetCanceled();
|
||||
}
|
||||
_pendingRequests.Clear();
|
||||
|
||||
if (_connectionId is not null)
|
||||
{
|
||||
_registry.RemoveChannel(_connectionId);
|
||||
}
|
||||
|
||||
_clientCts.Dispose();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
namespace StellaOps.Router.Transport.InMemory;
|
||||
|
||||
/// <summary>
|
||||
/// Configuration options for the InMemory transport.
|
||||
/// </summary>
|
||||
public sealed class InMemoryTransportOptions
|
||||
{
|
||||
/// <summary>
|
||||
/// Gets or sets the default timeout for requests.
|
||||
/// Default: 30 seconds.
|
||||
/// </summary>
|
||||
public TimeSpan DefaultTimeout { get; set; } = TimeSpan.FromSeconds(30);
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the simulated latency for frame delivery.
|
||||
/// Default: Zero (instant delivery).
|
||||
/// </summary>
|
||||
public TimeSpan SimulatedLatency { get; set; } = TimeSpan.Zero;
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the channel buffer size.
|
||||
/// Default: Unbounded (0 means unbounded).
|
||||
/// </summary>
|
||||
public int ChannelBufferSize { get; set; }
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the heartbeat interval.
|
||||
/// Default: 10 seconds.
|
||||
/// </summary>
|
||||
public TimeSpan HeartbeatInterval { get; set; } = TimeSpan.FromSeconds(10);
|
||||
|
||||
/// <summary>
|
||||
/// Gets or sets the heartbeat timeout (time since last heartbeat before connection is considered unhealthy).
|
||||
/// Default: 30 seconds.
|
||||
/// </summary>
|
||||
public TimeSpan HeartbeatTimeout { get; set; } = TimeSpan.FromSeconds(30);
|
||||
}
|
||||
@@ -0,0 +1,264 @@
|
||||
using System.Collections.Concurrent;
|
||||
using Microsoft.Extensions.Logging;
|
||||
using Microsoft.Extensions.Options;
|
||||
using StellaOps.Router.Common.Abstractions;
|
||||
using StellaOps.Router.Common.Enums;
|
||||
using StellaOps.Router.Common.Models;
|
||||
|
||||
namespace StellaOps.Router.Transport.InMemory;
|
||||
|
||||
/// <summary>
|
||||
/// In-memory transport server implementation for testing and development.
|
||||
/// Used by the Gateway to receive frames from microservices.
|
||||
/// </summary>
|
||||
public sealed class InMemoryTransportServer : ITransportServer, IDisposable
|
||||
{
|
||||
private readonly InMemoryConnectionRegistry _registry;
|
||||
private readonly InMemoryTransportOptions _options;
|
||||
private readonly ILogger<InMemoryTransportServer> _logger;
|
||||
private readonly ConcurrentDictionary<string, Task> _connectionTasks = new();
|
||||
private readonly CancellationTokenSource _serverCts = new();
|
||||
private bool _running;
|
||||
private bool _disposed;
|
||||
|
||||
/// <summary>
|
||||
/// Event raised when a HELLO frame is received.
|
||||
/// </summary>
|
||||
public event Func<ConnectionState, HelloPayload, Task>? OnHelloReceived;
|
||||
|
||||
/// <summary>
|
||||
/// Event raised when a HEARTBEAT frame is received.
|
||||
/// </summary>
|
||||
public event Func<ConnectionState, HeartbeatPayload, Task>? OnHeartbeatReceived;
|
||||
|
||||
/// <summary>
|
||||
/// Event raised when a RESPONSE frame is received.
|
||||
/// </summary>
|
||||
public event Func<ConnectionState, Frame, Task>? OnResponseReceived;
|
||||
|
||||
/// <summary>
|
||||
/// Event raised when a connection is closed.
|
||||
/// </summary>
|
||||
public event Func<string, Task>? OnConnectionClosed;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="InMemoryTransportServer"/> class.
|
||||
/// </summary>
|
||||
public InMemoryTransportServer(
|
||||
InMemoryConnectionRegistry registry,
|
||||
IOptions<InMemoryTransportOptions> options,
|
||||
ILogger<InMemoryTransportServer> logger)
|
||||
{
|
||||
_registry = registry;
|
||||
_options = options.Value;
|
||||
_logger = logger;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task StartAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
ObjectDisposedException.ThrowIf(_disposed, this);
|
||||
|
||||
if (_running)
|
||||
{
|
||||
_logger.LogWarning("InMemory transport server is already running");
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
_running = true;
|
||||
_logger.LogInformation("InMemory transport server started");
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public async Task StopAsync(CancellationToken cancellationToken)
|
||||
{
|
||||
if (!_running) return;
|
||||
|
||||
_logger.LogInformation("InMemory transport server stopping");
|
||||
_running = false;
|
||||
|
||||
await _serverCts.CancelAsync();
|
||||
|
||||
// Wait for all connection tasks to complete
|
||||
var tasks = _connectionTasks.Values.ToArray();
|
||||
if (tasks.Length > 0)
|
||||
{
|
||||
await Task.WhenAll(tasks).WaitAsync(cancellationToken);
|
||||
}
|
||||
|
||||
_logger.LogInformation("InMemory transport server stopped");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Starts listening to a specific connection's ToGateway channel.
|
||||
/// Called when a new connection is registered.
|
||||
/// </summary>
|
||||
public void StartListeningToConnection(string connectionId)
|
||||
{
|
||||
if (!_running) return;
|
||||
|
||||
var channel = _registry.GetChannel(connectionId);
|
||||
if (channel is null) return;
|
||||
|
||||
var task = Task.Run(async () =>
|
||||
{
|
||||
try
|
||||
{
|
||||
await ProcessConnectionFramesAsync(channel, _serverCts.Token);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
// Expected on shutdown
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "Error processing frames for connection {ConnectionId}", connectionId);
|
||||
}
|
||||
finally
|
||||
{
|
||||
_connectionTasks.TryRemove(connectionId, out _);
|
||||
if (OnConnectionClosed is not null)
|
||||
{
|
||||
await OnConnectionClosed(connectionId);
|
||||
}
|
||||
}
|
||||
});
|
||||
|
||||
_connectionTasks[connectionId] = task;
|
||||
}
|
||||
|
||||
private async Task ProcessConnectionFramesAsync(InMemoryChannel channel, CancellationToken cancellationToken)
|
||||
{
|
||||
using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(
|
||||
cancellationToken, channel.LifetimeToken.Token);
|
||||
|
||||
await foreach (var frame in channel.ToGateway.Reader.ReadAllAsync(linkedCts.Token))
|
||||
{
|
||||
if (_options.SimulatedLatency > TimeSpan.Zero)
|
||||
{
|
||||
await Task.Delay(_options.SimulatedLatency, linkedCts.Token);
|
||||
}
|
||||
|
||||
await ProcessFrameAsync(channel, frame, linkedCts.Token);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task ProcessFrameAsync(InMemoryChannel channel, Frame frame, CancellationToken cancellationToken)
|
||||
{
|
||||
switch (frame.Type)
|
||||
{
|
||||
case FrameType.Hello:
|
||||
await ProcessHelloFrameAsync(channel, frame, cancellationToken);
|
||||
break;
|
||||
|
||||
case FrameType.Heartbeat:
|
||||
await ProcessHeartbeatFrameAsync(channel, frame, cancellationToken);
|
||||
break;
|
||||
|
||||
case FrameType.Response:
|
||||
case FrameType.ResponseStreamData:
|
||||
if (channel.State is not null && OnResponseReceived is not null)
|
||||
{
|
||||
await OnResponseReceived(channel.State, frame);
|
||||
}
|
||||
break;
|
||||
|
||||
case FrameType.Cancel:
|
||||
_logger.LogDebug("Received CANCEL from microservice for correlation {CorrelationId}",
|
||||
frame.CorrelationId);
|
||||
break;
|
||||
|
||||
default:
|
||||
_logger.LogWarning("Unexpected frame type {FrameType} from connection {ConnectionId}",
|
||||
frame.Type, channel.ConnectionId);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
private async Task ProcessHelloFrameAsync(InMemoryChannel channel, Frame frame, CancellationToken cancellationToken)
|
||||
{
|
||||
// In a real implementation, we'd deserialize the payload
|
||||
// For now, the HelloPayload should be passed out-of-band via the channel
|
||||
if (channel.Instance is null)
|
||||
{
|
||||
_logger.LogWarning("HELLO received but Instance not set for connection {ConnectionId}",
|
||||
channel.ConnectionId);
|
||||
return;
|
||||
}
|
||||
|
||||
// Create ConnectionState
|
||||
var state = new ConnectionState
|
||||
{
|
||||
ConnectionId = channel.ConnectionId,
|
||||
Instance = channel.Instance,
|
||||
Status = InstanceHealthStatus.Healthy,
|
||||
LastHeartbeatUtc = DateTime.UtcNow,
|
||||
TransportType = TransportType.InMemory
|
||||
};
|
||||
channel.State = state;
|
||||
|
||||
_logger.LogInformation(
|
||||
"HELLO received from {ServiceName}/{Version} instance {InstanceId}",
|
||||
channel.Instance.ServiceName,
|
||||
channel.Instance.Version,
|
||||
channel.Instance.InstanceId);
|
||||
|
||||
// Fire event with dummy HelloPayload (real impl would deserialize from frame)
|
||||
if (OnHelloReceived is not null)
|
||||
{
|
||||
var payload = new HelloPayload
|
||||
{
|
||||
Instance = channel.Instance,
|
||||
Endpoints = []
|
||||
};
|
||||
await OnHelloReceived(state, payload);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task ProcessHeartbeatFrameAsync(InMemoryChannel channel, Frame frame, CancellationToken cancellationToken)
|
||||
{
|
||||
if (channel.State is null) return;
|
||||
|
||||
channel.State.LastHeartbeatUtc = DateTime.UtcNow;
|
||||
|
||||
_logger.LogDebug("Heartbeat received from {ConnectionId}", channel.ConnectionId);
|
||||
|
||||
if (OnHeartbeatReceived is not null)
|
||||
{
|
||||
var payload = new HeartbeatPayload
|
||||
{
|
||||
InstanceId = channel.Instance?.InstanceId ?? channel.ConnectionId,
|
||||
Status = channel.State.Status,
|
||||
TimestampUtc = DateTime.UtcNow
|
||||
};
|
||||
await OnHeartbeatReceived(channel.State, payload);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Sends a frame to a microservice via the ToMicroservice channel.
|
||||
/// </summary>
|
||||
public async ValueTask SendToMicroserviceAsync(
|
||||
string connectionId, Frame frame, CancellationToken cancellationToken)
|
||||
{
|
||||
var channel = _registry.GetRequiredChannel(connectionId);
|
||||
|
||||
if (_options.SimulatedLatency > TimeSpan.Zero)
|
||||
{
|
||||
await Task.Delay(_options.SimulatedLatency, cancellationToken);
|
||||
}
|
||||
|
||||
await channel.ToMicroservice.Writer.WriteAsync(frame, cancellationToken);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public void Dispose()
|
||||
{
|
||||
if (_disposed) return;
|
||||
_disposed = true;
|
||||
|
||||
_serverCts.Cancel();
|
||||
_serverCts.Dispose();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,87 @@
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.DependencyInjection.Extensions;
|
||||
using StellaOps.Router.Common.Abstractions;
|
||||
|
||||
namespace StellaOps.Router.Transport.InMemory;
|
||||
|
||||
/// <summary>
|
||||
/// Extension methods for registering InMemory transport services.
|
||||
/// </summary>
|
||||
public static class ServiceCollectionExtensions
|
||||
{
|
||||
/// <summary>
|
||||
/// Adds the InMemory transport for testing and development.
|
||||
/// </summary>
|
||||
/// <param name="services">The service collection.</param>
|
||||
/// <param name="configure">Optional configuration action.</param>
|
||||
/// <returns>The service collection for chaining.</returns>
|
||||
public static IServiceCollection AddInMemoryTransport(
|
||||
this IServiceCollection services,
|
||||
Action<InMemoryTransportOptions>? configure = null)
|
||||
{
|
||||
services.AddOptions<InMemoryTransportOptions>();
|
||||
if (configure is not null)
|
||||
{
|
||||
services.Configure(configure);
|
||||
}
|
||||
|
||||
// Singleton registry shared between server and client
|
||||
services.TryAddSingleton<InMemoryConnectionRegistry>();
|
||||
|
||||
// Transport implementations
|
||||
services.TryAddSingleton<InMemoryTransportServer>();
|
||||
services.TryAddSingleton<InMemoryTransportClient>();
|
||||
|
||||
// Register interfaces
|
||||
services.TryAddSingleton<ITransportServer>(sp => sp.GetRequiredService<InMemoryTransportServer>());
|
||||
services.TryAddSingleton<ITransportClient>(sp => sp.GetRequiredService<InMemoryTransportClient>());
|
||||
|
||||
return services;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Adds the InMemory transport server only (for Gateway).
|
||||
/// </summary>
|
||||
/// <param name="services">The service collection.</param>
|
||||
/// <param name="configure">Optional configuration action.</param>
|
||||
/// <returns>The service collection for chaining.</returns>
|
||||
public static IServiceCollection AddInMemoryTransportServer(
|
||||
this IServiceCollection services,
|
||||
Action<InMemoryTransportOptions>? configure = null)
|
||||
{
|
||||
services.AddOptions<InMemoryTransportOptions>();
|
||||
if (configure is not null)
|
||||
{
|
||||
services.Configure(configure);
|
||||
}
|
||||
|
||||
services.TryAddSingleton<InMemoryConnectionRegistry>();
|
||||
services.TryAddSingleton<InMemoryTransportServer>();
|
||||
services.TryAddSingleton<ITransportServer>(sp => sp.GetRequiredService<InMemoryTransportServer>());
|
||||
|
||||
return services;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Adds the InMemory transport client only (for Microservice SDK).
|
||||
/// </summary>
|
||||
/// <param name="services">The service collection.</param>
|
||||
/// <param name="configure">Optional configuration action.</param>
|
||||
/// <returns>The service collection for chaining.</returns>
|
||||
public static IServiceCollection AddInMemoryTransportClient(
|
||||
this IServiceCollection services,
|
||||
Action<InMemoryTransportOptions>? configure = null)
|
||||
{
|
||||
services.AddOptions<InMemoryTransportOptions>();
|
||||
if (configure is not null)
|
||||
{
|
||||
services.Configure(configure);
|
||||
}
|
||||
|
||||
services.TryAddSingleton<InMemoryConnectionRegistry>();
|
||||
services.TryAddSingleton<InMemoryTransportClient>();
|
||||
services.TryAddSingleton<ITransportClient>(sp => sp.GetRequiredService<InMemoryTransportClient>());
|
||||
|
||||
return services;
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,22 @@
|
||||
<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.InMemory</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" />
|
||||
</ItemGroup>
|
||||
|
||||
</Project>
|
||||
Reference in New Issue
Block a user