Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -183,12 +183,29 @@ public MediatorRuntime Build(IServiceProvider applicationServices)
var pipelines = new Dictionary<Type, MediatorDelegate>();
var notificationTerminals = new Dictionary<Type, NotificationTerminals>();

var serviceName = _options.ServiceName
?? Environment.GetEnvironmentVariable("SERVICE_NAME")
?? Environment.GetEnvironmentVariable("OTEL_SERVICE_NAME")
?? Assembly.GetEntryAssembly()?.GetName().Name
?? "default";
var handlerDescriptions = new List<MediatorHandlerDescription>();

foreach (var (handlerType, configureDelegate) in _handlerDescriptors)
{
var descriptor = new MediatorHandlerDescriptor(configContext, handlerType);
configureDelegate(descriptor);
var config = descriptor.CreateConfiguration();

var handlerName = MediatorDescriptionHelpers.GetTypeName(config.HandlerType!);
var messageName = MediatorDescriptionHelpers.GetTypeName(config.MessageType!);
handlerDescriptions.Add(new MediatorHandlerDescription(
MediatorUrn.Handler(serviceName, handlerName),
handlerName,
config.Kind,
MediatorUrn.Message(serviceName, messageName),
messageName,
config.ResponseType is { } responseType ? MediatorDescriptionHelpers.GetTypeName(responseType) : null));

var terminal = config.Delegate ?? BuildPipelineViaReflection(config);

if (config.Kind == MediatorHandlerKind.Notification)
Expand Down Expand Up @@ -236,12 +253,18 @@ public MediatorRuntime Build(IServiceProvider applicationServices)

var pools = applicationServices.GetRequiredService<IMediatorPools>();

var description = new MediatorDescription(
MediatorUrn.Mediator(serviceName),
serviceName,
handlerDescriptions.OrderBy(h => h.Id, StringComparer.Ordinal).ToArray());

return new MediatorRuntime(
pipelines.ToFrozenDictionary(),
notificationPipelines.ToFrozenDictionary(),
pools,
features,
_options.NotificationPublishMode);
_options.NotificationPublishMode,
description);
}

[RequiresDynamicCode("Use source-generated mediator module registration for AOT compatibility.")]
Expand Down
13 changes: 13 additions & 0 deletions src/Mocha/src/Mocha.Mediator/Descriptions/MediatorDescription.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
namespace Mocha.Mediator;

/// <summary>
/// Describes a configured mediator, including the service it is scoped to and its registered handlers,
/// for diagnostic and visualization purposes.
/// </summary>
/// <param name="Id">The stable URN identity of the mediator.</param>
/// <param name="ServiceName">The service name the mediator is scoped to.</param>
/// <param name="Handlers">The registered handlers.</param>
public sealed record MediatorDescription(
string Id,
string ServiceName,
IReadOnlyList<MediatorHandlerDescription> Handlers);
Original file line number Diff line number Diff line change
@@ -0,0 +1,16 @@
namespace Mocha.Mediator;

