fix(protocol): single protocol path — EventMeshFrame across HTTP ingress, filter chain and TCP egress (#5299)

docs/eventmesh-uni-architecture-redesign.md

@@ -4882,7 +4882,106 @@ EventMesh 实例本地自采负载指标(`LoadMeter`):
 
 1. **单测**:`EventMeshFrame` 全 msgType 互转(12 例);`OffsetStore` 两 key 空间共存;`LoadMeter` 指标 + 每 client 画像;`ClusterMembership` 心跳负载;dispatch 管线 Frame 化(ReliableDispatcher/SubscriptionManager/ClusterCoordinator)。
 2. **E2E**(真 broker):streaming 多轮 + Mode 2 pub/sub + **普通 pub/sub(RocketMQ5BrokerIntegrationTest 2/2)**全绿,内部全程 EventMeshFrame 往返正确;**LegacyTcpClientIntegrationTest(旧 TCP SDK)全绿**(MeshMessage↔Frame 直接转换)。
-3. **构建**:系统 gradle 8.5 + WEOA Nexus(offline)。
+3. **构建**:系统 gradle 8.5 + WEOA Nexus(offline)。
+
+
+### 19.6 #5299 验收矩阵：单协议路径 + ingress/egress 适配
+
+
+
+
+
+#### 协议状态标签
+
+
+
+
+
+| 标签 | 含义 | 当前协议 |
+
+
+|------|------|----------|
+
+
+| **primary** | ingress/egress 全程经 `FrameAdaptor` SPI，内部全程 `EventMeshFrame` | CloudEvents（HTTP/SSE/WS）、MeshMessage（legacy TCP） |
+
+
+| **beta** | 已有 `FrameAdaptor`，但端到端链路未完全收口 | A2A（JSON-RPC 2.0） |
+
+
+| **legacy** | 仅保留兼容桥，不再作为内部表示 | CloudEvent 作为内部中间表示（已废弃，见 §19.1） |
+
+
+
+
+
+#### 路径验收矩阵
+
+
+
+
+
+| 路径 | ingress | egress | 状态 |
+
+
+|------|---------|--------|------|
+
+
+| HTTP publish / publishBatch / lite publish / lite poll / request / reply | `FrameAdaptors.get("cloudevents").toFrame(...)` | `FrameAdaptor.toCloudEventsJson(...)` | ✅ primary |
+
+
+| ingress 安全链（TokenAuth / Acl / SignatureVerifier） | 直接读 `frame.attributes()` | — | ✅ primary |
+
+
+| legacy TCP ingress | `MeshMessagePackageRouter` → `FrameAdaptors.get("meshmessage").toFrameSilent(pkg)` | — | ✅ primary |
+
+
+| legacy TCP egress | `NettyTcpPushChannel` → `FrameAdaptors.get("meshmessage").fromFrameSilent(frame)` | MeshMessage `Package` | ✅ primary |
+
+
+| A2A gateway | `A2AFrameAdaptor` | A2A JSON-RPC bytes | 🟡 beta |
+
+
+
+
+
+#### 子 PR 落地情况
+
+
+
+
+
+| 子 PR | 内容 | 状态 |
+
+
+|-------|------|------|
+
+
+| A | `UniHttpServer` 8 个 ingress 端点改经 `FrameAdaptor` SPI；`UniIngressService` 新增 `publishBatchFrames` / `publishLiteFrame` / `pollLiteFrames` / `requestFrame` / `replyFrame` 五个 `EventMeshFrame` typed 方法 | ✅ |
+
+
+| B | 安全链（`IngressFilter` / `FilterChain` / `AclFilter` / `TokenAuthFilter` / `SignatureVerifierFilter`）改吃 `EventMeshFrame`；租户 / 签名 / token 直接读 `frame.attributes()`。CloudEvent 重载保留为 `@Deprecated` 桥 | ✅ |
+
+
+| C | TCP egress：`TcpFrameCodec.encodePush` / `TcpPushChannel.deliver` 改吃 `EventMeshFrame`（去掉 `frame.toCloudEvent()` 往返）；删除死代码 `CloudEventToPackageBody` / `MeshEventToPackageBody`；`UniTcpServer` 去掉未使用的 `bodyMapper` 构造参数 | ✅ |
+
+
+| D | 本节验收矩阵 + 协议状态标签 | ✅ |
+
+
+
+
+
+#### 随 #5299 删除 / 废弃
+
+
+
+
+
+- `CloudEventToPackageBody`、`MeshEventToPackageBody`：CloudEvent 时代的 TCP egress 编码接口。`NettyTcpPushChannel` 接管 egress 后成为死代码，随 Sub-PR C 删除。
+
+
+- `IngressFilter.check(CloudEvent, FilterContext)`：`@Deprecated` 桥接，仅供尚未迁移的自定义 filter 编译通过；等 A2A（beta）收口后移除。
 
 ---
 

eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/http/UniHttpServer.java

@@ -17,6 +17,9 @@
 
 package org.apache.eventmesh.runtime.http;
 
+import org.apache.eventmesh.common.protocol.ByteTransport;
+import org.apache.eventmesh.common.wire.EventMeshFrame;
+import org.apache.eventmesh.protocol.api.FrameAdaptors;
 import org.apache.eventmesh.runtime.admin.UniAdminService;
 import org.apache.eventmesh.runtime.ingress.UniIngressService;
 import org.apache.eventmesh.runtime.push.BufferedEvent;
@@ -35,10 +38,6 @@
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 
-import io.cloudevents.CloudEvent;
-import io.cloudevents.core.provider.EventFormatProvider;
-import io.cloudevents.jackson.JsonFormat;
-
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.node.ObjectNode;
@@ -249,10 +248,13 @@ private boolean checkSecurity(HttpExchange exchange, String topic, String client
         org.apache.eventmesh.runtime.security.FilterContext ctx =
             new org.apache.eventmesh.runtime.security.FilterContext(topic, clientId, tenant, credential,
                 exchange.getRemoteAddress().getAddress().getHostAddress());
-        // For non-publish endpoints there's no CloudEvent body to check; use a minimal stub.
-        io.cloudevents.CloudEvent stubEvent = io.cloudevents.core.builder.CloudEventBuilder.v1()
-            .withId("security-check").withSource(java.net.URI.create("eventmesh")).withType("security").build();
-        org.apache.eventmesh.runtime.security.FilterVerdict verdict = filterChain.check(stubEvent, ctx);
+        // For non-publish endpoints there's no EventMeshFrame to check; build a minimal
+        // security-check frame so the filter chain has *something* to evaluate. The filters
+        // themselves read tenant / credential from the FilterContext, so a frame with empty
+        // attributes is sufficient for the auth/acl decision (#5299 sub-PR B).
+        org.apache.eventmesh.common.wire.EventMeshFrame stubFrame =
+            org.apache.eventmesh.common.wire.EventMeshFrame.event(java.util.Collections.emptyMap(), new byte[0]);
+        org.apache.eventmesh.runtime.security.FilterVerdict verdict = filterChain.check(stubFrame, ctx);
         if (!verdict.isAllowed()) {
             writeJson(exchange, verdict.getRejectStatus(), error(verdict.getReason()));
             return false;
@@ -285,28 +287,34 @@ private void publish(HttpExchange exchange) throws IOException {
             return;
         }
         byte[] body = readAll(exchange);
-        CloudEvent event;
+        EventMeshFrame frame;
         try {
-            event = EventFormatProvider.getInstance().resolveFormat(JsonFormat.CONTENT_TYPE).deserialize(body);
-        } catch (RuntimeException e) {
+            // Ingress: structured CloudEvents JSON bytes → internal EventMeshFrame.
+            // (#5299: runtime no longer touches io.cloudevents.CloudEvent directly on the ingress
+            // path; the protocol adaptor owns the conversion.)
+            frame = FrameAdaptors.get("cloudevents").toFrame(new ByteTransport(body));
+        } catch (RuntimeException | org.apache.eventmesh.protocol.api.exception.ProtocolHandleException e) {
             writeJson(exchange, 400, error("invalid CloudEvent: " + e.getMessage()));
             return;
         }
         // Security filter chain (§4.5): auth/acl/signature run before the event enters the pipeline.
+        // #5299 sub-PR B: filters now read directly from EventMeshFrame.attributes() — no more
+        // CE bridge. Tenant still comes from the CloudEvent extension ("emtenantid") which the
+        // cloudevents FrameAdaptor round-trips into frame attributes under the same key.
         if (filterChain != null) {
             String credential = exchange.getRequestHeaders().getFirst("Authorization");
-            String tenant = event.getExtension("emtenantid") != null ? event.getExtension("emtenantid").toString() : null;
+            String tenant = frame.attributes().get("emtenantid");
             org.apache.eventmesh.runtime.security.FilterContext ctx =
                 new org.apache.eventmesh.runtime.security.FilterContext(topic, null, tenant, credential,
                     exchange.getRemoteAddress().getAddress().getHostAddress());
-            org.apache.eventmesh.runtime.security.FilterVerdict verdict = filterChain.check(event, ctx);
+            org.apache.eventmesh.runtime.security.FilterVerdict verdict = filterChain.check(frame, ctx);
             if (!verdict.isAllowed()) {
                 writeJson(exchange, verdict.getRejectStatus(), error(verdict.getReason()));
                 return;
             }
         }
         try {
-            ingress.publish(topic, event).get(10, TimeUnit.SECONDS);
+            ingress.publish(topic, frame).get(10, TimeUnit.SECONDS);
             writeJson(exchange, 202, ack("accepted"));
         } catch (Exception e) {
             // §6.6: a RateLimitedException (per-topic token bucket exhausted) is a 429, not a 500 —
@@ -337,12 +345,14 @@ private void publishBatch(HttpExchange exchange) throws IOException {
                 writeJson(exchange, 400, error("expected a CloudEvent JSON array"));
                 return;
             }
-            java.util.List<CloudEvent> events = new java.util.ArrayList<>(node.size());
+            // Batch ingress: each element is a CloudEvents-JSON object → internal EventMeshFrame.
+            // (#5299)
+            java.util.List<EventMeshFrame> frames = new java.util.ArrayList<>(node.size());
             for (com.fasterxml.jackson.databind.JsonNode el : node) {
-                events.add(EventFormatProvider.getInstance().resolveFormat(JsonFormat.CONTENT_TYPE)
-                    .deserialize(mapper.writeValueAsBytes(el)));
+                frames.add(FrameAdaptors.get("cloudevents")
+                    .toFrame(new ByteTransport(mapper.writeValueAsBytes(el))));
             }
-            ingress.publishBatch(topic, events).get(30, TimeUnit.SECONDS);
+            ingress.publishBatchFrames(topic, frames).get(30, TimeUnit.SECONDS);
             writeJson(exchange, 202, ack("accepted"));
         } catch (Exception e) {
             if (isRateLimited(e)) {
@@ -484,13 +494,16 @@ private void request(HttpExchange exchange) throws IOException {
         String topic = param(exchange.getRequestURI(), "topic");
         long timeout = longParam(exchange.getRequestURI(), "timeoutMs", 30_000L);
         try {
-            CloudEvent event = EventFormatProvider.getInstance().resolveFormat(JsonFormat.CONTENT_TYPE)
-                .deserialize(readAll(exchange));
+            // Ingress: structured CloudEvents JSON body → internal EventMeshFrame. (#5299)
+            EventMeshFrame event = FrameAdaptors.get("cloudevents")
+                .toFrame(new ByteTransport(readAll(exchange)));
             if (!checkSecurity(exchange, topic, null)) {
                 return;
             }
-            CloudEvent reply = ingress.request(topic, event, timeout);
-            byte[] replyBytes = EventFormatProvider.getInstance().resolveFormat(JsonFormat.CONTENT_TYPE).serialize(reply);
+            // request-reply: send Frame in, get Frame back, then serialize to CloudEvents JSON
+            // for the response body via the egress adapter.
+            EventMeshFrame reply = ingress.requestFrame(topic, event, timeout);
+            byte[] replyBytes = FrameAdaptors.toCloudEventsJson(reply);
             exchange.getResponseHeaders().add("Content-Type", "application/cloudevents+json");
             exchange.sendResponseHeaders(200, replyBytes.length);
             try (OutputStream os = exchange.getResponseBody()) {
@@ -509,14 +522,15 @@ private void reply(HttpExchange exchange) throws IOException {
         try {
             JsonNode body = readJson(exchange);
             String corrId = text(body, "correlationId");
-            CloudEvent replyEvent = EventFormatProvider.getInstance().resolveFormat(JsonFormat.CONTENT_TYPE)
-                .deserialize(mapper.writeValueAsBytes(body.get("event")));
+            // Ingress: CloudEvents JSON → internal EventMeshFrame. (#5299)
+            EventMeshFrame replyEvent = FrameAdaptors.get("cloudevents")
+                .toFrame(new ByteTransport(mapper.writeValueAsBytes(body.get("event"))));
             // §17.6 reply routing (sticky model - no cross-instance forwarding).
             // Cross-instance reply forwarding is REMOVED with the forward path: the client posts
             // the reply to the instance it sent the request to (pinned via instanceUrl); a reply
             // landing on the wrong instance 404s (unknown correlationId) and the caller retries
             // on the correct instance.
-            writeJson(exchange, ingress.reply(corrId, replyEvent) ? 200 : 404, ack("ok"));
+            writeJson(exchange, ingress.replyFrame(corrId, replyEvent) ? 200 : 404, ack("ok"));
         } catch (Exception e) {
             writeJson(exchange, 500, error("reply error: " + e.getMessage()));
         }
@@ -615,9 +629,10 @@ private void litePublish(HttpExchange exchange) throws IOException {
             return;
         }
         try {
-            CloudEvent event = EventFormatProvider.getInstance().resolveFormat(JsonFormat.CONTENT_TYPE)
-                .deserialize(readAll(exchange));
-            ingress.publishLite(parent, lite, event).get(10, TimeUnit.SECONDS);
+            // Ingress: structured CloudEvents JSON body → internal EventMeshFrame. (#5299)
+            EventMeshFrame event = FrameAdaptors.get("cloudevents")
+                .toFrame(new ByteTransport(readAll(exchange)));
+            ingress.publishLiteFrame(parent, lite, event).get(10, TimeUnit.SECONDS);
             writeJson(exchange, 202, ack("accepted"));
         } catch (Exception e) {
             writeJson(exchange, 500, error("lite publish failed: " + e.getMessage()));
@@ -646,11 +661,12 @@ private void litePoll(HttpExchange exchange) throws IOException {
         try {
             int max = intParam(exchange.getRequestURI(), "max", 100);
             long timeoutMs = longParam(exchange.getRequestURI(), "timeoutMs", 1000L);
-            List<CloudEvent> events = ingress.pollLite(parent, lite, max, timeoutMs);
+            // Egress: drain EventMeshFrames from the LMQ, serialize each as CloudEvents JSON
+            // via the egress adapter. (#5299)
+            List<EventMeshFrame> events = ingress.pollLiteFrames(parent, lite, max, timeoutMs);
             com.fasterxml.jackson.databind.node.ArrayNode arr = mapper.createArrayNode();
-            for (CloudEvent e : events) {
-                arr.add(mapper.readTree(EventFormatProvider.getInstance()
-                    .resolveFormat(JsonFormat.CONTENT_TYPE).serialize(e)));
+            for (EventMeshFrame e : events) {
+                arr.add(mapper.readTree(FrameAdaptors.toCloudEventsJson(e)));
             }
             writeJson(exchange, 200, arr);
         } catch (NumberFormatException e) {