Sparkdo.Messaging.Abstractions 3.0.0-preview.1

Sparkdo.Messaging.Abstractions

跨程序集边界的那一面：消息与身份怎么表示、处理器与总线长什么样、传输要实现哪三个端口、 载荷与消息头在谁的 ownership 之下。零依赖、零实现、零 DI——只引这个包不会被拖进内核。

NuGet 页上这一份就是包内 README；它与仓库里的是同一个文件，所以里面没有相对链接， 路径一律写成行内代码，需要跳转请回仓库 src/messaging。

<PackageReference Include="Sparkdo.Messaging.Abstractions" Version="1.0.0" />

什么时候只引这个包

处理器本身（IEventHandler<T>）也在这个包：一个只声明处理器、不接总线的库同样只需要它。

1. 定义一个消息：POCO 就够

不需要基类、接口或属性。内核按 CLR 类型找路由，线身份由命名策略算出来：

public sealed record OrderPlaced(string OrderId, decimal Amount);

默认身份（DefaultNamePolicy，实现在内核包里）是 urn:message:{命名空间 kebab}:{类型名 kebab}：

身份是线协议的一部分：队列名、话题、幂等键都从它派生。改了类型的命名空间或名字， 就等于换了一条队列——上一版攒下的消息没人认领。两种写法把它钉死：

[Message(Identity = "urn:message:sparkdo.orders:order-placed")]
public sealed record OrderPlaced(string OrderId, decimal Amount);

// 或在建图时覆盖（优先级更高）：
builder.AddMessage<OrderPlaced>("urn:message:sparkdo.orders:order-placed");

身份来源的优先级：AddMessage<T>(identity) > [Message(Identity = …)] > 命名策略。

[Message] 三个属性

[Message] 整体是可选的。ContentType 接受任意 MIME 串（不在 Json/Xml/Protobuf 三个驻留 实例里也照样成立），但建图期必须按它找到登记过的读写器，否则抛"没有可用的读写器"——不会静默 退回 JSON。SendOptions.ContentType 同理：它只改信封上的标签，不改写 bytes 的那个读写器。

两处默认命名会撞的形状

MessageIdentity 只用 Type.Name，不用 FullName：

（订阅者名不受这条影响：它走 FullName，闭合泛型的实参用 + 拼接并剥掉程序集版本， 所以队列名不会跟着版本号变。）

2. 值类型：身份、地址、标识

var byParse = EndpointAddress.Parse("memory://topic/urn:message:sparkdo.orders:order-placed");

new EndpointAddress("不合法也收");   // 收下了——构造只是赋值
EndpointAddress.Parse("orders");     // ArgumentException：端点地址必须是绝对 URI，收到 'orders'。
byParse.Scheme;                      // "memory"（"://" 之前那段；没有分隔符时是整个值）

Parse 只保证「绝对 URI」这一件事。四段文法不在 Parse 里校验——它在 MessagingAddresses（核心侧读回）与各传输自己的分解里。所以 memory://exchange/x 能 Parse 成功，之后要么被传输当场拒绝（Postgres 抛错并写出三种正确形态），要么被当成 认不出的地址退回按身份扇出（InMemory）。地址自己拼的时候，用 MessagingAddresses。

3. 地址文法：MessagingAddresses

四种形态，四个传输共用同一套（{scheme} 由传输自己给：memory / rabbitmq / postgres）：

EndpointAddress publish = MessagingAddresses.Publish("rabbitmq", identity);
EndpointAddress sub     = MessagingAddresses.Subscription("rabbitmq", identity, "sparkdo.orders.order-placed-handler");
EndpointAddress dead    = MessagingAddresses.DeadLetter("rabbitmq", identity, "sparkdo.orders.order-placed-handler");

// 反解：只认上面四种形态，认不出就 false，不猜
MessagingAddresses.TryReadIdentity(sub, out MessageIdentity back);   // true，back == identity

// 热路径只判「是不是这个身份的发布形」：命中直接给身份，不切新串
MessagingAddresses.IsPublishAddressOf(publish, identity);            // true