internal static class MediatorDescriptionHelpers
{
public static string GetTypeName(Type type)
{
if (!type.IsGenericType)
{
return type.Name;
}

var genericTypeName = type.Name.Split('`')[0];
var genericArgs = string.Join(", ", type.GetGenericArguments().Select(GetTypeName));
return $"{genericTypeName}<{genericArgs}>";
}
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,18 @@
namespace Mocha.Mediator;

/// <summary>
/// Describes a single registered mediator handler for diagnostic and visualization purposes.
/// </summary>
/// <param name="Id">The stable URN identity of the handler.</param>
/// <param name="Name">The readable handler type name.</param>
/// <param name="Kind">The kind of handler (command, command with response, query, or notification).</param>
/// <param name="MessageId">The URN identity of the message the handler handles.</param>
/// <param name="MessageName">The readable message type name.</param>
/// <param name="ResponseTypeName">The readable response type name, or <c>null</c> for void commands and notifications.</param>
public sealed record MediatorHandlerDescription(
string Id,
string Name,
MediatorHandlerKind Kind,
string MessageId,
string MessageName,
string? ResponseTypeName);
26 changes: 26 additions & 0 deletions src/Mocha/src/Mocha.Mediator/Descriptions/MediatorUrn.cs
Original file line number Diff line number Diff line change
@@ -0,0 +1,26 @@
namespace Mocha.Mediator;

/// <summary>
/// Builds the URN identity strings for mediator entities. URNs uniquely identify the mediator,
/// its handlers, and its messages within a service for use in descriptions, traces, and logs.
/// </summary>
internal static class MediatorUrn
{
/// <summary>
/// Returns a URN identifying the mediator within a service.
/// </summary>
public static string Mediator(string service)
=> $"urn:mocha:svc:{service}:mediator";

/// <summary>
/// Returns a URN identifying a handler within a service.
/// </summary>
public static string Handler(string service, string handler)
=> $"urn:mocha:svc:{service}:mediator:handler:{handler}";

/// <summary>
/// Returns a URN identifying a message (command, query, or notification) within a service.
/// </summary>
public static string Message(string service, string message)
=> $"urn:mocha:svc:{service}:mediator:message:{message}";
}
7 changes: 7 additions & 0 deletions src/Mocha/src/Mocha.Mediator/IMediatorRuntime.cs
Original file line number Diff line number Diff line change
Expand Up @@ -10,4 +10,11 @@ public interface IMediatorRuntime
/// Gets the read-only feature collection for this mediator runtime.
/// </summary>
IFeatureCollection Features { get; }

/// <summary>
/// Gets a description of the mediator and its registered handlers for diagnostic and
/// visualization purposes.
/// </summary>
/// <returns>A <see cref="MediatorDescription"/> describing the mediator.</returns>
MediatorDescription Describe();
}
7 changes: 7 additions & 0 deletions src/Mocha/src/Mocha.Mediator/MediatorOptions.cs
Original file line number Diff line number Diff line change
Expand Up @@ -18,4 +18,11 @@ public sealed class MediatorOptions
/// Default is <see cref="NotificationPublishMode.Sequential"/>.
/// </summary>
public NotificationPublishMode NotificationPublishMode { get; set; } = NotificationPublishMode.Sequential;

/// <summary>
/// Gets or sets the logical service name used to scope mediator URN identities.
/// When <c>null</c>, falls back to the <c>SERVICE_NAME</c> environment variable, then the
/// <c>OTEL_SERVICE_NAME</c> environment variable, then the entry assembly name.
/// </summary>
public string? ServiceName { get; set; }
}
8 changes: 7 additions & 1 deletion src/Mocha/src/Mocha.Mediator/MediatorRuntime.cs
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ public sealed class MediatorRuntime : IMediatorRuntime
private readonly FrozenDictionary<Type, MediatorDelegate> _pipelines;
private readonly FrozenDictionary<Type, ImmutableArray<MediatorDelegate>> _notificationPipelines;
private readonly ObjectPool<MediatorContext> _contextPool;
private readonly MediatorDescription _description;

[ThreadStatic]
private static MediatorContext? s_cached;
Expand All @@ -23,13 +24,15 @@ internal MediatorRuntime(
FrozenDictionary<Type, ImmutableArray<MediatorDelegate>> notificationPipelines,
IMediatorPools pools,
IFeatureCollection features,
NotificationPublishMode notificationPublishMode)
NotificationPublishMode notificationPublishMode,
MediatorDescription description)
{
_pipelines = pipelines;
_notificationPipelines = notificationPipelines;
_contextPool = pools.MediatorContext;
Features = features;
NotificationPublishMode = notificationPublishMode;
_description = description;
}

