using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Logging.Abstractions;
using System.Diagnostics;
using System.Text.Json;
using System.Threading.Channels;
namespace ModelContextProtocol.Protocol;
///
/// Provides a base class for implementing .
///
///
///
/// The class provides core functionality required by most
/// implementations, including message channel management, connection state tracking, and logging support.
///
///
/// Custom transport implementations should inherit from this class and implement the abstract
/// and methods
/// to handle the specific transport mechanism being used.
///
///
public abstract partial class TransportBase : ITransport
{
private readonly Channel _messageChannel;
private readonly ILogger _logger;
private volatile int _state = StateInitial;
/// The transport has not yet been connected.
private const int StateInitial = 0;
/// The transport is connected.
private const int StateConnected = 1;
/// The transport was previously connected and is now disconnected.
private const int StateDisconnected = 2;
///
/// Initializes a new instance of the class.
///
protected TransportBase(string name, ILoggerFactory? loggerFactory)
: this(name, null, loggerFactory)
{
}
///
/// Initializes a new instance of the class with a specified channel to back .
///
internal TransportBase(string name, Channel? messageChannel, ILoggerFactory? loggerFactory)
{
Name = name;
_logger = loggerFactory?.CreateLogger(GetType()) ?? NullLogger.Instance;
// Unbounded channel to prevent blocking on writes. Ensure AutoDetectingClientSessionTransport matches this.
_messageChannel = messageChannel ?? Channel.CreateUnbounded(new UnboundedChannelOptions
{
SingleReader = true,
SingleWriter = false,
});
}
/// Gets the logger used by this transport.
private protected ILogger Logger => _logger;
///
public virtual string? SessionId { get; protected set; }
///
/// Gets the name that identifies this transport endpoint in logs.
///
///
/// This name is used in log messages to identify the source of transport-related events.
///
protected string Name { get; }
///
public bool IsConnected => _state == StateConnected;
///
public ChannelReader MessageReader => _messageChannel.Reader;
///
public abstract Task SendMessageAsync(JsonRpcMessage message, CancellationToken cancellationToken = default);
///
public abstract ValueTask DisposeAsync();
///
/// Writes a message to the message channel.
///
/// The message to write.
/// The to monitor for cancellation requests. The default is .
protected async Task WriteMessageAsync(JsonRpcMessage message, CancellationToken cancellationToken = default)
{
cancellationToken.ThrowIfCancellationRequested();
if (!IsConnected)
{
// Transport disconnected concurrently. Silently drop rather than throw,
// to avoid surfacing spurious errors during shutdown races.
return;
}
if (_logger.IsEnabled(LogLevel.Debug))
{
var messageId = (message as JsonRpcMessageWithId)?.Id.ToString() ?? "(no id)";
LogTransportReceivedMessage(Name, messageId);
}
bool wrote = _messageChannel.Writer.TryWrite(message);
Debug.Assert(wrote || !IsConnected, "_messageChannel is unbounded; this should only ever return false if the channel has been closed.");
}
///
/// Sets the transport to a connected state.
///
protected void SetConnected()
{
while (true)
{
int state = _state;
switch (state)
{
case StateInitial:
if (Interlocked.CompareExchange(ref _state, StateConnected, StateInitial) == StateInitial)
{
return;
}
break;
case StateConnected:
return;
case StateDisconnected:
throw new IOException("Transport is already disconnected and can't be reconnected.");
default:
Debug.Fail($"Unexpected state: {state}");
return;
}
}
}
///
/// Sets the transport to a disconnected state.
///
/// Optional error information associated with the transport disconnecting. Should be if the disconnect was graceful and expected.
protected void SetDisconnected(Exception? error = null)
{
int state = _state;
switch (state)
{
case StateInitial:
case StateConnected:
_state = StateDisconnected;
_messageChannel.Writer.TryComplete(error);
break;
case StateDisconnected:
return;
default:
Debug.Fail($"Unexpected state: {state}");
break;
}
}
[LoggerMessage(Level = LogLevel.Error, Message = "{EndpointName} transport connect failed.")]
private protected partial void LogTransportConnectFailed(string endpointName, Exception exception);
[LoggerMessage(Level = LogLevel.Error, Message = "{EndpointName} transport send failed for message ID '{MessageId}'.")]
private protected partial void LogTransportSendFailed(string endpointName, string messageId, Exception exception);
[LoggerMessage(Level = LogLevel.Trace, Message = "{EndpointName} transport sending message. Message: '{Message}'.")]
private protected partial void LogTransportSendingMessageSensitive(string endpointName, string message);
///
/// Logs a sending message at Trace level if trace logging is enabled.
///
/// The JSON-RPC message to log.
private protected void LogTransportSendingMessageSensitive(JsonRpcMessage message)
{
if (_logger.IsEnabled(LogLevel.Trace))
{
LogTransportSendingMessageSensitive(Name, JsonSerializer.Serialize(message, McpJsonUtilities.JsonContext.Default.JsonRpcMessage));
}
}
[LoggerMessage(Level = LogLevel.Information, Message = "{EndpointName} transport reading messages.")]
private protected partial void LogTransportEnteringReadMessagesLoop(string endpointName);
[LoggerMessage(Level = LogLevel.Information, Message = "{EndpointName} transport completed reading messages.")]
private protected partial void LogTransportEndOfStream(string endpointName);
[LoggerMessage(Level = LogLevel.Trace, Message = "{EndpointName} transport received message. Message: '{Message}'.")]
private protected partial void LogTransportReceivedMessageSensitive(string endpointName, string message);
[LoggerMessage(Level = LogLevel.Debug, Message = "{EndpointName} transport received message with ID '{MessageId}'.")]
private protected partial void LogTransportReceivedMessage(string endpointName, string messageId);
[LoggerMessage(Level = LogLevel.Trace, Message = "{EndpointName} transport received unexpected message. Message: '{Message}'.")]
private protected partial void LogTransportMessageParseUnexpectedTypeSensitive(string endpointName, string message);
[LoggerMessage(Level = LogLevel.Information, Message = "{EndpointName} transport message parsing failed.")]
private protected partial void LogTransportMessageParseFailed(string endpointName, Exception exception);
[LoggerMessage(Level = LogLevel.Trace, Message = "{EndpointName} transport message parsing failed. Message: '{Message}'.")]
private protected partial void LogTransportMessageParseFailedSensitive(string endpointName, string message, Exception exception);
[LoggerMessage(Level = LogLevel.Information, Message = "{EndpointName} transport message reading canceled.")]
private protected partial void LogTransportReadMessagesCancelled(string endpointName);
[LoggerMessage(Level = LogLevel.Warning, Message = "{EndpointName} transport message reading failed.")]
private protected partial void LogTransportReadMessagesFailed(string endpointName, Exception exception);
[LoggerMessage(Level = LogLevel.Information, Message = "{EndpointName} shutting down.")]
private protected partial void LogTransportShuttingDown(string endpointName);
[LoggerMessage(Level = LogLevel.Warning, Message = "{EndpointName} shutdown failed.")]
private protected partial void LogTransportShutdownFailed(string endpointName, Exception exception);
[LoggerMessage(Level = LogLevel.Warning, Message = "{EndpointName} shutdown failed waiting for message reading completion.")]
private protected partial void LogTransportCleanupReadTaskFailed(string endpointName, Exception exception);
[LoggerMessage(Level = LogLevel.Information, Message = "{EndpointName} shut down.")]
private protected partial void LogTransportShutDown(string endpointName);
[LoggerMessage(Level = LogLevel.Warning, Message = "{EndpointName} received message before connected.")]
private protected partial void LogTransportMessageReceivedBeforeConnected(string endpointName);
[LoggerMessage(Level = LogLevel.Warning, Message = "{EndpointName} endpoint event received out of order.")]
private protected partial void LogTransportEndpointEventInvalid(string endpointName);
[LoggerMessage(Level = LogLevel.Warning, Message = "{EndpointName} failed to parse event.")]
private protected partial void LogTransportEndpointEventParseFailed(string endpointName, Exception exception);
[LoggerMessage(Level = LogLevel.Warning, Message = "{EndpointName} failed to parse event. Message: '{Message}'.")]
private protected partial void LogTransportEndpointEventParseFailedSensitive(string endpointName, string message, Exception exception);
}