构造那四个方法时 scheme、subscriber 为空即抛（发布地址没有订阅者段）。 TryReadIdentity 拒绝的形状：不认识的第一段、发布地址尾巴里多出的 /、空身份、订阅者段之后还有一段。 IsPublishAddressOf 是热路径版本：只认发布形，形状与期望身份都对得上时不切新串。

订阅者为什么必须进地址：一条 broker 队列只能代表一个订阅者。队列名若只由身份决定， 三个订阅者就变成"三个消费者抢一条队列"，每家只拿到三分之一——而"发布抵达每一个订阅者" 是核心的派发语义。反过来，"重启不丢"也靠它：同名重起才找得回自己那条队列。

4. 处理器契约

public interface IEventHandler<in TEvent> where TEvent : notnull
{
    ValueTask HandleAsync(TEvent message, CancellationToken cancellationToken);
}

请求/响应另有一对：

public interface IEventRequest<TResponse> : IEventRequest { }        // 请求消息自己声明返回什么

public interface IEventRequestHandler<in TRequest, TResponse>
    where TRequest : IEventRequest<TResponse>
{
    ValueTask<TResponse> HandleAsync(TRequest request, CancellationToken cancellationToken);
}

只要"收到就处理、不用回"的请求处理器还有单泛型版 IEventRequestHandler<in TRequest>。 返回的响应会自动回投给请求方——前提是这条投递的信封里有回投地址；当事件投进来（没有 Response）时，返回值被丢弃，处理器照样跑完。

5. 总线契约：发布、发送、请求

public interface IMessageBus : IPublisher, IRequester, ISender { }

ValueTask PublishAsync<TMessage>(in TMessage message, in PublishOptions options, CancellationToken ct);
ValueTask SendAsync<TMessage>(in TMessage message, in SendOptions options, CancellationToken ct);
ValueTask<TResponse> RequestAsync<TRequest, TResponse>(in TRequest request, in SendOptions options, CancellationToken ct)
    where TRequest : notnull, IEventRequest<TResponse>;

发布与发送走的是同一条派发路，PublishAsync 只是把 PublishOptions 折算成 SendOptions 再调同一个终端。差别只在能填哪些格子：PublishOptions 没有 Endpoint、Response、 MessageId、RequestTimeout。

不设 Endpoint 时，SendAsync 与 PublishAsync 落到的目的地是同一个发布地址—— "发送＝单点投递"这件事要靠显式 Endpoint 才成立，不是默认行为。

SendOptions / PublishOptions 逐格

两个都是 readonly struct，属性全 init、默认 null，Default 就是零值。零值不是"设成 null"， 而是"这一格让内核自己决定"。

ExpirationTime 拒得掉是刻意的：属性在但设了不生效，比根本没有这个属性难查得多。

请求方拿到的两种失败，是两件事

RequestAsync 会覆写 SendOptions.MessageId 与 Response：它自己生成关联标识、自己开回投订阅。 这两格在请求路径上不是调用方可控的输入。

6. Envelope：线上传的那 11 格

Envelope 是 readonly struct，全部 init。 typed 路径上由内核填写，调用方只在原样发布与 发件箱入队两处自己拼：

7. 传输端口：写一个传输要满足什么

public interface ITransportPort : IAsyncDisposable
{
    string Scheme { get; }
    TransportCapability Capabilities { get; }
    ValueTask InitializeAsync(TransportContext context, CancellationToken ct);   // 恰好一次
    ValueTask StartAsync(CancellationToken ct);                                   // 幂等
    ValueTask StopAsync(CancellationToken ct);                                    // 幂等；停后拒收发
    ValueTask<IOutboundPort> OpenOutboundAsync(EndpointAddress address, CancellationToken ct);
    ValueTask<IInboundPort> OpenInboundAsync(EndpointAddress address, CancellationToken ct);
}

public interface IOutboundPort
{
    ValueTask<SendReceipt> SendAsync(Envelope envelope, PayloadLease payload, CancellationToken ct);
}

public interface IInboundPort : IAsyncDisposable
{
    IAsyncEnumerable<IDeliveryLease> ReceiveAsync(CancellationToken ct);
    ValueTask DisposeAsync();          // 默认空实现：不放手也是合法的
}