/// <summary>
Expand All @@ -42,6 +45,9 @@ internal MediatorRuntime(
/// </summary>
internal NotificationPublishMode NotificationPublishMode { get; }

/// <inheritdoc />
public MediatorDescription Describe() => _description;

/// <summary>
/// Gets the compiled pipeline delegate for the specified message type.
/// </summary>
Expand Down
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
using System.Diagnostics.CodeAnalysis;
using Mocha.Middlewares;
using static System.StringSplitOptions;

namespace Mocha.Transport.InMemory;
Expand Down Expand Up @@ -33,13 +34,13 @@ public InMemoryMessagingTransport(Action<IInMemoryMessagingTransportDescriptor>
public override MessagingTopology Topology => _topology;

/// <summary>
/// Builds the in-memory topology URI from the host's assembly name and creates the
/// Builds the in-memory topology URI from the host's service name and creates the
/// <see cref="InMemoryMessagingTopology"/> that holds all topics, queues, and bindings for
/// this transport.
/// </summary>
/// <remarks>
/// Called once during the messaging host initialization phase, after the base transport has
/// been initialized. The topology address uses the transport schema and the assembly name as
/// been initialized. The topology address uses the transport schema and the service name as
/// the host component, ensuring that endpoint addresses are scoped to this application.
/// No network connections are established because all messaging is in-process.
/// </remarks>
Expand All @@ -49,7 +50,7 @@ protected override void OnAfterInitialized(IMessagingSetupContext context)
var builder = new UriBuilder
{
Scheme = Schema,
Host = context.Host.AssemblyName, // service name might be nicer
Host = context.Host.EffectiveServiceName,
Path = "/"
};
_topology = new InMemoryMessagingTopology(this, builder.Uri);
Expand Down Expand Up @@ -86,6 +87,7 @@ public override TransportDescription Describe()
{
entities.Add(
new TopologyEntityDescription(
MochaUrn.TopologyEntity(topic.Address?.ToString(), "topic", topic.Name),
"topic",
topic.Name,
topic.Address?.ToString(),
Expand All @@ -97,6 +99,7 @@ public override TransportDescription Describe()
{
entities.Add(
new TopologyEntityDescription(
MochaUrn.TopologyEntity(queue.Address?.ToString(), "queue", queue.Name),
"queue",
queue.Name,
queue.Address?.ToString(),
Expand All @@ -106,24 +109,29 @@ public override TransportDescription Describe()

foreach (var binding in _topology.Bindings)
{
var source = binding.Source.Address?.ToString();
var target = binding switch
{
InMemoryQueueBinding qb => qb.Destination.Address?.ToString(),
InMemoryTopicBinding tb => tb.Destination.Address?.ToString(),
_ => null
};

links.Add(
new TopologyLinkDescription(
MochaUrn.TopologyLink(binding.Address?.ToString(), "bind", source, target),
"bind",
binding.Address?.ToString(),
binding.Source.Address?.ToString(),
binding switch
{
InMemoryQueueBinding qb => qb.Destination.Address?.ToString(),
InMemoryTopicBinding tb => tb.Destination.Address?.ToString(),
_ => null
},
source,
target,
"forward",
new Dictionary<string, object?> { ["origin"] = binding.Origin }));
}

var topology = new TopologyDescription(_topology.Address.ToString(), entities, links);

return new TransportDescription(
Urn,
_topology.Address.ToString(),
Name,
Schema,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -215,6 +215,7 @@ public override TransportDescription Describe()
{
entities.Add(
new TopologyEntityDescription(
MochaUrn.TopologyEntity(topic.Address.ToString(), "topic", topic.Name),
"topic",
topic.Name,
topic.Address.ToString(),
Expand All @@ -230,6 +231,7 @@ public override TransportDescription Describe()
{
entities.Add(
new TopologyEntityDescription(
MochaUrn.TopologyEntity(queue.Address.ToString(), "queue", queue.Name),
"queue",
queue.Name,
queue.Address.ToString(),
Expand All @@ -246,6 +248,11 @@ public override TransportDescription Describe()
{
links.Add(
new TopologyLinkDescription(
MochaUrn.TopologyLink(
subscription.Address.ToString(),
"subscription",
subscription.Source.Address.ToString(),
subscription.Destination.Address.ToString()),
"subscription",
subscription.Address.ToString(),
subscription.Source.Address.ToString(),
Expand All @@ -261,6 +268,7 @@ public override TransportDescription Describe()
var topology = new TopologyDescription(_topology.Address.ToString(), entities, links);

return new TransportDescription(
Urn,
_topology.Address.ToString(),
Name,
Schema,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -156,6 +156,7 @@ public override TransportDescription Describe()
{
entities.Add(
new TopologyEntityDescription(
MochaUrn.TopologyEntity(exchange.Address?.ToString(), "exchange", exchange.Name),
"exchange",
exchange.Name,
exchange.Address?.ToString(),
Expand All @@ -174,6 +175,7 @@ public override TransportDescription Describe()
{
entities.Add(
new TopologyEntityDescription(
MochaUrn.TopologyEntity(queue.Address?.ToString(), "queue", queue.Name),
"queue",
queue.Name,
queue.Address?.ToString(),
Expand All @@ -190,17 +192,21 @@ public override TransportDescription Describe()

foreach (var binding in _topology.Bindings)
{
var source = binding.Source.Address?.ToString();
var target = binding switch
{
RabbitMQQueueBinding qb => qb.Destination.Address?.ToString(),
RabbitMQExchangeBinding eb => eb.Destination.Address?.ToString(),
_ => null
};

links.Add(
new TopologyLinkDescription(
MochaUrn.TopologyLink(binding.Address?.ToString(), "bind", source, target),
"bind",
binding.Address?.ToString(),
binding.Source.Address?.ToString(),
binding switch
{
RabbitMQQueueBinding qb => qb.Destination.Address?.ToString(),
RabbitMQExchangeBinding eb => eb.Destination.Address?.ToString(),
_ => null
},
source,
target,
"forward",
new Dictionary<string, object?>
{
Expand All @@ -213,6 +219,7 @@ public override TransportDescription Describe()
var topology = new TopologyDescription(_topology.Address.ToString(), entities, links);

return new TransportDescription(
Urn,
_topology.Address.ToString(),
Name,
Schema,
Expand Down
15 changes: 14 additions & 1 deletion src/Mocha/src/Mocha/Consumers/Consumer.cs
Original file line number Diff line number Diff line change
Expand Up @@ -48,6 +48,11 @@ protected virtual void Configure(IConsumerDescriptor descriptor) { }
/// </summary>
public Type Identity { get; }

/// <summary>
/// Gets the stable URN identity of this consumer.
/// </summary>
public string Urn { get; private set; } = null!;

private protected ConsumerDelegate Pipeline { get; private set; } = null!;

/// <summary>
Expand Down Expand Up @@ -115,6 +120,8 @@ internal void Initialize(IMessagingSetupContext context)

// TODO should we assign a default name in the Action? GetType().Name?
Name = Configuration.Name ?? throw ThrowHelper.ConsumerNameRequired();
Urn = MochaUrn.Consumer(context.Host.EffectiveServiceName, Name);

foreach (var route in Configuration!.Routes)
{
route.Consumer = this;
Expand Down Expand Up @@ -184,7 +191,13 @@ private void MarkInitialized()
/// <returns>A <see cref="ConsumerDescription"/> containing the consumer's name, type, and optional saga association.</returns>
public virtual ConsumerDescription Describe()
{
return new ConsumerDescription(Name, DescriptionHelpers.GetTypeName(Identity), Identity.FullName, null, false);
return new ConsumerDescription(
Urn,
Name,
DescriptionHelpers.GetTypeName(Identity),
Identity.FullName,
null,
false);
}

/// <summary>
Expand Down
Loading
Loading