Audience: contributors adding a protocol or understanding the internal
wire representation — the EventMeshFrame format, the FrameAdaptor SPI,
and end-to-end conversion chains per protocol. Part of the offset /
load-balancing / frame design trio; see also
Offset management.
┌─ external protocols (FrameAdaptor SPI) ─────────────────────────────┐
│ │
│ CloudEvents (HTTP/SSE/WS) → CloudEventsFrameAdaptor → EventMeshFrame │
│ MeshMessage (legacy TCP) → MeshMessageFrameAdaptor → EventMeshFrame │
│ A2A (JSON-RPC 2.0) → A2AFrameAdaptor → EventMeshFrame │
│ future protocol → new FrameAdaptor → EventMeshFrame │
│ │
│ The three protocols are peers; each converts directly to Frame — │
│ none of them passes through CloudEvent. │
└──────────────────────────────┬───────────────────────────────────────┘
│ external protocol → Frame, directly
┌──────────────────────────────▼───────────────────────────────────────┐
│ internal (runtime + storage): EventMeshFrame end to end │
│ │
│ ingress publish → Frame → storage.send(Frame) → MQ bytes │
│ MQ bytes → storage.poll() → Frame → dispatch/filter/TTL → Frame │
└──────────────────────────────┬───────────────────────────────────────┘
│ egress: Frame → the matching FrameAdaptor
┌──────────────────────────────▼───────────────────────────────────────┐
│ egress (FrameAdaptor SPI, per client connection protocol) │
│ │
│ Frame → CloudEvents-JSON (SSE / WS / HTTP poll) │
│ Frame → MeshMessage Package (legacy TCP) │
│ Frame → A2A JSON-RPC bytes (A2A callbacks) │
└────────────────────────────────────────────────────────────────────────┘
CloudEvent is not the internal representation — it is one of the
external ingress formats for CloudEvents clients, converted to Frame by
CloudEventsFrameAdaptor on entry. MeshMessage and A2A likewise each have
their own independent adaptor; they never pass through CloudEvent.
fixed header, 14 bytes:
[magic:1=0xEF][ver:1=1][msgType:1][flags:1][seq:4][keyCount:2][dataLen:4]
KV attribute section (keyCount ×):
[nameLen:2][name:UTF-8][valLen:4][value:UTF-8]
data:
raw bytes (streaming chunk/prompt / event business payload)
msgType = STREAM_REQ(1) | STREAM_CHUNK(2) | EVENT(3)
flags = bit0 done | bit1 hasError | bit2 hasMeta (streaming use)
| msgType | Fixed-header fields | KV attributes | data |
|---|---|---|---|
| STREAM_REQ | — | sid=sessionId, replyTo=reply address, model?, conv? |
prompt text |
| STREAM_CHUNK | seq=in-stream sequence, flags.done=termination mark |
sid, etype?, err?, meta?(JSON) |
chunk text |
| EVENT | — | id/type/source/subject/time/emttl/emcorrelationid/empopck/user extensions |
event payload |
Bidirectional conversion between external protocols and EventMeshFrame is
defined by the FrameAdaptor SPI
(eventmesh-protocol-plugin/eventmesh-protocol-api/.../FrameAdaptor.java):
@EventMeshSPI(eventMeshExtensionType = EventMeshExtensionType.PROTOCOL)
public interface FrameAdaptor {
EventMeshFrame toFrame(ProtocolTransportObject proto); // ingress
ProtocolTransportObject fromFrame(EventMeshFrame frame); // egress
String getProtocolType();
}The runtime never calls EventMeshFrame.fromCloudEvent() /
.toCloudEvent() / MeshMessageFrameCodec directly — everything goes
through FrameAdaptors.get(<protocol name>). Adding a new protocol is
just implementing FrameAdaptor + registering the SPI — no runtime code
changes.
| Module | FrameAdaptor | External protocol | Conversion |
|---|---|---|---|
| protocol-api | — (SPI interface + FrameAdaptors loader only) |
— | zero implementations |
| protocol-cloudevents | CloudEventsFrameAdaptor |
CloudEvents-JSON (HTTP/SSE/WS) | CE-JSON bytes ↔ Frame (field mapping via the CE object) |
| protocol-meshmessage | MeshMessageFrameAdaptor |
MeshMessage Package (TCP) | Package ↔ Frame (direct field mapping, zero CE intermediary) |
| protocol-a2a | A2AFrameAdaptor |
A2A JSON-RPC 2.0 | JSON-RPC bytes ↔ Frame (direct field mapping, zero CE intermediary) |
| — | — | deleted (empty shell, no sources) |
Byte-level encode/decode for the internal MQ wire is defined by the
WireCodec SPI (eventmesh-common/.../wire/):
public interface WireCodec {
byte[] encode(StreamRequest request);
StreamRequest decodeRequest(byte[] bytes);
byte[] encode(StreamChunk chunk);
StreamChunk decodeChunk(byte[] bytes);
byte[] encode(CloudEvent event);
CloudEvent decodeEvent(byte[] bytes);
}- Default implementation:
EventMeshFrameCodec(EventMeshFrame ↔ byte[]). - Replaceable via
-Deventmesh.wire.codec=<fqcn>.
Ingress (client → runtime):
SDK sends CloudEvents-JSON bytes
→ UniHttpServer.publish: body = CE-JSON bytes
→ UniIngressService.publish(topic, CloudEvent): CE → EventMeshFrame.fromCloudEvent(event)
→ storage.send(topic, frame): frame.encode() → MQ bytes
Egress (runtime → client):
storage.poll() → frame (EventMeshFrame)
→ SseConnection.send / WsConnection.send / UniHttpServer.poll
→ FrameAdaptors.toCloudEventsJson(frame): frame.toCloudEvent() → CE-JSON serialize
→ written as SSE data: / WS TextFrame / HTTP JSON
Ingress:
legacy TCP SDK sends Package (EventMeshMessage)
→ MeshMessagePackageRouter.route(pkg)
→ FrameAdaptors.get("meshmessage").toFrameSilent(pkg)
→ MeshMessageFrameAdaptor.toFrame:
topic → attributes["subject"]
body → data
header.seq → attributes["id"]
properties → attributes KV
→ TcpRequest.publish(topic, frame)
→ UniIngressService.publish(topic, frame) (Frame overload — skips CE conversion)
→ storage.send(topic, frame)
Egress:
ReliableDispatcher.deliver → NettyTcpPushChannel.deliver(deliveryId, frame, callback)
→ FrameAdaptors.get("meshmessage").fromFrameSilent(frame)
→ MeshMessageFrameAdaptor.fromFrame:
attributes["subject"] → topic
data → body
attributes KV → properties
→ Package(header=ASYNC_MESSAGE_TO_CLIENT, body=EventMeshMessage)
→ channel.writeAndFlush(pkg)
Ingress:
A2A client sends JSON-RPC 2.0 bytes
→ FrameAdaptors.get("a2a").toFrameSilent(new ByteTransport(jsonBytes))
→ A2AFrameAdaptor.toFrame:
parse JSON-RPC → extract method/params/_topic
method → attributes["ema2amethod"]
params._topic → attributes["subject"]
raw JSON → data
→ EventMeshFrame
Egress:
EventMeshA2ATransport receives frame
→ frame.data() = the original A2A JSON-RPC bytes (stored verbatim at ingress)
→ ByteTransport(jsonBytes) → A2A client
Storage plugins (MeshStoragePlugin + LiteTopicCapable) carry
EventMeshFrame directly on send/poll:
send(topic, EventMeshFrame frame):
→ rocketmq5: frame.encode() → MQ message body bytes
→ kafka: frame.encode() → ProducerRecord value
→ rocketmq4: frame.encode() → MQ message body bytes
poll(topic, partition, ...):
→ MQ bytes → EventMeshFrame.decode(bytes)
→ with legacy CE-JSON fallback (old messages → EventMeshFrame.fromCloudEvent(deserialize))
| Layer | Change | Status |
|---|---|---|
| storage SPI | MeshStoragePlugin.send/poll + LiteTopicCapable.sendLite/pullLite all EventMeshFrame |
✅ |
| runtime dispatch | Delivery/BufferedEvent/PushChannel/Connection/DeadLetterSink/ReliableDispatcher/PushService + CloudEventFilter/SubscriptionManager fully Frame-typed | ✅ |
| streaming | Mode-1 (runtime↔agent cross-process) + Mode-2 (in-runtime pub/sub) all EventMeshFrame | ✅ |
| legacy connectors | Producer/Consumer SPI untouched (separate subsystem, speaks CE, boundary conversion) | ✅ kept |
| WireCodec SPI | default EventMeshFrameCodec, replaceable | ✅ |
| FrameAdaptor SPI | CloudEvents as an independent plugin + MeshMessage + A2A each with an independent adaptor | ✅ |