public interface IDeliveryLease
{
    Envelope Envelope { get; }
    PayloadLease Payload { get; }
    ValueTask SettleAsync(DeliveryOutcome outcome, CancellationToken ct);   // 恰好一次
}

核心的调用顺序（写传输时要按这个前提设计资源）：

三条容易踩空的口径：

能力声明：声明什么就要被问责什么

[Flags] None = 0; Durable = 1; NativeSchedule = 4; Backpressure = 64;

位值 2 / 8 / 16 / 32 已删除（Ordered、Session、NativeDedup、Batch 都因为没有读者）， 且位值不复用。加一位的前提是有一位读者，None 是合法声明。

结算：DeliveryOutcome

SettleAsync 恰好一次：同一租约第二次结算必须失败（基座 settlement-exactly-once 逐传输问责）。

8. MessageHeaders：线上那格字节

信封只有一个头槽位，格式与值域在这里闭合。

偏移  内容
0     版本 = 1（1 字节，FormatVersion）
1     条目数（int32，小端）
5     每条：键（int32 长度 + UTF-8 字节）+ 值类型标签（1 字节）+ 值

// 写（调用方只能填自己的键；框架那四格是 Encode 的位置参数，由内核写）
var bytes = MessageHeaders.Encode(new Dictionary<string, object?> { ["tenant"] = "t1", ["attempt"] = 2 });

// 读：整本
Dictionary<string, object?> all = MessageHeaders.Decode(envelope.Headers.Span);

// 读：单键，不建字典，也不抛（版本不认识/截断/不是字符串/没这个键 → false）
bool has = MessageHeaders.TryGetString(envelope.Headers.Span, "tenant", out string? tenant);

三个必须知道的语义：

搬运由基座用例 headers-survive-the-round-trip 逐传输问责（含 null 值条目与键的大小写）。 RabbitMQ 把这段字节放进 AMQP 头 x-sparkdo-headers，Postgres 存成一列。

9. 载荷：PayloadBuffer 与 PayloadLease

typed 路径上你几乎不该碰这两个类型——发布、请求、调度、Saga 都在内核内部租借与归还。 需要碰的只有两种人：写 IMessageSerializer 的（拿到 IBufferWriter<byte>）和写传输的。

所有权转移点在 IOutboundPort.SendAsync：交接之后归传输/接收侧归还。释放之后 Memory 仍可读， 但内容不再有保证——这不是一个可依赖的状态，别把它当"缓冲区还留在原地"。

PayloadBuffer 自己实现 IBufferWriter<byte>，所以序列化器写它就行：

var buffer = PayloadBuffer.Rent();
serializer.Write(buffer, message);          // 写进 IBufferWriter
var lease = buffer.ToLease();               // 交接，不复制

10. 序列化契约

public interface IMessageSerializer
{
    MessageContentType ContentType { get; }
    void Write<TMessage>(IBufferWriter<byte> writer, TMessage message);
    TMessage Deserialize<TMessage>(ReadOnlySpan<byte> payload);
}

public interface IMessageSerializerRegistry
{
    IMessageSerializer? Find(MessageContentType contentType);   // 找不到返回 null，不抛
}

IMessageSerializerRegistry 的实现不在这个包（内核包里有 MessageSerializerRegistry， JSON 桥在 Sparkdo.Messaging.Serialization.Json）。这里只给契约，就是为了上面那张表不因实现变化而改。

11. 故障与结论：SendReceipt / TransportFault

public readonly record struct SendReceipt(bool Accepted, TransportFault? Fault);
public readonly record struct TransportFault(TransportFaultKind Kind, bool Retryable);
public enum TransportFaultKind { Transient = 0, Permanent = 1, Canceled = 2 }

Accepted = true 不等于"有人收到"：无人订阅的发布是受理的（基座 publish-without-subscriber-accepted）。 broker 档要先订阅、后发布。

12. 常见误用一览

这包刻意没有的东西

没有 DI 扩展、没有默认实现、没有 IMessageBus 的任何实现、没有 builder、没有配置对象。 它的全部价值是：这个包里的类型是线协议的一部分，改它要走公开面基线（approval 文件） 与逐传输的契约测试，不能顺手改。