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); }
EventMeshFrameCodec (EventMeshFrame ↔ byte[]).-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 | ✅ |