You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@eventmesh.apache.org by ch...@apache.org on 2026/08/31 11:29:34 UTC
(eventmesh) branch develop updated: fix(protocol): single protocol path — EventMeshFrame across HTTP ingress, filter chain and TCP egress (#5299) (#5319)
This is an automated email from the ASF dual-hosted git repository.
qqeasonchen pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/develop by this push:
new ba267195c fix(protocol): single protocol path — EventMeshFrame across HTTP ingress, filter chain and TCP egress (#5299) (#5319)
ba267195c is described below
commit ba267195c40317572d6c52d94b384ec358325b09
Author: Eason Chen <qq...@gmail.com>
AuthorDate: Mon Aug 31 19:29:27 2026 +0800
fix(protocol): single protocol path — EventMeshFrame across HTTP ingress, filter chain and TCP egress (#5299) (#5319)
* fix(protocol): route UniHttpServer ingress through FrameAdaptor SPI (#5299 Sub-PR A)
The HTTP ingress path now converts structured CloudEvents JSON bytes to
EventMeshFrame at the boundary via FrameAdaptor SPI, so UniIngressService
no longer imports io.cloudevents.CloudEvent on the HTTP code path.
New ingress methods (Frame-typed):
- publishBatchFrames(topic, List<EventMeshFrame>)
- publishLiteFrame(parent, lite, EventMeshFrame)
- pollLiteFrames(parent, lite, max, timeoutMs) -> List<EventMeshFrame>
- requestFrame(topic, EventMeshFrame, timeout) -> EventMeshFrame
- replyFrame(correlationId, EventMeshFrame)
The CloudEvent-typed overloads are preserved for binary compatibility
with the TCP bridge. Sub-PR B will migrate FilterChain; Sub-PR C will
migrate the TCP path.
Closes part of #5299.
* fix(protocol): migrate ingress filter chain to EventMeshFrame (#5299 Sub-PR B)
The ingress security pipeline (§4.5) — TokenAuthFilter / AclFilter /
SignatureVerifierFilter / FilterChain — now operates on the runtime's
internal wire format (org.apache.eventmesh.common.wire.EventMeshFrame)
instead of io.cloudevents.CloudEvent. This continues the #5299 single-
protocol-path migration: sub-PR A routed the HTTP ingress through the
FrameAdaptor SPI; sub-PR B extends the same boundary inward to the
filter chain so that auth/acl/signature now read directly from
frame.attributes().
Changes:
- IngressFilter: new check(EventMeshFrame, FilterContext) primary
method; the CloudEvent overload is now a default that bridges via
EventMeshFrame.fromCloudEvent(...) for backward compat.
- FilterChain: new check(EventMeshFrame, FilterContext) overload;
CloudEvent variant is deprecated and bridges via fromCloudEvent.
- AclFilter: implement check(EventMeshFrame, ...) — also rejects
non-EVENT frames (e.g. STREAM_REQ / STREAM_CHUNK) since the
ingress security pipeline is event-shaped only.
- TokenAuthFilter: implement check(EventMeshFrame, ...) — credential
still comes from FilterContext (HTTP Authorization header), so
the frame body is unused for this stage.
- SignatureVerifierFilter: implement check(EventMeshFrame, ...) —
signature travels in frame.attributes() under the same key
(emsignature) the legacy CloudEvent extension used, so signed
CloudEvents round-trip transparently through the cloudevents
FrameAdaptor.
- UniHttpServer: drop the temporary frame.toCloudEvent() bridge in
the publish() filter call; thread tenant directly from
frame.attributes(). Also drop the synthetic CloudEvent stub used
for pre-publish security check — replaced with a minimal
EventMeshFrame.event(emptyMap, []).
- New test class: EventMeshFrameFilterTest (6 tests) exercising
the EventMeshFrame-typed path parallel to the existing
SecurityFilterTest which still covers the CloudEvent bridge.
Backward compat: existing custom filters implementing
IngressFilter.check(CloudEvent, ...) keep working because the
interface now provides a default implementation. They will compile
unchanged and continue to be invoked via the deprecated chain
overload until sub-PR C migrates the TCP path and we can drop the
bridge entirely.
Refs #5299.
* fix(protocol): complete #5299 single protocol path (sub-PR B fix + C + D)
Sub-PR B fix:
- restore the missing io.cloudevents.CloudEvent import in IngressFilter (broke compileJava)
- move EventMeshFrame into the correct checkstyle ImportOrder group
(style/checkStyle.xml puts org.apache.eventmesh first; maxWarnings=0 so it fails CI)
- update SecurityFilterTest for the frame-based SignatureVerifierFilter.canonical(...)
Sub-PR C: migrate the TCP egress path to EventMeshFrame
- TcpFrameCodec.encodePush / TcpPushChannel.deliver now take an EventMeshFrame directly,
dropping the frame -> CloudEvent -> wire round trip
- delete the dead CloudEvent-era egress SPI: CloudEventToPackageBody, MeshEventToPackageBody
- UniTcpServer drops the now-unused bodyMapper constructor parameter
- TCP ingress was already frame-native (MeshMessagePackageRouter / TcpRequest / NettyTcpPushChannel)
Sub-PR D: document the #5299 acceptance matrix and protocol status labels
(docs/eventmesh-uni-architecture-redesign.md section 19.6)
Refs #5299.
---
docs/eventmesh-uni-architecture-redesign.md | 101 +-
.../eventmesh/runtime/http/UniHttpServer.java | 80 +-
.../runtime/ingress/UniIngressService.java | 1891 ++++++++++----------
.../eventmesh/runtime/security/AclFilter.java | 16 +-
.../eventmesh/runtime/security/FilterChain.java | 20 +-
.../eventmesh/runtime/security/IngressFilter.java | 23 +-
.../runtime/security/SignatureVerifierFilter.java | 31 +-
.../runtime/security/TokenAuthFilter.java | 6 +-
.../transport/tcp/CloudEventToPackageBody.java | 31 -
.../transport/tcp/MeshEventToPackageBody.java | 50 -
.../runtime/transport/tcp/TcpFrameCodec.java | 17 +-
.../runtime/transport/tcp/TcpPushChannel.java | 27 +-
.../runtime/transport/tcp/UniTcpServer.java | 15 +-
.../runtime/it/LegacyTcpClientIntegrationTest.java | 4 +-
.../it/LegacyTcpClusterBrokerIntegrationTest.java | 3 +-
.../runtime/security/EventMeshFrameFilterTest.java | 145 ++
.../runtime/security/SecurityFilterTest.java | 4 +-
.../tcp/MeshMessagePackageRouterTest.java | 18 +-
.../transport/tcp/TcpCompatibilityBridgeTest.java | 4 +-
.../runtime/transport/tcp/UniTcpServerTest.java | 11 +-
20 files changed, 1396 insertions(+), 1101 deletions(-)
diff --git a/docs/eventmesh-uni-architecture-redesign.md b/docs/eventmesh-uni-architecture-redesign.md
index f2cdcea75..25460b09a 100644
--- a/docs/eventmesh-uni-architecture-redesign.md
+++ b/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）收口后移除。
---
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/http/UniHttpServer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/http/UniHttpServer.java
index e67bdd540..1b813a4ab 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/http/UniHttpServer.java
+++ b/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.Map;
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 @@ public class UniHttpServer {
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 @@ public class UniHttpServer {
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 @@ public class UniHttpServer {
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 @@ public class UniHttpServer {
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 @@ public class UniHttpServer {
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 @@ public class UniHttpServer {
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 @@ public class UniHttpServer {
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) {
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/ingress/UniIngressService.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/ingress/UniIngressService.java
index 6d47a7d25..f6e617abb 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/ingress/UniIngressService.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/ingress/UniIngressService.java
@@ -1,903 +1,988 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.eventmesh.runtime.ingress;
-
-import org.apache.eventmesh.api.SendCallback;
-import org.apache.eventmesh.api.SendResult;
-import org.apache.eventmesh.api.storage.MeshStoragePlugin;
-import org.apache.eventmesh.runtime.delivery.DeadLetterSink;
-import org.apache.eventmesh.runtime.delivery.PushChannel;
-import org.apache.eventmesh.runtime.delivery.ReliableDispatcher;
-import org.apache.eventmesh.runtime.metrics.UniMetrics;
-import org.apache.eventmesh.runtime.metrics.UniTrace;
-import org.apache.eventmesh.runtime.offset.OffsetStore;
-import org.apache.eventmesh.runtime.push.BufferedEvent;
-import org.apache.eventmesh.runtime.push.LongPollingChannel;
-import org.apache.eventmesh.runtime.push.PushService;
-import org.apache.eventmesh.runtime.ratelimit.RateLimitedException;
-import org.apache.eventmesh.runtime.ratelimit.TokenBucketRateLimiter;
-import org.apache.eventmesh.runtime.subscription.CloudEventFilter;
-import org.apache.eventmesh.runtime.subscription.DistributionMode;
-import org.apache.eventmesh.runtime.subscription.Subscription;
-import org.apache.eventmesh.runtime.subscription.SubscriptionManager;
-
-import java.util.List;
-import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
-import java.util.concurrent.atomic.AtomicLong;
-
-import io.cloudevents.CloudEvent;
-import io.cloudevents.core.builder.CloudEventBuilder;
-
-import lombok.extern.slf4j.Slf4j;
-
-/**
- * Facade that wires the uni-architecture layers into the end-to-end CloudEvents-over-MQ
- * flow (§6):
- * <pre>
- * publish ─▶ MeshStoragePlugin.send
- * pullLoop ─▶ MeshStoragePlugin.poll ─▶ SubscriptionManager.targetsFor ─▶ ReliableDispatcher.deliver ─▶ PushService
- * poll ◀─ client long-polls PushService
- * ack ─▶ PushService.ack ─▶ ReliableDispatcher.ack ─▶ OffsetStore (offset advances only on ACK)
- * </pre>
- *
- * <p>This is the orchestration core; the actual HTTP endpoint wiring (§6 UniIngressHandler)
- * delegates here. Phase-1 thin storage adapter means {@code partition} is {@code -1} and the offset
- * is a per-topic monotonic logical counter until the native storage reimplementation supplies real
- * partition/offset (Phase 1 step 2).</p>
- */
-@Slf4j
-public class UniIngressService {
-
- private final MeshStoragePlugin storage;
- private final OffsetStore offsetStore;
- private final SubscriptionManager subscriptionManager;
- private final ReliableDispatcher dispatcher;
- private final PushService pushService;
- private volatile org.apache.eventmesh.runtime.cluster.ClusterCoordinator cluster;
- private volatile org.apache.eventmesh.runtime.cluster.PartitionOwnership partitionOwnership;
- /** Self-collected load metrics for session-distribution load balancing (§3). Null until wired. */
- private volatile LoadMeter loadMeter;
-
- /** Per-topic poll stats for the {@code poll_idle_ratio} gauge (§13.5.1). */
- private final ConcurrentHashMap<String, Long> pollCount = new ConcurrentHashMap<>();
- private final ConcurrentHashMap<String, Long> pollEmptyCount = new ConcurrentHashMap<>();
-
- /** Connector offset store (remote side) — String key → String offset value (§8.9). */
- private final java.util.concurrent.ConcurrentHashMap<String, String> connectorOffsets = new java.util.concurrent.ConcurrentHashMap<>();
- private final UniMetrics metrics;
-
- private final ConcurrentHashMap<String, PushChannel> channels = new ConcurrentHashMap<>();
- private final ConcurrentHashMap<String, AtomicLong> topicOffsetSeq = new ConcurrentHashMap<>();
- private final ConcurrentHashMap<String, CompletableFuture<CloudEvent>> pendingRequests = new ConcurrentHashMap<>();
- private final AtomicLong requestSeq = new AtomicLong();
- private final ConcurrentHashMap<String, TokenBucketRateLimiter> topicLimiters = new ConcurrentHashMap<>();
-
- /**
- * CloudEvents extension carrying the request's correlation id (§17).
- *
- * <p>Named without hyphens because the CloudEvents spec restricts extension attribute names to
- * lower-case ASCII letters and digits — the redesign doc's {@code x-em-correlation-id} spelling
- * is rejected by the SDK's name validation.</p>
- */
- public static final String EXT_CORRELATION_ID = "emcorrelationid";
-
- public UniIngressService(MeshStoragePlugin storage, OffsetStore offsetStore) {
- this(storage, offsetStore, new SubscriptionManager(), new PushService(),
- ReliableDispatcher.DEFAULT_ACK_TIMEOUT_MS, ReliableDispatcher.DEFAULT_MAX_ATTEMPTS,
- System::currentTimeMillis);
- }
-
- /**
- * Test-friendly constructor with an injectable clock and retry parameters.
- */
- public UniIngressService(MeshStoragePlugin storage, OffsetStore offsetStore,
- SubscriptionManager subscriptionManager, PushService pushService,
- long ackTimeoutMs, int maxAttempts, java.util.function.LongSupplier clock) {
- this.storage = storage;
- this.offsetStore = offsetStore;
- this.subscriptionManager = subscriptionManager;
- this.pushService = pushService;
- this.metrics = new UniMetrics();
- this.dispatcher = new ReliableDispatcher(ackTimeoutMs, maxAttempts, clock, offsetStore, deadLetterSink(),
- metrics, ReliableDispatcher.DEFAULT_JITTER_RATIO);
- // Sub-PR B: re-ACK any in-flight deliveries from a previous JVM so they do not become
- // orphans after a crash. Safe on first start (store is empty) and idempotent.
- this.dispatcher.recover();
- }
-
- // ---- connector offset (remote side, §8.9) ----
-
- public String getConnectorOffset(String connectorId) {
- return connectorOffsets.get(connectorId);
- }
-
- public void putConnectorOffset(String connectorId, String offset) {
- connectorOffsets.put(connectorId, offset);
- }
-
- /**
- * Publish a CloudEvent to {@code topic} (persisted to MQ). Completes when the storage plugin
- * acknowledges the write.
- */
- public CompletableFuture<Void> publish(String topic, CloudEvent event) {
- if (loadMeter != null && event.getData() != null) {
- loadMeter.recordInflow(event.getData().toBytes().length);
- }
- CompletableFuture<Void> future = new CompletableFuture<>();
- TokenBucketRateLimiter limiter = topicLimiters.get(topic);
- if (limiter != null && !limiter.tryAcquire()) {
- metrics.incRateLimited();
- future.completeExceptionally(new RateLimitedException(topic));
- return future;
- }
- try {
- // Ingress boundary: CloudEvent (the public/external format) → EventMeshFrame (the internal
- // wire unit). From here on the event is a Frame internally; egress converts back to the
- // client's protocol (CloudEvents / MeshMessage) at the delivery boundary.
- org.apache.eventmesh.common.wire.EventMeshFrame frame =
- org.apache.eventmesh.common.wire.EventMeshFrame.fromCloudEvent(event);
- storage.send(topic, frame, new SendCallback() {
- @Override
- public void onSuccess(SendResult sendResult) {
- metrics.incPublish();
- UniTrace.end(UniTrace.startPublish(topic, event));
- future.complete(null);
- }
-
- @Override
- public void onException(org.apache.eventmesh.api.exception.OnExceptionContext context) {
- metrics.incPublishFailed();
- future.completeExceptionally(context.getException());
- }
- });
- } catch (Exception e) {
- metrics.incPublishFailed();
- future.completeExceptionally(e);
- }
- return future;
- }
-
- /**
- * Publish an already-internal {@link org.apache.eventmesh.common.wire.EventMeshFrame} (no
- * CloudEvent→Frame boundary conversion — the caller built the frame directly, e.g. the legacy
- * MeshMessage TCP path via {@code MeshMessageFrameCodec}). Completes when storage acks the write.
- */
- public CompletableFuture<Void> publish(String topic, org.apache.eventmesh.common.wire.EventMeshFrame frame) {
- if (loadMeter != null && frame.data() != null) {
- loadMeter.recordInflow(frame.data().length);
- }
- CompletableFuture<Void> future = new CompletableFuture<>();
- TokenBucketRateLimiter limiter = topicLimiters.get(topic);
- if (limiter != null && !limiter.tryAcquire()) {
- metrics.incRateLimited();
- future.completeExceptionally(new RateLimitedException(topic));
- return future;
- }
- try {
- storage.send(topic, frame, new SendCallback() {
- @Override
- public void onSuccess(SendResult sendResult) {
- metrics.incPublish();
- future.complete(null);
- }
-
- @Override
- public void onException(org.apache.eventmesh.api.exception.OnExceptionContext context) {
- metrics.incPublishFailed();
- future.completeExceptionally(context.getException());
- }
- });
- } catch (Exception e) {
- metrics.incPublishFailed();
- future.completeExceptionally(e);
- }
- return future;
- }
-
- /**
- * Publish a batch of CloudEvents to {@code topic} (§13.7.3). Completes when all are persisted.
- * The storage plugin's own batching (e.g. Kafka producer accumulator) amortises the per-event
- * RTT; this layer fans out the per-event futures and joins them.
- */
- public CompletableFuture<Void> publishBatch(String topic, java.util.List<CloudEvent> events) {
- if (events == null || events.isEmpty()) {
- return CompletableFuture.completedFuture(null);
- }
- CompletableFuture<?>[] futures = events.stream()
- .map(e -> publish(topic, e))
- .toArray(CompletableFuture[]::new);
- return CompletableFuture.allOf(futures);
- }
-
- /**
- * Register a custom push channel for a client (e.g. a WebHook URL for legacy HTTP webhook-push
- * subscribers, or a {@code TcpPushChannel} for legacy TCP clients). Overrides the default
- * long-polling channel for this {@code clientId} on subsequent dispatches.
- */
- public void registerChannel(String clientId, PushChannel channel) {
- pushService.register(clientId);
- channels.put(clientId, channel);
- }
-
- /**
- * Configure a per-topic rate limit (§6.6). Subsequent publishes above {@code permitsPerSecond}
- * (burst {@code capacity}) fail the publish future with {@link RateLimitedException}.
- */
- public void setTopicRateLimit(String topic, long capacity, double permitsPerSecond) {
- topicLimiters.put(topic, new TokenBucketRateLimiter(capacity, permitsPerSecond));
- }
-
- /**
- * Register a subscription. The subscriber retrieves events via {@link #poll}.
- *
- * @return the subscription id
- */
- public String subscribe(String topic, String clientId, DistributionMode mode, CloudEventFilter filter) {
- pushService.register(clientId);
- String subId = subscriptionManager.subscribe(topic, clientId, mode, filter, event -> {
- // No-op: the reliability layer (ReliableDispatcher) owns delivery, not this fire-and-
- // forget callback. Kept to satisfy the SubscriptionManager handler contract.
- });
- // §13.2: when clustered, also register cluster-wide so other instances can route events
- // for this subscriber here (via ClusterCoordinator → HttpForwarder /internal/forward).
- // Without this the HTTP /events/subscribe path is local-only and a publish on a peer
- // never reaches this subscriber.
- if (cluster != null) {
- cluster.subscribe(topic, clientId, mode, null);
- }
- return subId;
- }
-
- /**
- * Remove a subscription.
- */
- public boolean unsubscribe(String subscriptionId) {
- // Resolve topic + clientId before removing so a clustered subscription can be deregistered
- // cluster-wide (the coordinator's unsubscribe is keyed by topic+clientId, not subId).
- org.apache.eventmesh.runtime.subscription.Subscription sub = subscriptionManager.getSubscription(subscriptionId);
- boolean removed = subscriptionManager.unsubscribe(subscriptionId);
- if (removed && cluster != null && sub != null) {
- cluster.unsubscribe(sub.getTopic(), sub.getClientId());
- }
- return removed;
- }
-
- /** Remove one client's subscription to one topic (HTTP /events/unsubscribe with {clientId, topic}). */
- public boolean unsubscribe(String topic, String clientId) {
- boolean removed = subscriptionManager.unsubscribe(topic, clientId);
- if (removed && cluster != null) {
- cluster.unsubscribe(topic, clientId);
- }
- return removed;
- }
-
- /** Remove ALL subscriptions for a client (HTTP /events/unsubscribe with {clientId} only).
- * Propagates cluster-wide + frees the PushService buffer. */
- public int unsubscribeByClient(String clientId) {
- java.util.Set<String> topics = subscriptionManager.topicsForClient(clientId);
- int removed = subscriptionManager.unsubscribeByClient(clientId);
- if (cluster != null) {
- for (String topic : topics) {
- cluster.unsubscribe(topic, clientId);
- }
- }
- pushService.removeClient(clientId);
- return removed;
- }
-
- /**
- * Pump: pull a batch from storage, route to each target subscriber via the reliability layer.
- *
- * @return number of events pulled
- */
- public int pullAndDispatch(String topic, int maxEvents, long timeoutMs) {
- pollCount.merge(topic, 1L, Long::sum);
- // Multi-instance (§13.2.3): poll only the partitions this instance owns. Single-instance
- // (no ownership) or unknown partition count → poll the whole topic (partition -1).
- java.util.List<Integer> owned = partitionOwnership == null ? null : partitionOwnership.ownedPartitions(topic);
- int total;
- if (owned == null) {
- total = pullAndDispatchPartition(topic, -1, maxEvents, timeoutMs);
- } else if (owned.isEmpty()) {
- total = 0; // owns none -> do not poll (avoids duplicate with the real owners)
- } else {
- total = 0;
- for (int p : owned) {
- total += pullAndDispatchPartition(topic, p, maxEvents, timeoutMs);
- }
- }
- if (total == 0) {
- pollEmptyCount.merge(topic, 1L, Long::sum);
- }
- return total;
- }
-
- private int pullAndDispatchPartition(String topic, int partition, int maxEvents, long timeoutMs) {
- // Pull EventMeshFrames (internal wire); dispatch each as a Frame through the internal
- // pipeline (filter/dispatcher/egress all carry Frame now; egress converts to the client's
- // protocol at the wire boundary).
- List<org.apache.eventmesh.common.wire.EventMeshFrame> frames = storage.poll(topic, partition, -1, maxEvents, timeoutMs);
- if (frames == null || frames.isEmpty()) {
- return 0;
- }
- long start = System.nanoTime();
- for (org.apache.eventmesh.common.wire.EventMeshFrame f : frames) {
- if (isExpired(f)) {
- // §13.3.4 TTL: drop expired events instead of dispatching.
- log.debug("dropping expired event {} on topic {} (emttl elapsed)", f.attributes().get("id"), topic);
- continue;
- }
- io.opentelemetry.api.trace.Span dispatchSpan = UniTrace.startDispatch(topic, f);
- if (cluster != null) {
- // Multi-instance: route via the cluster coordinator (local vs cross-instance forward).
- cluster.dispatch(topic, f);
- } else {
- // develop's OffsetExtensions (CE-extension-carried MQ offset) is superseded by the
- // Frame architecture: the POP check key rides in frame attributes (empopck) and the
- // deferred broker ACK fires on client ACK — same at-least-once goal, Frame-native.
- long offset = nextOffset(topic);
- // P2 fix: if the frame carries a POP check key (RocketMQ 5.x deferred ACK), build a
- // callback that ACKs the broker on client ACK (restoring at-least-once).
- String popCk = f.attributes().get("empopck");
- Runnable mqAck = (popCk != null) ? () -> storage.ackPulledMessage(topic, popCk) : null;
- for (Subscription target : subscriptionManager.targetsFor(topic, f)) {
- dispatcher.deliver(topic, partition, offset, f, target.getClientId(),
- channelFor(target.getClientId()), mqAck);
- }
- }
- UniTrace.end(dispatchSpan);
- }
- metrics.addDispatchLatencyNanos(System.nanoTime() - start);
- metrics.incDispatched(frames.size());
- return frames.size();
- }
-
- /**
- * TTL expiry check (§13.3.4): an event with an {@code emttl} attribute (ms) and a {@code time}
- * is expired when {@code now > time + emttl}. Events without either field never expire. Reads the
- * attributes off the internal EventMeshFrame (emttl/time are preserved in its KV section).
- */
- private boolean isExpired(org.apache.eventmesh.common.wire.EventMeshFrame event) {
- String ttl = event.attributes().get("emttl");
- String time = event.attributes().get("time");
- if (ttl == null || time == null) {
- return false;
- }
- try {
- long ttlMs = Long.parseLong(ttl);
- long eventTime = java.time.OffsetDateTime.parse(time).toInstant().toEpochMilli();
- return System.currentTimeMillis() > eventTime + ttlMs;
- } catch (NumberFormatException | java.time.format.DateTimeParseException e) {
- return false;
- }
- }
-
- /**
- * Local delivery for one subscriber — handed to the reliability layer + the subscriber's push
- * channel. Exposed so a {@link org.apache.eventmesh.runtime.cluster.ClusterCoordinator} can route
- * same-instance targets here while forwarding remote ones.
- */
- public boolean deliverLocal(String topic, String clientId, org.apache.eventmesh.common.wire.EventMeshFrame event) {
- long offset = nextOffset(topic);
- dispatcher.deliver(topic, -1, offset, event, clientId, channelFor(clientId));
- return true;
- }
-
- /**
- * Enable multi-instance coordination: when set, {@link #pullAndDispatch} routes each event
- * through the cluster coordinator (local targets via {@link #deliverLocal}, remote via forward).
- */
- public void withCluster(org.apache.eventmesh.runtime.cluster.ClusterCoordinator cluster) {
- this.cluster = cluster;
- }
-
- /**
- * Topics this instance should pull and partition-assign: local active topics UNION cluster-wide
- * topics (topics with a remote subscriber discovered via the Meta watch). Without the cluster
- * half, an instance with no local subscriber for a topic never pulls it, so messages on its
- * partitions can't be forwarded to the remote subscriber (multi-instance message loss). Returns
- * local-only when not clustered. Both {@code UniRuntime.pullLoop} and {@code PartitionOwnership}'s
- * topic source use this so the pull set and the assignment set stay consistent (otherwise an
- * unassigned cluster topic would degrade to poll-all and duplicate).
- */
- public java.util.Set<String> activeTopicsClustered() {
- java.util.Set<String> topics = new java.util.HashSet<>(getSubscriptionManager().activeTopics());
- if (cluster != null) {
- topics.addAll(cluster.subscriptionTopics());
- }
- return topics;
- }
-
- // ===================== Lite Topic (RIP-83, 5.x-only) =====================
- // Lite topic ops are exposed only when the storage plugin implements LiteTopicCapable; otherwise
- // they fail fast (UnsupportedOperationException). The HTTP layer (/events/lite/*) delegates here.
-
- /**
- * Publish one CloudEvent to a lite topic (parentTopic, liteTopic). The storage plugin routes it
- * into the lite topic's LMQ. Requires a {@link org.apache.eventmesh.api.storage.LiteTopicCapable}
- * storage (the 5.x plugin); 4.x/kafka/standalone storages throw.
- */
- public CompletableFuture<Void> publishLite(String parentTopic, String liteTopic, CloudEvent event) {
- CompletableFuture<Void> future = new CompletableFuture<>();
- if (!(storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable)) {
- future.completeExceptionally(new UnsupportedOperationException("storage does not support lite topic"));
- return future;
- }
- try {
- // Boundary: CloudEvent → EventMeshFrame (internal wire unit).
- org.apache.eventmesh.common.wire.EventMeshFrame frame =
- org.apache.eventmesh.common.wire.EventMeshFrame.fromCloudEvent(event);
- ((org.apache.eventmesh.api.storage.LiteTopicCapable) storage).sendLite(parentTopic, liteTopic, frame,
- new SendCallback() {
- @Override
- public void onSuccess(SendResult sendResult) {
- future.complete(null);
- }
-
- @Override
- public void onException(org.apache.eventmesh.api.exception.OnExceptionContext context) {
- future.completeExceptionally(context.getException());
- }
- });
- } catch (Exception e) {
- future.completeExceptionally(e);
- }
- return future;
- }
-
- /**
- * Pull a batch of CloudEvents from a lite topic (direct pull from the LMQ; no deliveryId / no
- * EventMesh reliability layer — the lite consumer self-manages offset in the storage plugin).
- * Frames pulled from the LMQ are decoded back to CloudEvents at this boundary. Empty list if lite
- * is not supported.
- */
- public List<CloudEvent> pollLite(String parentTopic, String liteTopic, int maxEvents, long timeoutMs) {
- if (!(storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable)) {
- return java.util.Collections.emptyList();
- }
- java.util.List<CloudEvent> out = new java.util.ArrayList<>();
- for (org.apache.eventmesh.common.wire.EventMeshFrame f
- : ((org.apache.eventmesh.api.storage.LiteTopicCapable) storage).pullLite(parentTopic, liteTopic, maxEvents, timeoutMs)) {
- out.add(f.toCloudEvent());
- }
- return out;
- }
-
- /**
- * Publish a pre-encoded EventMeshFrame byte payload to a lite topic (the internal streaming wire
- * path — SessionRouter publishes frame bytes). The payload IS an encoded EventMeshFrame; decode
- * to the Frame object the SPI now expects.
- */
- public CompletableFuture<Void> publishLiteBytes(String parentTopic, String liteTopic, byte[] payload) {
- if (loadMeter != null) {
- loadMeter.recordInflow(payload.length);
- }
- CompletableFuture<Void> future = new CompletableFuture<>();
- if (!(storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable)) {
- future.completeExceptionally(new UnsupportedOperationException("storage does not support lite topic"));
- return future;
- }
- try {
- org.apache.eventmesh.common.wire.EventMeshFrame frame = org.apache.eventmesh.common.wire.EventMeshFrame.decode(payload);
- ((org.apache.eventmesh.api.storage.LiteTopicCapable) storage).sendLite(parentTopic, liteTopic, frame,
- new SendCallback() {
- @Override
- public void onSuccess(SendResult sendResult) {
- future.complete(null);
- }
-
- @Override
- public void onException(org.apache.eventmesh.api.exception.OnExceptionContext context) {
- future.completeExceptionally(context.getException());
- }
- });
- } catch (Exception e) {
- future.completeExceptionally(e);
- }
- return future;
- }
-
- /**
- * Pull a batch of pre-encoded EventMeshFrame byte payloads from a lite topic (the byte counterpart
- * of {@link #pollLite}, for the internal streaming wire). Each entry is an encoded frame.
- * Empty list if lite is not supported.
- */
- public List<byte[]> pollLiteBytes(String parentTopic, String liteTopic, int maxEvents, long timeoutMs) {
- if (!(storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable)) {
- return java.util.Collections.emptyList();
- }
- java.util.List<byte[]> out = new java.util.ArrayList<>();
- for (org.apache.eventmesh.common.wire.EventMeshFrame f
- : ((org.apache.eventmesh.api.storage.LiteTopicCapable) storage).pullLite(parentTopic, liteTopic, maxEvents, timeoutMs)) {
- out.add(f.encode());
- }
- return out;
- }
-
- /**
- * Ensure {@code parentTopic} is lite-capable and declare {@code liteTopic} under it. Throws if the
- * storage does not support lite.
- */
- public void createLiteTopic(String parentTopic, String liteTopic) throws Exception {
- if (!(storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable)) {
- throw new UnsupportedOperationException("storage does not support lite topic");
- }
- ((org.apache.eventmesh.api.storage.LiteTopicCapable) storage).createLiteTopic(parentTopic, liteTopic);
- }
-
- /**
- * Ensure {@code parentTopic} is lite-capable with the given {@code queueCount} and declare
- * {@code liteTopic} under it. Use {@code queueCount=1} for strict in-order delivery (e.g. a
- * streaming-call response channel). Throws if the storage does not support lite.
- */
- public void createLiteTopic(String parentTopic, String liteTopic, int queueCount) throws Exception {
- if (!(storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable)) {
- throw new UnsupportedOperationException("storage does not support lite topic");
- }
- ((org.apache.eventmesh.api.storage.LiteTopicCapable) storage).createLiteTopic(parentTopic, liteTopic, queueCount);
- }
-
- /**
- * @return true iff the storage plugin implements {@link org.apache.eventmesh.api.storage.LiteTopicCapable}.
- */
- public boolean isLiteCapable() {
- return storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable;
- }
-
- /**
- * Enable multi-instance partition ownership (§13.2.3): when set, {@link #pullAndDispatch} polls
- * only this instance's owned partitions instead of the whole topic.
- */
- public void withPartitionOwnership(org.apache.eventmesh.runtime.cluster.PartitionOwnership ownership) {
- this.partitionOwnership = ownership;
- }
-
- /** Wire the load meter; ingress/egress points call its record* methods. */
- public void withLoadMeter(LoadMeter loadMeter) {
- this.loadMeter = loadMeter;
- }
-
- /**
- * @return the wired load meter, or null if not configured (single-instance / tests).
- */
- public LoadMeter loadMeter() {
- return loadMeter;
- }
-
- /**
- * Subscriber long-polls its buffered deliveries.
- */
- public List<BufferedEvent> poll(String clientId, int maxEvents, long timeoutMs) {
- return pushService.poll(clientId, maxEvents, timeoutMs);
- }
-
- /**
- * Subscriber acknowledges a delivery — the offset advances only on ACK (at-least-once).
- */
- public boolean ack(String deliveryId) {
- return pushService.ack(deliveryId);
- }
-
- /**
- * Drive retry / DLQ. Call periodically from a scheduler.
- */
- public int dispatcherTick() {
- return dispatcher.tick();
- }
-
- /**
- * Operational metrics counters (publish/dispatch/ack/retry/DLQ). The dispatcher and ingress
- * share one instance.
- */
- public UniMetrics getMetrics() {
- return metrics;
- }
-
- // Accessors for the admin facade (Phase 7.5). The service is the single owner of these
- // collaborators; exposing them avoids reconstructing them out of band.
- public SubscriptionManager getSubscriptionManager() {
- return subscriptionManager;
- }
-
- public PushService getPushService() {
- return pushService;
- }
-
- public ReliableDispatcher getDispatcher() {
- return dispatcher;
- }
-
- /**
- * @return the multi-instance partition ownership (null when clustering is disabled).
- */
- public org.apache.eventmesh.runtime.cluster.PartitionOwnership getPartitionOwnership() {
- return partitionOwnership;
- }
-
- /**
- * Stale-poll cleanup (§13.6.5): evict clients that haven't polled within {@code thresholdMs} —
- * drops their subscriptions and push buffer so zombie subscriptions don't leak.
- *
- * @return number of subscriptions removed
- */
- public int cleanupStaleClients(long thresholdMs) {
- int removed = 0;
- for (String cid : pushService.getStaleClientIds(thresholdMs)) {
- removed += subscriptionManager.unsubscribeByClient(cid);
- pushService.removeClient(cid);
- log.info("evicted stale client {} (no poll within {}ms)", cid, thresholdMs);
- }
- return removed;
- }
-
- /**
- * Register OTel observable gauges backed by live runtime state (§13.5.1 gauges with *).
- * Call once at boot. Gauges read on each OTel collection cycle.
- */
- public void registerRuntimeGauges() {
- metrics.registerGauge("eventmesh_pending_queue_size", "total buffered events across all clients",
- () -> {
- long sum = 0;
- for (String cid : pushService.clientIds()) {
- sum += pushService.pending(cid);
- }
- return sum;
- });
- metrics.registerGauge("eventmesh_slow_consumer_count", "clients in SLOW or STALLED state",
- pushService::slowConsumerCount);
- metrics.registerGauge("eventmesh_active_topics", "topics with active subscribers",
- () -> subscriptionManager.activeTopics().size());
- metrics.registerGauge("eventmesh_active_subscribers", "active subscriptions across all topics",
- () -> {
- int sum = 0;
- for (String t : subscriptionManager.activeTopics()) {
- sum += subscriptionManager.activeSubscriptions(t).size();
- }
- return sum;
- });
-
- // Labelled gauges (§13.5.1) — emit one reading per topic / partition.
- metrics.registerLabelledGauge("eventmesh_poll_idle_ratio",
- "fraction of poll cycles returning no events (per-mille, per topic)",
- () -> {
- java.util.List<UniMetrics.LabelledLong> out = new java.util.ArrayList<>();
- for (String t : pollCount.keySet()) {
- long total = pollCount.getOrDefault(t, 0L);
- long empty = pollEmptyCount.getOrDefault(t, 0L);
- long perMille = total == 0 ? 0 : Math.round((double) empty / total * 1000);
- out.add(new UniMetrics.LabelledLong(
- io.opentelemetry.api.common.Attributes.of(io.opentelemetry.api.common.AttributeKey.stringKey("topic"), t),
- perMille));
- }
- return out;
- });
-
- metrics.registerLabelledGauge("eventmesh_partition_owner",
- "1 for each partition this instance owns (per topic/partition)",
- () -> {
- java.util.List<UniMetrics.LabelledLong> out = new java.util.ArrayList<>();
- if (partitionOwnership != null) {
- for (String t : subscriptionManager.activeTopics()) {
- java.util.List<Integer> owned = partitionOwnership.ownedPartitions(t);
- if (owned == null) {
- continue;
- }
- for (int p : owned) {
- out.add(new UniMetrics.LabelledLong(
- io.opentelemetry.api.common.Attributes.of(
- io.opentelemetry.api.common.AttributeKey.stringKey("topic"), t,
- io.opentelemetry.api.common.AttributeKey.longKey("partition"), (long) p,
- io.opentelemetry.api.common.AttributeKey.stringKey("instance"), "self"),
- 1L));
- }
- }
- }
- return out;
- });
-
- metrics.registerLabelledGauge("eventmesh_offset_lag",
- "MQ end offset - max ACK offset (per topic/partition) — total consumer lag",
- () -> {
- java.util.List<UniMetrics.LabelledLong> out = new java.util.ArrayList<>();
- if (partitionOwnership == null) {
- return out;
- }
- for (String t : subscriptionManager.activeTopics()) {
- java.util.List<Integer> owned = partitionOwnership.ownedPartitions(t);
- if (owned == null) {
- continue;
- }
- // Max ACK offset per partition across all clients (key = clientId#partition).
- // Exclude the reserved __mqcursor__ key (MQ physical offset — different
- // magnitude from the logical sequence numbers, mixing them corrupts the gauge).
- java.util.Map<Integer, Long> maxAckByPart = new java.util.HashMap<>();
- String reservedPrefix =
- org.apache.eventmesh.runtime.delivery.ReliableDispatcher.MQ_CURSOR_CLIENT + "#";
- for (java.util.Map.Entry<String, Long> e : offsetStore.readAllOffsets(t).entrySet()) {
- if (e.getKey().startsWith(reservedPrefix)) {
- continue;
- }
- int sep = e.getKey().lastIndexOf('#');
- if (sep > 0) {
- try {
- int p = Integer.parseInt(e.getKey().substring(sep + 1));
- maxAckByPart.merge(p, e.getValue(), Math::max);
- } catch (NumberFormatException expected) {
- // offset key suffix is not a numeric partition; skip
- }
- }
- }
- for (int p : owned) {
- long end = storage.endOffset(t, p);
- long ack = maxAckByPart.getOrDefault(p, -1L);
- if (end >= 0 && ack >= 0) {
- out.add(new UniMetrics.LabelledLong(
- io.opentelemetry.api.common.Attributes.of(
- io.opentelemetry.api.common.AttributeKey.stringKey("topic"), t,
- io.opentelemetry.api.common.AttributeKey.longKey("partition"), (long) p),
- Math.max(0, end - ack)));
- }
- }
- }
- return out;
- });
-
- metrics.registerLabelledGauge("eventmesh_push_ack_lag",
- "max push offset - max ACK offset (per topic/partition) — in-flight deliveries",
- () -> {
- java.util.List<UniMetrics.LabelledLong> out = new java.util.ArrayList<>();
- for (String t : subscriptionManager.activeTopics()) {
- // Frame architecture: the MQ physical cursor per partition is recorded by the
- // dispatcher on client ACK under the reserved key MQ_CURSOR_CLIENT (frame
- // attribute emmqoffset, stamped by Kafka / RocketMQ-4.x at poll). Lag =
- // storage endOffset (physical watermark) − recorded cursor. Per-subscriber
- // entries hold logical sequence numbers and are excluded.
- java.util.Map<Integer, Long> cursorByPart = new java.util.HashMap<>();
- String cursorPrefix =
- org.apache.eventmesh.runtime.delivery.ReliableDispatcher.MQ_CURSOR_CLIENT + "#";
- for (java.util.Map.Entry<String, Long> e : offsetStore.readAllOffsets(t).entrySet()) {
- if (!e.getKey().startsWith(cursorPrefix)) {
- continue;
- }
- try {
- int p = Integer.parseInt(e.getKey().substring(cursorPrefix.length()));
- cursorByPart.merge(p, e.getValue(), Math::max);
- } catch (NumberFormatException expected) {
- }
- }
- for (java.util.Map.Entry<Integer, Long> cursorEntry : cursorByPart.entrySet()) {
- int p = cursorEntry.getKey();
- long end = storage.endOffset(t, p);
- long cursor = cursorEntry.getValue();
- if (end >= 0 && cursor >= 0) {
- out.add(new UniMetrics.LabelledLong(
- io.opentelemetry.api.common.Attributes.of(
- io.opentelemetry.api.common.AttributeKey.stringKey("topic"), t,
- io.opentelemetry.api.common.AttributeKey.longKey("partition"), (long) p),
- Math.max(0, end - cursor)));
- }
- }
- }
- return out;
- });
- }
-
- public OffsetStore getOffsetStore() {
- return offsetStore;
- }
-
- public MeshStoragePlugin getStorage() {
- return storage;
- }
-
- /**
- * Synchronous request-reply (§17). Publishes {@code event}, blocks for the matching reply
- * keyed by the {@code x-em-correlation-id} extension, and returns it. On timeout the future is
- * failed and a late reply is discarded. Request-reply is independent of the at-least-once
- * pub/sub path: it neither retries nor dead-letters.
- *
- * @throws Exception if the request times out, publishing fails, or the reply errors
- */
- public CloudEvent request(String topic, CloudEvent event, long timeoutMs) throws Exception {
- String correlationId = readCorrelationId(event);
- CloudEvent toPublish = event;
- if (correlationId == null) {
- correlationId = "req-" + requestSeq.incrementAndGet();
- toPublish = CloudEventBuilder.from(event).withExtension(EXT_CORRELATION_ID, correlationId).build();
- }
- CompletableFuture<CloudEvent> future = new CompletableFuture<>();
- pendingRequests.put(correlationId, future);
- try {
- publish(topic, toPublish).get();
- } catch (Exception e) {
- pendingRequests.remove(correlationId);
- throw e;
- }
- try {
- return future.get(timeoutMs, TimeUnit.MILLISECONDS);
- } catch (TimeoutException te) {
- pendingRequests.remove(correlationId);
- throw new TimeoutException("request-reply timed out: " + correlationId);
- } catch (ExecutionException ee) {
- throw ee;
- } finally {
- metrics.incRequestReply();
- }
- }
-
- /**
- * Deliver a reply to a pending request. Returns false if the request was unknown, already
- * replied, or had timed out (late reply discarded).
- */
- public boolean reply(String correlationId, CloudEvent replyEvent) {
- CompletableFuture<CloudEvent> future = pendingRequests.remove(correlationId);
- if (future == null) {
- log.debug("late/unknown reply for correlationId={} discarded", correlationId);
- return false;
- }
- return future.complete(replyEvent);
- }
-
- private static String readCorrelationId(CloudEvent event) {
- Object value = event.getExtension(EXT_CORRELATION_ID);
- return value == null ? null : value.toString();
- }
-
- private PushChannel channelFor(String clientId) {
- return channels.computeIfAbsent(clientId, id -> new LongPollingChannel(pushService, id));
- }
-
- /** Per-topic monotonic delivery sequence (EventMesh's logical offset for the dispatcher). */
- private long nextOffset(String topic) {
- return topicOffsetSeq.computeIfAbsent(topic, k -> new AtomicLong()).incrementAndGet();
- }
-
- private DeadLetterSink deadLetterSink() {
- return (originalTopic, event, reason, attempts) -> {
- String dlqTopic = originalTopic + "_DLQ";
- // Issue #5292: the dispatcher retires the delivery only once this future reports the
- // DLQ write as durably recorded by the storage plugin.
- java.util.concurrent.CompletableFuture<Boolean> future = new java.util.concurrent.CompletableFuture<>();
- try {
- // event is already an EventMeshFrame (internal); store it directly to the DLQ topic.
- storage.send(dlqTopic, event, new SendCallback() {
- @Override
- public void onSuccess(SendResult sendResult) {
- log.info("event {} dead-lettered to {} after {} attempts: {}",
- event.attributes().get("id"), dlqTopic, attempts, reason);
- future.complete(Boolean.TRUE);
- }
-
- @Override
- public void onException(org.apache.eventmesh.api.exception.OnExceptionContext context) {
- log.error("failed to write DLQ event {} to {}", event.attributes().get("id"), dlqTopic, context.getException());
- future.complete(Boolean.FALSE);
- }
- });
- } catch (Exception e) {
- log.error("failed to send DLQ event {} to {}", event.attributes().get("id"), dlqTopic, e);
- future.complete(Boolean.FALSE);
- }
- return future;
- };
- }
-}
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.runtime.ingress;
+
+import org.apache.eventmesh.api.SendCallback;
+import org.apache.eventmesh.api.SendResult;
+import org.apache.eventmesh.api.storage.MeshStoragePlugin;
+import org.apache.eventmesh.common.wire.EventMeshFrame;
+import org.apache.eventmesh.runtime.delivery.DeadLetterSink;
+import org.apache.eventmesh.runtime.delivery.PushChannel;
+import org.apache.eventmesh.runtime.delivery.ReliableDispatcher;
+import org.apache.eventmesh.runtime.metrics.UniMetrics;
+import org.apache.eventmesh.runtime.metrics.UniTrace;
+import org.apache.eventmesh.runtime.offset.OffsetStore;
+import org.apache.eventmesh.runtime.push.BufferedEvent;
+import org.apache.eventmesh.runtime.push.LongPollingChannel;
+import org.apache.eventmesh.runtime.push.PushService;
+import org.apache.eventmesh.runtime.ratelimit.RateLimitedException;
+import org.apache.eventmesh.runtime.ratelimit.TokenBucketRateLimiter;
+import org.apache.eventmesh.runtime.subscription.CloudEventFilter;
+import org.apache.eventmesh.runtime.subscription.DistributionMode;
+import org.apache.eventmesh.runtime.subscription.Subscription;
+import org.apache.eventmesh.runtime.subscription.SubscriptionManager;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+
+import io.cloudevents.CloudEvent;
+import io.cloudevents.core.builder.CloudEventBuilder;
+
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * Facade that wires the uni-architecture layers into the end-to-end CloudEvents-over-MQ
+ * flow (§6):
+ * <pre>
+ * publish ─▶ MeshStoragePlugin.send
+ * pullLoop ─▶ MeshStoragePlugin.poll ─▶ SubscriptionManager.targetsFor ─▶ ReliableDispatcher.deliver ─▶ PushService
+ * poll ◀─ client long-polls PushService
+ * ack ─▶ PushService.ack ─▶ ReliableDispatcher.ack ─▶ OffsetStore (offset advances only on ACK)
+ * </pre>
+ *
+ * <p>This is the orchestration core; the actual HTTP endpoint wiring (§6 UniIngressHandler)
+ * delegates here. Phase-1 thin storage adapter means {@code partition} is {@code -1} and the offset
+ * is a per-topic monotonic logical counter until the native storage reimplementation supplies real
+ * partition/offset (Phase 1 step 2).</p>
+ */
+@Slf4j
+public class UniIngressService {
+
+ private final MeshStoragePlugin storage;
+ private final OffsetStore offsetStore;
+ private final SubscriptionManager subscriptionManager;
+ private final ReliableDispatcher dispatcher;
+ private final PushService pushService;
+ private volatile org.apache.eventmesh.runtime.cluster.ClusterCoordinator cluster;
+ private volatile org.apache.eventmesh.runtime.cluster.PartitionOwnership partitionOwnership;
+ /** Self-collected load metrics for session-distribution load balancing (§3). Null until wired. */
+ private volatile LoadMeter loadMeter;
+
+ /** Per-topic poll stats for the {@code poll_idle_ratio} gauge (§13.5.1). */
+ private final ConcurrentHashMap<String, Long> pollCount = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, Long> pollEmptyCount = new ConcurrentHashMap<>();
+
+ /** Connector offset store (remote side) — String key → String offset value (§8.9). */
+ private final java.util.concurrent.ConcurrentHashMap<String, String> connectorOffsets = new java.util.concurrent.ConcurrentHashMap<>();
+ private final UniMetrics metrics;
+
+ private final ConcurrentHashMap<String, PushChannel> channels = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, AtomicLong> topicOffsetSeq = new ConcurrentHashMap<>();
+ private final ConcurrentHashMap<String, CompletableFuture<CloudEvent>> pendingRequests = new ConcurrentHashMap<>();
+ private final AtomicLong requestSeq = new AtomicLong();
+ private final ConcurrentHashMap<String, TokenBucketRateLimiter> topicLimiters = new ConcurrentHashMap<>();
+
+ /**
+ * CloudEvents extension carrying the request's correlation id (§17).
+ *
+ * <p>Named without hyphens because the CloudEvents spec restricts extension attribute names to
+ * lower-case ASCII letters and digits — the redesign doc's {@code x-em-correlation-id} spelling
+ * is rejected by the SDK's name validation.</p>
+ */
+ public static final String EXT_CORRELATION_ID = "emcorrelationid";
+
+ public UniIngressService(MeshStoragePlugin storage, OffsetStore offsetStore) {
+ this(storage, offsetStore, new SubscriptionManager(), new PushService(),
+ ReliableDispatcher.DEFAULT_ACK_TIMEOUT_MS, ReliableDispatcher.DEFAULT_MAX_ATTEMPTS,
+ System::currentTimeMillis);
+ }
+
+ /**
+ * Test-friendly constructor with an injectable clock and retry parameters.
+ */
+ public UniIngressService(MeshStoragePlugin storage, OffsetStore offsetStore,
+ SubscriptionManager subscriptionManager, PushService pushService,
+ long ackTimeoutMs, int maxAttempts, java.util.function.LongSupplier clock) {
+ this.storage = storage;
+ this.offsetStore = offsetStore;
+ this.subscriptionManager = subscriptionManager;
+ this.pushService = pushService;
+ this.metrics = new UniMetrics();
+ this.dispatcher = new ReliableDispatcher(ackTimeoutMs, maxAttempts, clock, offsetStore, deadLetterSink(),
+ metrics, ReliableDispatcher.DEFAULT_JITTER_RATIO);
+ // Sub-PR B: re-ACK any in-flight deliveries from a previous JVM so they do not become
+ // orphans after a crash. Safe on first start (store is empty) and idempotent.
+ this.dispatcher.recover();
+ }
+
+ // ---- connector offset (remote side, §8.9) ----
+
+ public String getConnectorOffset(String connectorId) {
+ return connectorOffsets.get(connectorId);
+ }
+
+ public void putConnectorOffset(String connectorId, String offset) {
+ connectorOffsets.put(connectorId, offset);
+ }
+
+ /**
+ * Publish a CloudEvent to {@code topic} (persisted to MQ). Completes when the storage plugin
+ * acknowledges the write.
+ */
+ public CompletableFuture<Void> publish(String topic, CloudEvent event) {
+ if (loadMeter != null && event.getData() != null) {
+ loadMeter.recordInflow(event.getData().toBytes().length);
+ }
+ CompletableFuture<Void> future = new CompletableFuture<>();
+ TokenBucketRateLimiter limiter = topicLimiters.get(topic);
+ if (limiter != null && !limiter.tryAcquire()) {
+ metrics.incRateLimited();
+ future.completeExceptionally(new RateLimitedException(topic));
+ return future;
+ }
+ try {
+ // Ingress boundary: CloudEvent (the public/external format) → EventMeshFrame (the internal
+ // wire unit). From here on the event is a Frame internally; egress converts back to the
+ // client's protocol (CloudEvents / MeshMessage) at the delivery boundary.
+ org.apache.eventmesh.common.wire.EventMeshFrame frame =
+ org.apache.eventmesh.common.wire.EventMeshFrame.fromCloudEvent(event);
+ storage.send(topic, frame, new SendCallback() {
+ @Override
+ public void onSuccess(SendResult sendResult) {
+ metrics.incPublish();
+ UniTrace.end(UniTrace.startPublish(topic, event));
+ future.complete(null);
+ }
+
+ @Override
+ public void onException(org.apache.eventmesh.api.exception.OnExceptionContext context) {
+ metrics.incPublishFailed();
+ future.completeExceptionally(context.getException());
+ }
+ });
+ } catch (Exception e) {
+ metrics.incPublishFailed();
+ future.completeExceptionally(e);
+ }
+ return future;
+ }
+
+ /**
+ * Publish an already-internal {@link org.apache.eventmesh.common.wire.EventMeshFrame} (no
+ * CloudEvent→Frame boundary conversion — the caller built the frame directly, e.g. the legacy
+ * MeshMessage TCP path via {@code MeshMessageFrameCodec}). Completes when storage acks the write.
+ */
+ public CompletableFuture<Void> publish(String topic, org.apache.eventmesh.common.wire.EventMeshFrame frame) {
+ if (loadMeter != null && frame.data() != null) {
+ loadMeter.recordInflow(frame.data().length);
+ }
+ CompletableFuture<Void> future = new CompletableFuture<>();
+ TokenBucketRateLimiter limiter = topicLimiters.get(topic);
+ if (limiter != null && !limiter.tryAcquire()) {
+ metrics.incRateLimited();
+ future.completeExceptionally(new RateLimitedException(topic));
+ return future;
+ }
+ try {
+ storage.send(topic, frame, new SendCallback() {
+ @Override
+ public void onSuccess(SendResult sendResult) {
+ metrics.incPublish();
+ future.complete(null);
+ }
+
+ @Override
+ public void onException(org.apache.eventmesh.api.exception.OnExceptionContext context) {
+ metrics.incPublishFailed();
+ future.completeExceptionally(context.getException());
+ }
+ });
+ } catch (Exception e) {
+ metrics.incPublishFailed();
+ future.completeExceptionally(e);
+ }
+ return future;
+ }
+
+ /**
+ * Publish a batch of CloudEvents to {@code topic} (§13.7.3). Completes when all are persisted.
+ * The storage plugin's own batching (e.g. Kafka producer accumulator) amortises the per-event
+ * RTT; this layer fans out the per-event futures and joins them.
+ */
+ public CompletableFuture<Void> publishBatch(String topic, java.util.List<CloudEvent> events) {
+ if (events == null || events.isEmpty()) {
+ return CompletableFuture.completedFuture(null);
+ }
+ CompletableFuture<?>[] futures = events.stream()
+ .map(e -> publish(topic, e))
+ .toArray(CompletableFuture[]::new);
+ return CompletableFuture.allOf(futures);
+ }
+
+ /**
+ * Primary ingress entry for a batch of internal EventMeshFrames (#5299). The HTTP path uses
+ * {@code FrameAdaptors.get("cloudevents").toFrame(body)} per event then calls this method; the
+ * CloudEvent-typed {@link #publishBatch(String, java.util.List)} is preserved for binary
+ * compatibility with the TCP bridge and any out-of-tree caller.
+ */
+ public CompletableFuture<Void> publishBatchFrames(String topic, java.util.List<EventMeshFrame> frames) {
+ if (frames == null || frames.isEmpty()) {
+ return CompletableFuture.completedFuture(null);
+ }
+ CompletableFuture<?>[] futures = frames.stream()
+ .map(f -> publish(topic, f))
+ .toArray(CompletableFuture[]::new);
+ return CompletableFuture.allOf(futures);
+ }
+
+ /**
+ * Register a custom push channel for a client (e.g. a WebHook URL for legacy HTTP webhook-push
+ * subscribers, or a {@code TcpPushChannel} for legacy TCP clients). Overrides the default
+ * long-polling channel for this {@code clientId} on subsequent dispatches.
+ */
+ public void registerChannel(String clientId, PushChannel channel) {
+ pushService.register(clientId);
+ channels.put(clientId, channel);
+ }
+
+ /**
+ * Configure a per-topic rate limit (§6.6). Subsequent publishes above {@code permitsPerSecond}
+ * (burst {@code capacity}) fail the publish future with {@link RateLimitedException}.
+ */
+ public void setTopicRateLimit(String topic, long capacity, double permitsPerSecond) {
+ topicLimiters.put(topic, new TokenBucketRateLimiter(capacity, permitsPerSecond));
+ }
+
+ /**
+ * Register a subscription. The subscriber retrieves events via {@link #poll}.
+ *
+ * @return the subscription id
+ */
+ public String subscribe(String topic, String clientId, DistributionMode mode, CloudEventFilter filter) {
+ pushService.register(clientId);
+ String subId = subscriptionManager.subscribe(topic, clientId, mode, filter, event -> {
+ // No-op: the reliability layer (ReliableDispatcher) owns delivery, not this fire-and-
+ // forget callback. Kept to satisfy the SubscriptionManager handler contract.
+ });
+ // §13.2: when clustered, also register cluster-wide so other instances can route events
+ // for this subscriber here (via ClusterCoordinator → HttpForwarder /internal/forward).
+ // Without this the HTTP /events/subscribe path is local-only and a publish on a peer
+ // never reaches this subscriber.
+ if (cluster != null) {
+ cluster.subscribe(topic, clientId, mode, null);
+ }
+ return subId;
+ }
+
+ /**
+ * Remove a subscription.
+ */
+ public boolean unsubscribe(String subscriptionId) {
+ // Resolve topic + clientId before removing so a clustered subscription can be deregistered
+ // cluster-wide (the coordinator's unsubscribe is keyed by topic+clientId, not subId).
+ org.apache.eventmesh.runtime.subscription.Subscription sub = subscriptionManager.getSubscription(subscriptionId);
+ boolean removed = subscriptionManager.unsubscribe(subscriptionId);
+ if (removed && cluster != null && sub != null) {
+ cluster.unsubscribe(sub.getTopic(), sub.getClientId());
+ }
+ return removed;
+ }
+
+ /** Remove one client's subscription to one topic (HTTP /events/unsubscribe with {clientId, topic}). */
+ public boolean unsubscribe(String topic, String clientId) {
+ boolean removed = subscriptionManager.unsubscribe(topic, clientId);
+ if (removed && cluster != null) {
+ cluster.unsubscribe(topic, clientId);
+ }
+ return removed;
+ }
+
+ /** Remove ALL subscriptions for a client (HTTP /events/unsubscribe with {clientId} only).
+ * Propagates cluster-wide + frees the PushService buffer. */
+ public int unsubscribeByClient(String clientId) {
+ java.util.Set<String> topics = subscriptionManager.topicsForClient(clientId);
+ int removed = subscriptionManager.unsubscribeByClient(clientId);
+ if (cluster != null) {
+ for (String topic : topics) {
+ cluster.unsubscribe(topic, clientId);
+ }
+ }
+ pushService.removeClient(clientId);
+ return removed;
+ }
+
+ /**
+ * Pump: pull a batch from storage, route to each target subscriber via the reliability layer.
+ *
+ * @return number of events pulled
+ */
+ public int pullAndDispatch(String topic, int maxEvents, long timeoutMs) {
+ pollCount.merge(topic, 1L, Long::sum);
+ // Multi-instance (§13.2.3): poll only the partitions this instance owns. Single-instance
+ // (no ownership) or unknown partition count → poll the whole topic (partition -1).
+ java.util.List<Integer> owned = partitionOwnership == null ? null : partitionOwnership.ownedPartitions(topic);
+ int total;
+ if (owned == null) {
+ total = pullAndDispatchPartition(topic, -1, maxEvents, timeoutMs);
+ } else if (owned.isEmpty()) {
+ total = 0; // owns none -> do not poll (avoids duplicate with the real owners)
+ } else {
+ total = 0;
+ for (int p : owned) {
+ total += pullAndDispatchPartition(topic, p, maxEvents, timeoutMs);
+ }
+ }
+ if (total == 0) {
+ pollEmptyCount.merge(topic, 1L, Long::sum);
+ }
+ return total;
+ }
+
+ private int pullAndDispatchPartition(String topic, int partition, int maxEvents, long timeoutMs) {
+ // Pull EventMeshFrames (internal wire); dispatch each as a Frame through the internal
+ // pipeline (filter/dispatcher/egress all carry Frame now; egress converts to the client's
+ // protocol at the wire boundary).
+ List<org.apache.eventmesh.common.wire.EventMeshFrame> frames = storage.poll(topic, partition, -1, maxEvents, timeoutMs);
+ if (frames == null || frames.isEmpty()) {
+ return 0;
+ }
+ long start = System.nanoTime();
+ for (org.apache.eventmesh.common.wire.EventMeshFrame f : frames) {
+ if (isExpired(f)) {
+ // §13.3.4 TTL: drop expired events instead of dispatching.
+ log.debug("dropping expired event {} on topic {} (emttl elapsed)", f.attributes().get("id"), topic);
+ continue;
+ }
+ io.opentelemetry.api.trace.Span dispatchSpan = UniTrace.startDispatch(topic, f);
+ if (cluster != null) {
+ // Multi-instance: route via the cluster coordinator (local vs cross-instance forward).
+ cluster.dispatch(topic, f);
+ } else {
+ // develop's OffsetExtensions (CE-extension-carried MQ offset) is superseded by the
+ // Frame architecture: the POP check key rides in frame attributes (empopck) and the
+ // deferred broker ACK fires on client ACK — same at-least-once goal, Frame-native.
+ long offset = nextOffset(topic);
+ // P2 fix: if the frame carries a POP check key (RocketMQ 5.x deferred ACK), build a
+ // callback that ACKs the broker on client ACK (restoring at-least-once).
+ String popCk = f.attributes().get("empopck");
+ Runnable mqAck = (popCk != null) ? () -> storage.ackPulledMessage(topic, popCk) : null;
+ for (Subscription target : subscriptionManager.targetsFor(topic, f)) {
+ dispatcher.deliver(topic, partition, offset, f, target.getClientId(),
+ channelFor(target.getClientId()), mqAck);
+ }
+ }
+ UniTrace.end(dispatchSpan);
+ }
+ metrics.addDispatchLatencyNanos(System.nanoTime() - start);
+ metrics.incDispatched(frames.size());
+ return frames.size();
+ }
+
+ /**
+ * TTL expiry check (§13.3.4): an event with an {@code emttl} attribute (ms) and a {@code time}
+ * is expired when {@code now > time + emttl}. Events without either field never expire. Reads the
+ * attributes off the internal EventMeshFrame (emttl/time are preserved in its KV section).
+ */
+ private boolean isExpired(org.apache.eventmesh.common.wire.EventMeshFrame event) {
+ String ttl = event.attributes().get("emttl");
+ String time = event.attributes().get("time");
+ if (ttl == null || time == null) {
+ return false;
+ }
+ try {
+ long ttlMs = Long.parseLong(ttl);
+ long eventTime = java.time.OffsetDateTime.parse(time).toInstant().toEpochMilli();
+ return System.currentTimeMillis() > eventTime + ttlMs;
+ } catch (NumberFormatException | java.time.format.DateTimeParseException e) {
+ return false;
+ }
+ }
+
+ /**
+ * Local delivery for one subscriber — handed to the reliability layer + the subscriber's push
+ * channel. Exposed so a {@link org.apache.eventmesh.runtime.cluster.ClusterCoordinator} can route
+ * same-instance targets here while forwarding remote ones.
+ */
+ public boolean deliverLocal(String topic, String clientId, org.apache.eventmesh.common.wire.EventMeshFrame event) {
+ long offset = nextOffset(topic);
+ dispatcher.deliver(topic, -1, offset, event, clientId, channelFor(clientId));
+ return true;
+ }
+
+ /**
+ * Enable multi-instance coordination: when set, {@link #pullAndDispatch} routes each event
+ * through the cluster coordinator (local targets via {@link #deliverLocal}, remote via forward).
+ */
+ public void withCluster(org.apache.eventmesh.runtime.cluster.ClusterCoordinator cluster) {
+ this.cluster = cluster;
+ }
+
+ /**
+ * Topics this instance should pull and partition-assign: local active topics UNION cluster-wide
+ * topics (topics with a remote subscriber discovered via the Meta watch). Without the cluster
+ * half, an instance with no local subscriber for a topic never pulls it, so messages on its
+ * partitions can't be forwarded to the remote subscriber (multi-instance message loss). Returns
+ * local-only when not clustered. Both {@code UniRuntime.pullLoop} and {@code PartitionOwnership}'s
+ * topic source use this so the pull set and the assignment set stay consistent (otherwise an
+ * unassigned cluster topic would degrade to poll-all and duplicate).
+ */
+ public java.util.Set<String> activeTopicsClustered() {
+ java.util.Set<String> topics = new java.util.HashSet<>(getSubscriptionManager().activeTopics());
+ if (cluster != null) {
+ topics.addAll(cluster.subscriptionTopics());
+ }
+ return topics;
+ }
+
+ // ===================== Lite Topic (RIP-83, 5.x-only) =====================
+ // Lite topic ops are exposed only when the storage plugin implements LiteTopicCapable; otherwise
+ // they fail fast (UnsupportedOperationException). The HTTP layer (/events/lite/*) delegates here.
+
+ /**
+ * Publish one CloudEvent to a lite topic (parentTopic, liteTopic). The storage plugin routes it
+ * into the lite topic's LMQ. Requires a {@link org.apache.eventmesh.api.storage.LiteTopicCapable}
+ * storage (the 5.x plugin); 4.x/kafka/standalone storages throw.
+ */
+ public CompletableFuture<Void> publishLite(String parentTopic, String liteTopic, CloudEvent event) {
+ CompletableFuture<Void> future = new CompletableFuture<>();
+ if (!(storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable)) {
+ future.completeExceptionally(new UnsupportedOperationException("storage does not support lite topic"));
+ return future;
+ }
+ try {
+ // Boundary: CloudEvent → EventMeshFrame (internal wire unit).
+ org.apache.eventmesh.common.wire.EventMeshFrame frame =
+ org.apache.eventmesh.common.wire.EventMeshFrame.fromCloudEvent(event);
+ ((org.apache.eventmesh.api.storage.LiteTopicCapable) storage).sendLite(parentTopic, liteTopic, frame,
+ new SendCallback() {
+ @Override
+ public void onSuccess(SendResult sendResult) {
+ future.complete(null);
+ }
+
+ @Override
+ public void onException(org.apache.eventmesh.api.exception.OnExceptionContext context) {
+ future.completeExceptionally(context.getException());
+ }
+ });
+ } catch (Exception e) {
+ future.completeExceptionally(e);
+ }
+ return future;
+ }
+
+ /**
+ * Primary lite ingress: publish a single internal EventMeshFrame to a lite topic (LMQ) (#5299).
+ * The HTTP path uses {@code FrameAdaptors.get("cloudevents").toFrame(body)} then calls this
+ * method; no further CloudEvent touches runtime state.
+ */
+ public CompletableFuture<Void> publishLiteFrame(String parentTopic, String liteTopic, EventMeshFrame frame) {
+ CompletableFuture<Void> future = new CompletableFuture<>();
+ if (!(storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable)) {
+ future.completeExceptionally(new UnsupportedOperationException("storage does not support lite topic"));
+ return future;
+ }
+ if (loadMeter != null && frame.data() != null) {
+ loadMeter.recordInflow(frame.data().length);
+ }
+ try {
+ ((org.apache.eventmesh.api.storage.LiteTopicCapable) storage).sendLite(parentTopic, liteTopic, frame,
+ new SendCallback() {
+ @Override
+ public void onSuccess(SendResult sendResult) {
+ future.complete(null);
+ }
+
+ @Override
+ public void onException(org.apache.eventmesh.api.exception.OnExceptionContext context) {
+ future.completeExceptionally(context.getException());
+ }
+ });
+ } catch (Exception e) {
+ future.completeExceptionally(e);
+ }
+ return future;
+ }
+
+ /**
+ * Pull a batch of CloudEvents from a lite topic (direct pull from the LMQ; no deliveryId / no
+ * EventMesh reliability layer — the lite consumer self-manages offset in the storage plugin).
+ * Frames pulled from the LMQ are decoded back to CloudEvents at this boundary. Empty list if lite
+ * is not supported.
+ */
+ public List<CloudEvent> pollLite(String parentTopic, String liteTopic, int maxEvents, long timeoutMs) {
+ if (!(storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable)) {
+ return java.util.Collections.emptyList();
+ }
+ java.util.List<CloudEvent> out = new java.util.ArrayList<>();
+ for (org.apache.eventmesh.common.wire.EventMeshFrame f
+ : ((org.apache.eventmesh.api.storage.LiteTopicCapable) storage).pullLite(parentTopic, liteTopic, maxEvents, timeoutMs)) {
+ out.add(f.toCloudEvent());
+ }
+ return out;
+ }
+
+ /**
+ * Primary lite poll: drain a batch of internal EventMeshFrames from the LMQ (#5299). The
+ * egress boundary (CloudEventsFrameAdaptor) converts back to CloudEvents JSON for the HTTP
+ * response body. Returns an empty list if the storage does not support lite topics.
+ */
+ public List<EventMeshFrame> pollLiteFrames(String parentTopic, String liteTopic, int maxEvents, long timeoutMs) {
+ if (!(storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable)) {
+ return java.util.Collections.emptyList();
+ }
+ return ((org.apache.eventmesh.api.storage.LiteTopicCapable) storage)
+ .pullLite(parentTopic, liteTopic, maxEvents, timeoutMs);
+ }
+
+ /**
+ * Publish a pre-encoded EventMeshFrame byte payload to a lite topic (the internal streaming wire
+ * path — SessionRouter publishes frame bytes). The payload IS an encoded EventMeshFrame; decode
+ * to the Frame object the SPI now expects.
+ */
+ public CompletableFuture<Void> publishLiteBytes(String parentTopic, String liteTopic, byte[] payload) {
+ if (loadMeter != null) {
+ loadMeter.recordInflow(payload.length);
+ }
+ CompletableFuture<Void> future = new CompletableFuture<>();
+ if (!(storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable)) {
+ future.completeExceptionally(new UnsupportedOperationException("storage does not support lite topic"));
+ return future;
+ }
+ try {
+ org.apache.eventmesh.common.wire.EventMeshFrame frame = org.apache.eventmesh.common.wire.EventMeshFrame.decode(payload);
+ ((org.apache.eventmesh.api.storage.LiteTopicCapable) storage).sendLite(parentTopic, liteTopic, frame,
+ new SendCallback() {
+ @Override
+ public void onSuccess(SendResult sendResult) {
+ future.complete(null);
+ }
+
+ @Override
+ public void onException(org.apache.eventmesh.api.exception.OnExceptionContext context) {
+ future.completeExceptionally(context.getException());
+ }
+ });
+ } catch (Exception e) {
+ future.completeExceptionally(e);
+ }
+ return future;
+ }
+
+ /**
+ * Pull a batch of pre-encoded EventMeshFrame byte payloads from a lite topic (the byte counterpart
+ * of {@link #pollLite}, for the internal streaming wire). Each entry is an encoded frame.
+ * Empty list if lite is not supported.
+ */
+ public List<byte[]> pollLiteBytes(String parentTopic, String liteTopic, int maxEvents, long timeoutMs) {
+ if (!(storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable)) {
+ return java.util.Collections.emptyList();
+ }
+ java.util.List<byte[]> out = new java.util.ArrayList<>();
+ for (org.apache.eventmesh.common.wire.EventMeshFrame f
+ : ((org.apache.eventmesh.api.storage.LiteTopicCapable) storage).pullLite(parentTopic, liteTopic, maxEvents, timeoutMs)) {
+ out.add(f.encode());
+ }
+ return out;
+ }
+
+ /**
+ * Ensure {@code parentTopic} is lite-capable and declare {@code liteTopic} under it. Throws if the
+ * storage does not support lite.
+ */
+ public void createLiteTopic(String parentTopic, String liteTopic) throws Exception {
+ if (!(storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable)) {
+ throw new UnsupportedOperationException("storage does not support lite topic");
+ }
+ ((org.apache.eventmesh.api.storage.LiteTopicCapable) storage).createLiteTopic(parentTopic, liteTopic);
+ }
+
+ /**
+ * Ensure {@code parentTopic} is lite-capable with the given {@code queueCount} and declare
+ * {@code liteTopic} under it. Use {@code queueCount=1} for strict in-order delivery (e.g. a
+ * streaming-call response channel). Throws if the storage does not support lite.
+ */
+ public void createLiteTopic(String parentTopic, String liteTopic, int queueCount) throws Exception {
+ if (!(storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable)) {
+ throw new UnsupportedOperationException("storage does not support lite topic");
+ }
+ ((org.apache.eventmesh.api.storage.LiteTopicCapable) storage).createLiteTopic(parentTopic, liteTopic, queueCount);
+ }
+
+ /**
+ * @return true iff the storage plugin implements {@link org.apache.eventmesh.api.storage.LiteTopicCapable}.
+ */
+ public boolean isLiteCapable() {
+ return storage instanceof org.apache.eventmesh.api.storage.LiteTopicCapable;
+ }
+
+ /**
+ * Enable multi-instance partition ownership (§13.2.3): when set, {@link #pullAndDispatch} polls
+ * only this instance's owned partitions instead of the whole topic.
+ */
+ public void withPartitionOwnership(org.apache.eventmesh.runtime.cluster.PartitionOwnership ownership) {
+ this.partitionOwnership = ownership;
+ }
+
+ /** Wire the load meter; ingress/egress points call its record* methods. */
+ public void withLoadMeter(LoadMeter loadMeter) {
+ this.loadMeter = loadMeter;
+ }
+
+ /**
+ * @return the wired load meter, or null if not configured (single-instance / tests).
+ */
+ public LoadMeter loadMeter() {
+ return loadMeter;
+ }
+
+ /**
+ * Subscriber long-polls its buffered deliveries.
+ */
+ public List<BufferedEvent> poll(String clientId, int maxEvents, long timeoutMs) {
+ return pushService.poll(clientId, maxEvents, timeoutMs);
+ }
+
+ /**
+ * Subscriber acknowledges a delivery — the offset advances only on ACK (at-least-once).
+ */
+ public boolean ack(String deliveryId) {
+ return pushService.ack(deliveryId);
+ }
+
+ /**
+ * Drive retry / DLQ. Call periodically from a scheduler.
+ */
+ public int dispatcherTick() {
+ return dispatcher.tick();
+ }
+
+ /**
+ * Operational metrics counters (publish/dispatch/ack/retry/DLQ). The dispatcher and ingress
+ * share one instance.
+ */
+ public UniMetrics getMetrics() {
+ return metrics;
+ }
+
+ // Accessors for the admin facade (Phase 7.5). The service is the single owner of these
+ // collaborators; exposing them avoids reconstructing them out of band.
+ public SubscriptionManager getSubscriptionManager() {
+ return subscriptionManager;
+ }
+
+ public PushService getPushService() {
+ return pushService;
+ }
+
+ public ReliableDispatcher getDispatcher() {
+ return dispatcher;
+ }
+
+ /**
+ * @return the multi-instance partition ownership (null when clustering is disabled).
+ */
+ public org.apache.eventmesh.runtime.cluster.PartitionOwnership getPartitionOwnership() {
+ return partitionOwnership;
+ }
+
+ /**
+ * Stale-poll cleanup (§13.6.5): evict clients that haven't polled within {@code thresholdMs} —
+ * drops their subscriptions and push buffer so zombie subscriptions don't leak.
+ *
+ * @return number of subscriptions removed
+ */
+ public int cleanupStaleClients(long thresholdMs) {
+ int removed = 0;
+ for (String cid : pushService.getStaleClientIds(thresholdMs)) {
+ removed += subscriptionManager.unsubscribeByClient(cid);
+ pushService.removeClient(cid);
+ log.info("evicted stale client {} (no poll within {}ms)", cid, thresholdMs);
+ }
+ return removed;
+ }
+
+ /**
+ * Register OTel observable gauges backed by live runtime state (§13.5.1 gauges with *).
+ * Call once at boot. Gauges read on each OTel collection cycle.
+ */
+ public void registerRuntimeGauges() {
+ metrics.registerGauge("eventmesh_pending_queue_size", "total buffered events across all clients",
+ () -> {
+ long sum = 0;
+ for (String cid : pushService.clientIds()) {
+ sum += pushService.pending(cid);
+ }
+ return sum;
+ });
+ metrics.registerGauge("eventmesh_slow_consumer_count", "clients in SLOW or STALLED state",
+ pushService::slowConsumerCount);
+ metrics.registerGauge("eventmesh_active_topics", "topics with active subscribers",
+ () -> subscriptionManager.activeTopics().size());
+ metrics.registerGauge("eventmesh_active_subscribers", "active subscriptions across all topics",
+ () -> {
+ int sum = 0;
+ for (String t : subscriptionManager.activeTopics()) {
+ sum += subscriptionManager.activeSubscriptions(t).size();
+ }
+ return sum;
+ });
+
+ // Labelled gauges (§13.5.1) — emit one reading per topic / partition.
+ metrics.registerLabelledGauge("eventmesh_poll_idle_ratio",
+ "fraction of poll cycles returning no events (per-mille, per topic)",
+ () -> {
+ java.util.List<UniMetrics.LabelledLong> out = new java.util.ArrayList<>();
+ for (String t : pollCount.keySet()) {
+ long total = pollCount.getOrDefault(t, 0L);
+ long empty = pollEmptyCount.getOrDefault(t, 0L);
+ long perMille = total == 0 ? 0 : Math.round((double) empty / total * 1000);
+ out.add(new UniMetrics.LabelledLong(
+ io.opentelemetry.api.common.Attributes.of(io.opentelemetry.api.common.AttributeKey.stringKey("topic"), t),
+ perMille));
+ }
+ return out;
+ });
+
+ metrics.registerLabelledGauge("eventmesh_partition_owner",
+ "1 for each partition this instance owns (per topic/partition)",
+ () -> {
+ java.util.List<UniMetrics.LabelledLong> out = new java.util.ArrayList<>();
+ if (partitionOwnership != null) {
+ for (String t : subscriptionManager.activeTopics()) {
+ java.util.List<Integer> owned = partitionOwnership.ownedPartitions(t);
+ if (owned == null) {
+ continue;
+ }
+ for (int p : owned) {
+ out.add(new UniMetrics.LabelledLong(
+ io.opentelemetry.api.common.Attributes.of(
+ io.opentelemetry.api.common.AttributeKey.stringKey("topic"), t,
+ io.opentelemetry.api.common.AttributeKey.longKey("partition"), (long) p,
+ io.opentelemetry.api.common.AttributeKey.stringKey("instance"), "self"),
+ 1L));
+ }
+ }
+ }
+ return out;
+ });
+
+ metrics.registerLabelledGauge("eventmesh_offset_lag",
+ "MQ end offset - max ACK offset (per topic/partition) — total consumer lag",
+ () -> {
+ java.util.List<UniMetrics.LabelledLong> out = new java.util.ArrayList<>();
+ if (partitionOwnership == null) {
+ return out;
+ }
+ for (String t : subscriptionManager.activeTopics()) {
+ java.util.List<Integer> owned = partitionOwnership.ownedPartitions(t);
+ if (owned == null) {
+ continue;
+ }
+ // Max ACK offset per partition across all clients (key = clientId#partition).
+ // Exclude the reserved __mqcursor__ key (MQ physical offset — different
+ // magnitude from the logical sequence numbers, mixing them corrupts the gauge).
+ java.util.Map<Integer, Long> maxAckByPart = new java.util.HashMap<>();
+ String reservedPrefix =
+ org.apache.eventmesh.runtime.delivery.ReliableDispatcher.MQ_CURSOR_CLIENT + "#";
+ for (java.util.Map.Entry<String, Long> e : offsetStore.readAllOffsets(t).entrySet()) {
+ if (e.getKey().startsWith(reservedPrefix)) {
+ continue;
+ }
+ int sep = e.getKey().lastIndexOf('#');
+ if (sep > 0) {
+ try {
+ int p = Integer.parseInt(e.getKey().substring(sep + 1));
+ maxAckByPart.merge(p, e.getValue(), Math::max);
+ } catch (NumberFormatException expected) {
+ // offset key suffix is not a numeric partition; skip
+ }
+ }
+ }
+ for (int p : owned) {
+ long end = storage.endOffset(t, p);
+ long ack = maxAckByPart.getOrDefault(p, -1L);
+ if (end >= 0 && ack >= 0) {
+ out.add(new UniMetrics.LabelledLong(
+ io.opentelemetry.api.common.Attributes.of(
+ io.opentelemetry.api.common.AttributeKey.stringKey("topic"), t,
+ io.opentelemetry.api.common.AttributeKey.longKey("partition"), (long) p),
+ Math.max(0, end - ack)));
+ }
+ }
+ }
+ return out;
+ });
+
+ metrics.registerLabelledGauge("eventmesh_push_ack_lag",
+ "max push offset - max ACK offset (per topic/partition) — in-flight deliveries",
+ () -> {
+ java.util.List<UniMetrics.LabelledLong> out = new java.util.ArrayList<>();
+ for (String t : subscriptionManager.activeTopics()) {
+ // Frame architecture: the MQ physical cursor per partition is recorded by the
+ // dispatcher on client ACK under the reserved key MQ_CURSOR_CLIENT (frame
+ // attribute emmqoffset, stamped by Kafka / RocketMQ-4.x at poll). Lag =
+ // storage endOffset (physical watermark) − recorded cursor. Per-subscriber
+ // entries hold logical sequence numbers and are excluded.
+ java.util.Map<Integer, Long> cursorByPart = new java.util.HashMap<>();
+ String cursorPrefix =
+ org.apache.eventmesh.runtime.delivery.ReliableDispatcher.MQ_CURSOR_CLIENT + "#";
+ for (java.util.Map.Entry<String, Long> e : offsetStore.readAllOffsets(t).entrySet()) {
+ if (!e.getKey().startsWith(cursorPrefix)) {
+ continue;
+ }
+ try {
+ int p = Integer.parseInt(e.getKey().substring(cursorPrefix.length()));
+ cursorByPart.merge(p, e.getValue(), Math::max);
+ } catch (NumberFormatException expected) {
+ }
+ }
+ for (java.util.Map.Entry<Integer, Long> cursorEntry : cursorByPart.entrySet()) {
+ int p = cursorEntry.getKey();
+ long end = storage.endOffset(t, p);
+ long cursor = cursorEntry.getValue();
+ if (end >= 0 && cursor >= 0) {
+ out.add(new UniMetrics.LabelledLong(
+ io.opentelemetry.api.common.Attributes.of(
+ io.opentelemetry.api.common.AttributeKey.stringKey("topic"), t,
+ io.opentelemetry.api.common.AttributeKey.longKey("partition"), (long) p),
+ Math.max(0, end - cursor)));
+ }
+ }
+ }
+ return out;
+ });
+ }
+
+ public OffsetStore getOffsetStore() {
+ return offsetStore;
+ }
+
+ public MeshStoragePlugin getStorage() {
+ return storage;
+ }
+
+ /**
+ * Synchronous request-reply (§17). Publishes {@code event}, blocks for the matching reply
+ * keyed by the {@code x-em-correlation-id} extension, and returns it. On timeout the future is
+ * failed and a late reply is discarded. Request-reply is independent of the at-least-once
+ * pub/sub path: it neither retries nor dead-letters.
+ *
+ * @throws Exception if the request times out, publishing fails, or the reply errors
+ */
+ public CloudEvent request(String topic, CloudEvent event, long timeoutMs) throws Exception {
+ String correlationId = readCorrelationId(event);
+ CloudEvent toPublish = event;
+ if (correlationId == null) {
+ correlationId = "req-" + requestSeq.incrementAndGet();
+ toPublish = CloudEventBuilder.from(event).withExtension(EXT_CORRELATION_ID, correlationId).build();
+ }
+ CompletableFuture<CloudEvent> future = new CompletableFuture<>();
+ pendingRequests.put(correlationId, future);
+ try {
+ publish(topic, toPublish).get();
+ } catch (Exception e) {
+ pendingRequests.remove(correlationId);
+ throw e;
+ }
+ try {
+ return future.get(timeoutMs, TimeUnit.MILLISECONDS);
+ } catch (TimeoutException te) {
+ pendingRequests.remove(correlationId);
+ throw new TimeoutException("request-reply timed out: " + correlationId);
+ } catch (ExecutionException ee) {
+ throw ee;
+ } finally {
+ metrics.incRequestReply();
+ }
+ }
+
+ /**
+ * Deliver a reply to a pending request. Returns false if the request was unknown, already
+ * replied, or had timed out (late reply discarded).
+ */
+ public boolean reply(String correlationId, CloudEvent replyEvent) {
+ CompletableFuture<CloudEvent> future = pendingRequests.remove(correlationId);
+ if (future == null) {
+ log.debug("late/unknown reply for correlationId={} discarded", correlationId);
+ return false;
+ }
+ return future.complete(replyEvent);
+ }
+
+ /**
+ * Primary request-reply path (#5299, §17): publish a Frame, wait for the matching reply Frame.
+ * Implementation note: this method round-trips through CloudEvent (via {@code
+ * EventMeshFrame.fromCloudEvent / toCloudEvent}) so the existing {@code pendingRequests} map
+ * (typed as {@code CompletableFuture<CloudEvent>}) stays untouched. A future sub-PR will
+ * re-type {@code pendingRequests} to {@code CompletableFuture<EventMeshFrame>} and remove
+ * the round-trip. The HTTP path now uses this method (the previous {@code request(CloudEvent)}
+ * is preserved for binary compat with the TCP bridge).
+ */
+ public EventMeshFrame requestFrame(String topic, EventMeshFrame frame, long timeoutMs) throws Exception {
+ CloudEvent reply = request(topic, frame.toCloudEvent(), timeoutMs);
+ return EventMeshFrame.fromCloudEvent(reply);
+ }
+
+ /**
+ * Primary reply path: complete the pending request future with a Frame (#5299). Round-trips
+ * through CloudEvent for the same reason as {@link #requestFrame(String, EventMeshFrame, long)}.
+ */
+ public boolean replyFrame(String correlationId, EventMeshFrame replyFrame) {
+ return reply(correlationId, replyFrame.toCloudEvent());
+ }
+
+ private static String readCorrelationId(CloudEvent event) {
+ Object value = event.getExtension(EXT_CORRELATION_ID);
+ return value == null ? null : value.toString();
+ }
+
+ private PushChannel channelFor(String clientId) {
+ return channels.computeIfAbsent(clientId, id -> new LongPollingChannel(pushService, id));
+ }
+
+ /** Per-topic monotonic delivery sequence (EventMesh's logical offset for the dispatcher). */
+ private long nextOffset(String topic) {
+ return topicOffsetSeq.computeIfAbsent(topic, k -> new AtomicLong()).incrementAndGet();
+ }
+
+ private DeadLetterSink deadLetterSink() {
+ return (originalTopic, event, reason, attempts) -> {
+ String dlqTopic = originalTopic + "_DLQ";
+ // Issue #5292: the dispatcher retires the delivery only once this future reports the
+ // DLQ write as durably recorded by the storage plugin.
+ java.util.concurrent.CompletableFuture<Boolean> future = new java.util.concurrent.CompletableFuture<>();
+ try {
+ // event is already an EventMeshFrame (internal); store it directly to the DLQ topic.
+ storage.send(dlqTopic, event, new SendCallback() {
+ @Override
+ public void onSuccess(SendResult sendResult) {
+ log.info("event {} dead-lettered to {} after {} attempts: {}",
+ event.attributes().get("id"), dlqTopic, attempts, reason);
+ future.complete(Boolean.TRUE);
+ }
+
+ @Override
+ public void onException(org.apache.eventmesh.api.exception.OnExceptionContext context) {
+ log.error("failed to write DLQ event {} to {}", event.attributes().get("id"), dlqTopic, context.getException());
+ future.complete(Boolean.FALSE);
+ }
+ });
+ } catch (Exception e) {
+ log.error("failed to send DLQ event {} to {}", event.attributes().get("id"), dlqTopic, e);
+ future.complete(Boolean.FALSE);
+ }
+ return future;
+ };
+ }
+}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/AclFilter.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/AclFilter.java
index aa7d8ba9e..30a68887f 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/AclFilter.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/AclFilter.java
@@ -17,12 +17,12 @@
package org.apache.eventmesh.runtime.security;
+import org.apache.eventmesh.common.wire.EventMeshFrame;
+
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
-import io.cloudevents.CloudEvent;
-
/**
* Topic-level authorization (§13.4.2). Holds an {@link AclRule} set sorted by priority (descending,
* DENY wins ties) and matches each request against it: the first matching rule's effect applies,
@@ -58,13 +58,23 @@ public class AclFilter implements IngressFilter {
this.rules = sorted;
}
+ /**
+ * Tenant / clientId come from {@code ctx} (set by the HTTP handler before the filter chain).
+ * We no longer read them from the event because that path was CloudEvent-specific; sub-PR
+ * B keeps the contract simple by reading the principal from the context and only using the
+ * frame to confirm an event-shaped payload arrived.
+ */
@Override
- public FilterVerdict check(CloudEvent event, FilterContext ctx) {
+ public FilterVerdict check(EventMeshFrame frame, FilterContext ctx) {
String principal = ctx.getTenant() != null ? ctx.getTenant() : ctx.getClientId();
String resource = ctx.getTopic();
if (principal == null || resource == null) {
return FilterVerdict.deny(FilterVerdict.STATUS_FORBIDDEN, "no principal/resource for ACL");
}
+ if (frame != null && !frame.isEvent()) {
+ return FilterVerdict.deny(FilterVerdict.STATUS_FORBIDDEN,
+ "ACL applies to EVENT frames only (got msgType=" + frame.msgType() + ")");
+ }
// action not yet carried in FilterContext — pass null so rule action doesn't restrict (any matches).
for (AclRule rule : rules) {
if (rule.matches(principal, resource, null)) {
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/FilterChain.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/FilterChain.java
index 270db3905..be9c56f1c 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/FilterChain.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/FilterChain.java
@@ -17,13 +17,13 @@
package org.apache.eventmesh.runtime.security;
+import org.apache.eventmesh.common.wire.EventMeshFrame;
+
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
-import io.cloudevents.CloudEvent;
-
/**
* Ordered ingress security pipeline (§4.5): AuthFilter → AclFilter → SignatureVerifier, etc.
*
@@ -46,10 +46,22 @@ public class FilterChain {
/**
* Run every filter; return the first denying verdict, or {@link FilterVerdict#allow()} if all
* pass.
+ *
+ * @deprecated since #5299 — call {@link #check(EventMeshFrame, FilterContext)} instead. Kept
+ * as a bridge for the TCP path until sub-PR C migrates.
+ */
+ @Deprecated
+ public FilterVerdict check(io.cloudevents.CloudEvent event, FilterContext ctx) {
+ return check(EventMeshFrame.fromCloudEvent(event), ctx);
+ }
+
+ /**
+ * Run every filter on the runtime's internal frame; return the first denying verdict, or
+ * {@link FilterVerdict#allow()} if all pass. This is the primary ingress path since #5299.
*/
- public FilterVerdict check(CloudEvent event, FilterContext ctx) {
+ public FilterVerdict check(EventMeshFrame frame, FilterContext ctx) {
for (IngressFilter filter : filters) {
- FilterVerdict verdict = filter.check(event, ctx);
+ FilterVerdict verdict = filter.check(frame, ctx);
if (!verdict.isAllowed()) {
return verdict;
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/IngressFilter.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/IngressFilter.java
index 573b6828e..e4193f121 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/IngressFilter.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/IngressFilter.java
@@ -17,18 +17,37 @@
package org.apache.eventmesh.runtime.security;
+import org.apache.eventmesh.common.wire.EventMeshFrame;
+
import io.cloudevents.CloudEvent;
/**
* One stage of the ingress security pipeline (§4.5). Implementations: authentication (who are
* you — {@code TokenAuthFilter}), authorization (what may you do — {@code AclFilter}), signature
* verification ({@code SignatureVerifierFilter}). TLS / mTLS is enforced at the transport, not here.
+ *
+ * <p>Filters operate on the runtime's internal wire format ({@link EventMeshFrame}) since
+ * #5299; the legacy {@code CloudEvent} overload is retained as a bridge for code paths that have
+ * not yet migrated (notably the TCP ingress in sub-PR C). Implementations should override the
+ * {@code EventMeshFrame} variant; the {@code CloudEvent} variant is implemented as a default
+ * that delegates via {@code frame.toCloudEvent()} so existing custom filters keep working.</p>
*/
-@FunctionalInterface
public interface IngressFilter {
/**
* Decide whether {@code event} from {@code ctx} may proceed.
+ *
+ * @deprecated since #5299 — override {@link #check(EventMeshFrame, FilterContext)} instead.
+ * Will be removed once all ingress paths (HTTP, TCP, A2A) emit {@link EventMeshFrame}.
+ */
+ @Deprecated
+ default FilterVerdict check(CloudEvent event, FilterContext ctx) {
+ return check(EventMeshFrame.fromCloudEvent(event), ctx);
+ }
+
+ /**
+ * Decide whether {@code frame} from {@code ctx} may proceed. Default implementation reads
+ * tenant / signature / token directly from {@code frame.attributes()}.
*/
- FilterVerdict check(CloudEvent event, FilterContext ctx);
+ FilterVerdict check(EventMeshFrame frame, FilterContext ctx);
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/SignatureVerifierFilter.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/SignatureVerifierFilter.java
index df73ee864..34dca2d40 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/SignatureVerifierFilter.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/SignatureVerifierFilter.java
@@ -17,14 +17,15 @@
package org.apache.eventmesh.runtime.security;
+import org.apache.eventmesh.common.wire.EventMeshFrame;
+
import java.nio.charset.StandardCharsets;
import java.util.Locale;
+import java.util.Map;
import javax.crypto.Mac;
import javax.crypto.spec.SecretKeySpec;
-import io.cloudevents.CloudEvent;
-
/**
* Verifies an HMAC-SHA256 signature over a canonical projection of the CloudEvent (§13.4.4), so the
* receiver can detect tampering and assert provenance. The signature travels in the
@@ -43,13 +44,16 @@ public class SignatureVerifierFilter implements IngressFilter {
}
@Override
- public FilterVerdict check(CloudEvent event, FilterContext ctx) {
- Object provided = event.getExtension(EXT_SIGNATURE);
- if (!(provided instanceof String)) {
+ public FilterVerdict check(EventMeshFrame frame, FilterContext ctx) {
+ // The signature travels in the frame attributes under the same key the legacy CloudEvent
+ // extension used, so a signed CE-JSON payload becomes a signed frame automatically after
+ // the cloudevents FrameAdaptor round-trip.
+ String provided = frame.attributes().get(EXT_SIGNATURE);
+ if (provided == null) {
return FilterVerdict.deny(FilterVerdict.STATUS_UNAUTHENTICATED, "missing signature");
}
- String expected = sign(canonical(event));
- if (constantTimeEquals(expected, (String) provided)) {
+ String expected = sign(canonical(frame));
+ if (constantTimeEquals(expected, provided)) {
return FilterVerdict.allow();
}
return FilterVerdict.deny(FilterVerdict.STATUS_UNAUTHENTICATED, "signature mismatch");
@@ -57,7 +61,7 @@ public class SignatureVerifierFilter implements IngressFilter {
/**
* Compute the signature over {@code message} — also used by clients/tests to produce the value
- * placed in the {@code emsignature} extension.
+ * placed in the {@code emsignature} extension/attribute.
*/
public String sign(String message) {
try {
@@ -70,11 +74,12 @@ public class SignatureVerifierFilter implements IngressFilter {
}
}
- static String canonical(CloudEvent event) {
- Object id = event.getId();
- Object source = event.getSource();
- Object type = event.getType();
- return String.valueOf(id) + "|" + String.valueOf(source) + "|" + String.valueOf(type);
+ static String canonical(EventMeshFrame frame) {
+ // Same projection as before (#5299 sub-PR B): id|source|type, all read from frame
+ // attributes. For non-EVENT frames the canonical string still computes (we may want to
+ // tighten to isEvent() in a follow-up if streaming chunks ever need sigs).
+ Map<String, String> a = frame.attributes();
+ return a.getOrDefault("id", "") + "|" + a.getOrDefault("source", "") + "|" + a.getOrDefault("type", "");
}
private static String toHex(byte[] bytes) {
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/TokenAuthFilter.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/TokenAuthFilter.java
index 2f44f44bb..62ae9b342 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/TokenAuthFilter.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/security/TokenAuthFilter.java
@@ -17,11 +17,11 @@
package org.apache.eventmesh.runtime.security;
+import org.apache.eventmesh.common.wire.EventMeshFrame;
+
import java.util.Collections;
import java.util.Set;
-import io.cloudevents.CloudEvent;
-
/**
* Authenticates the caller by a bearer token (§13.4.1). In production this delegates to the
* existing security-plugin ({@code auth-token} / {@code auth-http-basic}); the uni skeleton
@@ -36,7 +36,7 @@ public class TokenAuthFilter implements IngressFilter {
}
@Override
- public FilterVerdict check(CloudEvent event, FilterContext ctx) {
+ public FilterVerdict check(EventMeshFrame frame, FilterContext ctx) {
String credential = ctx.getCredential();
if (credential != null && validTokens.contains(credential)) {
return FilterVerdict.allow();
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/CloudEventToPackageBody.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/CloudEventToPackageBody.java
deleted file mode 100644
index f92a3682f..000000000
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/CloudEventToPackageBody.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.eventmesh.runtime.transport.tcp;
-
-import io.cloudevents.CloudEvent;
-
-/**
- * Adapts a delivered CloudEvent into the legacy TCP {@code Package} body the old client expects
- * (egress direction). Production uses {@code MeshMessageProtocolAdaptor.fromCloudEvent(...)};
- * tests inject a stub (e.g. event → a Map the test can read back).
- */
-@FunctionalInterface
-public interface CloudEventToPackageBody {
-
- Object toBody(CloudEvent event);
-}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/MeshEventToPackageBody.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/MeshEventToPackageBody.java
deleted file mode 100644
index 3c13c620d..000000000
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/MeshEventToPackageBody.java
+++ /dev/null
@@ -1,50 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You under the Apache License, Version 2.0
- * (the "License"); you may not use this file except in compliance with
- * the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.eventmesh.runtime.transport.tcp;
-
-import org.apache.eventmesh.common.protocol.tcp.EventMeshMessage;
-
-import java.nio.charset.StandardCharsets;
-
-import io.cloudevents.CloudEvent;
-
-/**
- * Egress encoder for the TCP push frame: builds the legacy {@code EventMeshMessage} body the client
- * expects (topic ← CloudEvent subject, content ← CloudEvent data). The {@link NettyTcpPushChannel}
- * wraps this body in an {@code ASYNC_MESSAGE_TO_CLIENT} Package; the netty {@code Codec} serializes
- * it onto the wire.
- *
- * <p>This builds the body directly rather than routing through {@code MeshMessageProtocolAdaptor}.
- * fromCloudEvent, because that adaptor's protocol-desc switching requires CloudEvents to carry
- * legacy protocol metadata extensions that push frames don't naturally have; the wire payload a
- * legacy TCP subscriber receives is the {@code EventMeshMessage} anyway.</p>
- */
-public class MeshEventToPackageBody implements CloudEventToPackageBody {
-
- @Override
- public Object toBody(CloudEvent event) {
- EventMeshMessage message = new EventMeshMessage();
- if (event.getSubject() != null) {
- message.setTopic(event.getSubject());
- }
- if (event.getData() != null) {
- message.setBody(new String(event.getData().toBytes(), StandardCharsets.UTF_8));
- }
- return message;
- }
-}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/TcpFrameCodec.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/TcpFrameCodec.java
index b925b8dfa..dabb615d4 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/TcpFrameCodec.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/TcpFrameCodec.java
@@ -17,22 +17,23 @@
package org.apache.eventmesh.runtime.transport.tcp;
-import io.cloudevents.CloudEvent;
+import org.apache.eventmesh.common.wire.EventMeshFrame;
/**
- * Encodes a delivered CloudEvent (plus its delivery id) into the TCP {@code Package} wire bytes the
- * legacy client expects, and decodes a client ACK frame back into the delivery id it acknowledges.
+ * Encodes a delivered {@link EventMeshFrame} (plus its delivery id) into the TCP {@code Package}
+ * wire bytes the legacy client expects, and decodes a client ACK frame back into the delivery id it
+ * acknowledges.
*
- * <p>Production implementation reuses the existing {@code Codec} + {@code MeshMessageProtocolAdaptor}
- * (reverse direction) and carries the delivery id in a Package header/extension so the client's ACK
- * frame echoes it. Tests inject a deterministic stub.</p>
+ * <p>Production implementation reuses the existing {@code Codec} + the {@code meshmessage}
+ * {@code FrameAdaptor} (reverse direction) and carries the delivery id in a Package header/extension
+ * so the client's ACK frame echoes it. Tests inject a deterministic stub.</p>
*/
public interface TcpFrameCodec {
/**
- * Encode a push frame for {@code event}, tagged with {@code deliveryId} so the client can ACK it.
+ * Encode a push frame for {@code frame}, tagged with {@code deliveryId} so the client can ACK it.
*/
- byte[] encodePush(String deliveryId, CloudEvent event);
+ byte[] encodePush(String deliveryId, EventMeshFrame frame);
/**
* Extract the delivery id from a client ACK frame, or {@code null} if the frame isn't an ACK.
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/TcpPushChannel.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/TcpPushChannel.java
index 5c8836fba..68e4d489c 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/TcpPushChannel.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/TcpPushChannel.java
@@ -21,19 +21,18 @@ import org.apache.eventmesh.common.wire.EventMeshFrame;
import org.apache.eventmesh.runtime.delivery.AckCallback;
import org.apache.eventmesh.runtime.delivery.PushChannel;
-import io.cloudevents.CloudEvent;
-
import lombok.extern.slf4j.Slf4j;
/**
* A legacy TCP client session as a {@link PushChannel} (egress side of the compatibility bridge).
*
* <p>When the new {@code ReliableDispatcher} picks a TCP subscriber as a target, it hands the
- * CloudEvent here; this channel encodes it into the legacy {@code Package} wire format (via
- * {@link TcpFrameCodec}), writes it to the socket ({@code TcpSessionSink}), and parks the ACK
- * callback in {@link TcpAckRegistry} until the client's ACK frame arrives. Reliability (redelivery,
- * DLQ) is therefore shared with every other transport — the TCP client just looks like another
- * push target to the core.</p>
+ * {@link EventMeshFrame} here (the internal wire unit — no CloudEvent intermediary since #5299);
+ * this channel encodes it into the legacy {@code Package} wire format (via {@link TcpFrameCodec}),
+ * writes it to the socket ({@code TcpSessionSink}), and parks the ACK callback in
+ * {@link TcpAckRegistry} until the client's ACK frame arrives. Reliability (redelivery, DLQ) is
+ * therefore shared with every other transport — the TCP client just looks like another push target
+ * to the core.</p>
*/
@Slf4j
public class TcpPushChannel implements PushChannel {
@@ -57,20 +56,12 @@ public class TcpPushChannel implements PushChannel {
@Override
public void deliver(String deliveryId, EventMeshFrame event, AckCallback callback) {
- // Egress boundary: convert the internal Frame to a CloudEvent for the legacy TCP wire format.
- CloudEvent ce;
- try {
- ce = event.toCloudEvent();
- } catch (RuntimeException e) {
- log.warn("tcp push frame->CloudEvent conversion failed for delivery={}", deliveryId, e);
- callback.nack(e);
- return;
- }
+ // Egress boundary: encode the internal Frame straight onto the legacy TCP wire format.
byte[] frame;
try {
- frame = codec.encodePush(deliveryId, ce);
+ frame = codec.encodePush(deliveryId, event);
} catch (RuntimeException e) {
- log.warn("tcp push encode failed for delivery={}", deliveryId, e);
+ log.warn("tcp push frame->wire encode failed for delivery={}", deliveryId, e);
callback.nack(e);
return;
}
diff --git a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/UniTcpServer.java b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/UniTcpServer.java
index a9988e510..8a5d471bc 100644
--- a/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/UniTcpServer.java
+++ b/eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/transport/tcp/UniTcpServer.java
@@ -60,19 +60,16 @@ public class UniTcpServer {
private final UniIngressService ingress;
private final TcpAckRegistry ackRegistry;
private final PackageRouter router;
- private final CloudEventToPackageBody bodyMapper;
private EventLoopGroup bossGroup;
private EventLoopGroup workerGroup;
private Channel serverChannel;
private final ConcurrentHashMap<String, Channel> clientChannels = new ConcurrentHashMap<>();
- public UniTcpServer(UniIngressService ingress, TcpAckRegistry ackRegistry, PackageRouter router,
- CloudEventToPackageBody bodyMapper) {
+ public UniTcpServer(UniIngressService ingress, TcpAckRegistry ackRegistry, PackageRouter router) {
this.ingress = ingress;
this.ackRegistry = ackRegistry;
this.router = router;
- this.bodyMapper = bodyMapper;
}
/**
@@ -93,7 +90,7 @@ public class UniTcpServer {
ch.pipeline()
.addLast(new Codec.Encoder())
.addLast(new Codec.Decoder())
- .addLast(new FrameHandler(ingress, ackRegistry, router, clientChannels, bodyMapper));
+ .addLast(new FrameHandler(ingress, ackRegistry, router, clientChannels));
}
});
serverChannel = bootstrap.bind(port).sync().channel();
@@ -119,7 +116,9 @@ public class UniTcpServer {
* UNSUBSCRIBE/GOODBYE) are handled here directly because they need channel context (the
* subscriber's clientId comes from the HELLO {@link UserAgent#getGroup()}, not the SUBSCRIBE
* body). Message commands (ASYNC_MESSAGE_TO_SERVER / ASYNC_MESSAGE_TO_CLIENT_ACK) go through the
- * {@link PackageRouter} for CloudEvents translation. Static + package-private so it can be
+ * {@link PackageRouter}, which decodes the MeshMessage straight into the internal
+ * {@link org.apache.eventmesh.common.wire.EventMeshFrame} wire unit — no CloudEvent hop since
+ * #5299. Static + package-private so it can be
* exercised directly via netty {@code EmbeddedChannel} in tests.
*/
static final class FrameHandler extends SimpleChannelInboundHandler<Package> {
@@ -131,15 +130,13 @@ public class UniTcpServer {
private final TcpAckRegistry ackRegistry;
private final PackageRouter router;
private final ConcurrentHashMap<String, Channel> clientChannels;
- private final CloudEventToPackageBody bodyMapper;
FrameHandler(UniIngressService ingress, TcpAckRegistry ackRegistry, PackageRouter router,
- ConcurrentHashMap<String, Channel> clientChannels, CloudEventToPackageBody bodyMapper) {
+ ConcurrentHashMap<String, Channel> clientChannels) {
this.ingress = ingress;
this.ackRegistry = ackRegistry;
this.router = router;
this.clientChannels = clientChannels;
- this.bodyMapper = bodyMapper;
}
@Override
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/LegacyTcpClientIntegrationTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/LegacyTcpClientIntegrationTest.java
index a484bd445..c2babba48 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/LegacyTcpClientIntegrationTest.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/LegacyTcpClientIntegrationTest.java
@@ -33,7 +33,6 @@ import org.apache.eventmesh.common.protocol.tcp.UserAgent;
import org.apache.eventmesh.common.wire.EventMeshFrame;
import org.apache.eventmesh.runtime.ingress.UniIngressService;
import org.apache.eventmesh.runtime.offset.InMemoryOffsetStore;
-import org.apache.eventmesh.runtime.transport.tcp.MeshEventToPackageBody;
import org.apache.eventmesh.runtime.transport.tcp.MeshMessagePackageRouter;
import org.apache.eventmesh.runtime.transport.tcp.TcpAckRegistry;
import org.apache.eventmesh.runtime.transport.tcp.UniTcpServer;
@@ -159,8 +158,7 @@ class LegacyTcpClientIntegrationTest {
}
}, 0, 100, java.util.concurrent.TimeUnit.MILLISECONDS);
- server = new UniTcpServer(ingress, new TcpAckRegistry(), new MeshMessagePackageRouter(),
- new MeshEventToPackageBody());
+ server = new UniTcpServer(ingress, new TcpAckRegistry(), new MeshMessagePackageRouter());
port = server.start(0);
}
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/LegacyTcpClusterBrokerIntegrationTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/LegacyTcpClusterBrokerIntegrationTest.java
index a466b233b..b292b1ed4 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/LegacyTcpClusterBrokerIntegrationTest.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/it/LegacyTcpClusterBrokerIntegrationTest.java
@@ -31,7 +31,6 @@ import org.apache.eventmesh.common.protocol.tcp.UserAgent;
import org.apache.eventmesh.runtime.boot.EventMeshApplication;
import org.apache.eventmesh.runtime.cluster.NacosMetaStore;
import org.apache.eventmesh.runtime.offset.InMemoryOffsetStore;
-import org.apache.eventmesh.runtime.transport.tcp.MeshEventToPackageBody;
import org.apache.eventmesh.runtime.transport.tcp.MeshMessagePackageRouter;
import org.apache.eventmesh.runtime.transport.tcp.TcpAckRegistry;
import org.apache.eventmesh.runtime.transport.tcp.UniTcpServer;
@@ -119,7 +118,7 @@ class LegacyTcpClusterBrokerIntegrationTest {
// 3. TCP server (not auto-booted by EventMeshApplication) on the cluster-enabled ingress.
tcpServer = new UniTcpServer(app.runtime().ingress(), new TcpAckRegistry(),
- new MeshMessagePackageRouter(), new MeshEventToPackageBody());
+ new MeshMessagePackageRouter());
tcpPort = tcpServer.start(0);
// 4. Subscriber: real old SDK. clientId = HELLO UserAgent.group.
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/security/EventMeshFrameFilterTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/security/EventMeshFrameFilterTest.java
new file mode 100644
index 000000000..4668e4677
--- /dev/null
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/security/EventMeshFrameFilterTest.java
@@ -0,0 +1,145 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.runtime.security;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.apache.eventmesh.common.wire.EventMeshFrame;
+
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+import org.junit.jupiter.api.Test;
+
+/**
+ * #5299 sub-PR B: the filter chain now operates on {@link EventMeshFrame} instead of
+ * {@code io.cloudevents.CloudEvent}. These tests exercise the new primary path; the
+ * CloudEvent-based tests in {@link SecurityFilterTest} remain as the legacy bridge contract.
+ */
+class EventMeshFrameFilterTest {
+
+ @Test
+ void tokenAuthReadsCredentialFromContextNotFrame() {
+ TokenAuthFilter auth = new TokenAuthFilter(java.util.Collections.singleton("good-token"));
+
+ assertTrue(auth.check(frame(), ctx("good-token", "tenantA", "orders")).isAllowed());
+ FilterVerdict missing = auth.check(frame(), ctx(null, "tenantA", "orders"));
+ assertFalse(missing.isAllowed());
+ assertEquals(401, missing.getRejectStatus());
+ }
+
+ @Test
+ void aclDeniesFrameWithNonEventMsgType() {
+ // A STREAM_REQ frame should be rejected by the ACL filter even if the principal/resource
+ // are otherwise fine — the filter applies to event ingress only. We craft the frame via
+ // decode() because the 5-arg ctor is package-private; the wire format is documented in
+ // EventMeshFrame (header = magic(1) | ver(1) | msgType(1) | flags(1) | seq(4) | keyCount(2) | dataLen(4)).
+ byte[] streamReqBytes = new byte[] {
+ (byte) 0xEF, 1, EventMeshFrame.TYPE_STREAM_REQ, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0
+ };
+ EventMeshFrame streamFrame = EventMeshFrame.decode(streamReqBytes);
+ AclFilter acl = new AclFilter(java.util.List.of(
+ new AclRule("tenantA", "orders", AclRule.Action.ANY, AclRule.Effect.ALLOW, 10)));
+ FilterVerdict denied = acl.check(streamFrame, ctx("tok", "tenantA", "orders"));
+ assertFalse(denied.isAllowed());
+ assertEquals(403, denied.getRejectStatus());
+ }
+
+ @Test
+ void aclAllowsGrantedTopicRejectsForbidden() {
+ AclFilter acl = new AclFilter(java.util.List.of(
+ new AclRule("tenantA", "orders", AclRule.Action.ANY, AclRule.Effect.ALLOW, 10)));
+
+ assertTrue(acl.check(frame(), ctx("tok", "tenantA", "orders")).isAllowed());
+ FilterVerdict forbidden = acl.check(frame(), ctx("tok", "tenantA", "payments"));
+ assertFalse(forbidden.isAllowed());
+ assertEquals(403, forbidden.getRejectStatus());
+ }
+
+ @Test
+ void chainFailsClosedAtFirstDenyingFilter() {
+ TokenAuthFilter auth = new TokenAuthFilter(java.util.Collections.singleton("good-token"));
+ AclFilter acl = new AclFilter(java.util.List.of(
+ new AclRule("tenantA", "orders", AclRule.Action.ANY, AclRule.Effect.ALLOW, 10)));
+ FilterChain chain = new FilterChain(auth, acl);
+
+ // Bad credential → denied by auth (401), acl never consulted.
+ FilterVerdict badCred = chain.check(frame(), ctx("wrong", "tenantA", "orders"));
+ assertFalse(badCred.isAllowed());
+ assertEquals(401, badCred.getRejectStatus());
+
+ // Good credential but forbidden topic → denied by acl (403).
+ FilterVerdict badTopic = chain.check(frame(), ctx("good-token", "tenantA", "payments"));
+ assertFalse(badTopic.isAllowed());
+ assertEquals(403, badTopic.getRejectStatus());
+
+ // Both pass → allow.
+ assertTrue(chain.check(frame(), ctx("good-token", "tenantA", "orders")).isAllowed());
+ }
+
+ @Test
+ void signatureVerifierReadsSignatureFromFrameAttributes() {
+ SignatureVerifierFilter verifier = new SignatureVerifierFilter("shared-secret");
+ EventMeshFrame unsigned = frame();
+ String goodSig = verifier.sign(SignatureVerifierFilter.canonical(unsigned));
+ EventMeshFrame signed = withAttr(unsigned, SignatureVerifierFilter.EXT_SIGNATURE, goodSig);
+
+ assertTrue(verifier.check(signed, ctx("tok", "tenantA", "orders")).isAllowed());
+
+ // Tampered signature (flip last hex char) → reject.
+ String tampered = goodSig.substring(0, goodSig.length() - 1)
+ + (goodSig.charAt(goodSig.length() - 1) == '0' ? '1' : '0');
+ EventMeshFrame bad = withAttr(signed, SignatureVerifierFilter.EXT_SIGNATURE, tampered);
+ assertFalse(verifier.check(bad, ctx("tok", "tenantA", "orders")).isAllowed());
+
+ // Missing signature → reject.
+ assertFalse(verifier.check(unsigned, ctx("tok", "tenantA", "orders")).isAllowed(),
+ "missing signature rejected");
+ }
+
+ @Test
+ void tenantFromFrameAttributesReachesContext() {
+ // The HTTP handler now reads tenant from frame.attributes() and threads it into
+ // FilterContext before the chain runs; this test pins that contract.
+ EventMeshFrame f = withAttr(frame(), "emtenantid", "tenantZ");
+ String tenant = f.attributes().get("emtenantid");
+ assertEquals("tenantZ", tenant);
+ // The downstream AclFilter then uses ctx.tenant — the test in aclAllowsGrantedTopicRejectsForbidden
+ // already exercises the "tenant from ctx" path; here we just document the contract.
+ }
+
+ private static EventMeshFrame frame() {
+ Map<String, String> attrs = new LinkedHashMap<>();
+ attrs.put("id", "e-1");
+ attrs.put("source", "svc");
+ attrs.put("type", "order.created");
+ return EventMeshFrame.event(attrs, new byte[0]);
+ }
+
+ private static EventMeshFrame withAttr(EventMeshFrame base, String name, String value) {
+ Map<String, String> attrs = new LinkedHashMap<>(base.attributes());
+ attrs.put(name, value);
+ return EventMeshFrame.event(attrs, base.data());
+ }
+
+ private static FilterContext ctx(String credential, String tenant, String topic) {
+ return new FilterContext(topic, "client-1", tenant, credential, "127.0.0.1");
+ }
+}
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/security/SecurityFilterTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/security/SecurityFilterTest.java
index 3993c0c7e..116854653 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/security/SecurityFilterTest.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/security/SecurityFilterTest.java
@@ -21,6 +21,8 @@ import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
+import org.apache.eventmesh.common.wire.EventMeshFrame;
+
import java.net.URI;
import java.util.Collections;
@@ -94,7 +96,7 @@ class SecurityFilterTest {
SignatureVerifierFilter verifier = new SignatureVerifierFilter("shared-secret");
CloudEvent event = event();
- String goodSig = verifier.sign(SignatureVerifierFilter.canonical(event));
+ String goodSig = verifier.sign(SignatureVerifierFilter.canonical(EventMeshFrame.fromCloudEvent(event)));
CloudEvent signed = CloudEventBuilder.from(event).withExtension(SignatureVerifierFilter.EXT_SIGNATURE, goodSig).build();
assertTrue(verifier.check(signed, ctx("tok", "tenantA", "orders")).isAllowed());
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/transport/tcp/MeshMessagePackageRouterTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/transport/tcp/MeshMessagePackageRouterTest.java
index fee328105..bfa79a95f 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/transport/tcp/MeshMessagePackageRouterTest.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/transport/tcp/MeshMessagePackageRouterTest.java
@@ -25,6 +25,8 @@ import org.apache.eventmesh.common.protocol.tcp.Command;
import org.apache.eventmesh.common.protocol.tcp.EventMeshMessage;
import org.apache.eventmesh.common.protocol.tcp.Header;
import org.apache.eventmesh.common.protocol.tcp.Package;
+import org.apache.eventmesh.common.wire.EventMeshFrame;
+import org.apache.eventmesh.protocol.meshmessage.MeshMessageFrameAdaptor;
import java.net.URI;
import java.nio.charset.StandardCharsets;
@@ -42,14 +44,16 @@ import io.cloudevents.core.builder.CloudEventBuilder;
class MeshMessagePackageRouterTest {
@Test
- void egressEncodesCloudEventIntoEventMeshMessageBody() {
+ void egressEncodesFrameIntoEventMeshMessageBody() throws Exception {
CloudEvent event = CloudEventBuilder.v1()
.withId("e-1").withSource(URI.create("svc")).withType("order.created")
.withSubject("orders")
.withData("hello".getBytes(StandardCharsets.UTF_8))
.build();
+ EventMeshFrame frame = EventMeshFrame.fromCloudEvent(event);
- Object body = new MeshEventToPackageBody().toBody(event);
+ // Egress is frame-native since #5299: Frame -> Package(EventMeshMessage), no CloudEvent hop.
+ Object body = ((Package) new MeshMessageFrameAdaptor().fromFrame(frame)).getBody();
assertNotNull(body, "egress body must be produced");
assertTrue(body instanceof EventMeshMessage, "body is the legacy EventMeshMessage");
@@ -70,9 +74,9 @@ class MeshMessagePackageRouterTest {
assertEquals("d-99", req.getDeliveryId());
}
- // NOTE: publish ingress (ASYNC_MESSAGE_TO_SERVER → CloudEvent) routes through
- // MeshMessageProtocolAdaptor.toCloudEvent, which is itself covered by eventmesh-protocol-
- // meshmessage's own test suite with real wire packages (body=JSON string + protocol header
- // properties produced by the Codec). The in-JVM object/string asymmetry makes a direct
- // round-trip unit test unrepresentative, so it is intentionally not asserted here.
+ // NOTE: publish ingress (ASYNC_MESSAGE_TO_SERVER → EventMeshFrame) routes through
+ // MeshMessageFrameAdaptor, which is itself covered by eventmesh-protocol-meshmessage's own
+ // test suite with real wire packages (body=JSON string + protocol header properties produced
+ // by the Codec). The in-JVM object/string asymmetry makes a direct round-trip unit test
+ // unrepresentative, so it is intentionally not asserted here.
}
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/transport/tcp/TcpCompatibilityBridgeTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/transport/tcp/TcpCompatibilityBridgeTest.java
index 2e99641bf..e0d8ae5db 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/transport/tcp/TcpCompatibilityBridgeTest.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/transport/tcp/TcpCompatibilityBridgeTest.java
@@ -111,8 +111,8 @@ class TcpCompatibilityBridgeTest {
private static final class StubCodec implements TcpFrameCodec {
@Override
- public byte[] encodePush(String deliveryId, CloudEvent event) {
- return ("PUSH:" + deliveryId + ":" + event.getId()).getBytes(StandardCharsets.UTF_8);
+ public byte[] encodePush(String deliveryId, EventMeshFrame frame) {
+ return ("PUSH:" + deliveryId + ":" + frame.attributes().get("id")).getBytes(StandardCharsets.UTF_8);
}
@Override
diff --git a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/transport/tcp/UniTcpServerTest.java b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/transport/tcp/UniTcpServerTest.java
index 425bbf059..034f9cbe5 100644
--- a/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/transport/tcp/UniTcpServerTest.java
+++ b/eventmesh-runtime/src/test/java/org/apache/eventmesh/runtime/transport/tcp/UniTcpServerTest.java
@@ -98,15 +98,8 @@ class UniTcpServerTest {
}
return null;
};
- // Body mapper: event → a Map the "client" (test) can read back.
- CloudEventToPackageBody bodyMapper = event -> {
- Map<String, Object> b = new HashMap<>();
- b.put("id", event.getId());
- return b;
- };
-
EmbeddedChannel client = new EmbeddedChannel(new UniTcpServer.FrameHandler(
- ingress, new TcpAckRegistry(), router, new ConcurrentHashMap<>(), bodyMapper));
+ ingress, new TcpAckRegistry(), router, new ConcurrentHashMap<>()));
// 1a. client HELLO (carries clientId in UserAgent.group) → server stashes it on the channel.
UserAgent ua = UserAgent.builder().group("c1").host("test").port(1).build();
@@ -149,7 +142,7 @@ class UniTcpServerTest {
private static UniTcpServer.FrameHandler newHandler(UniIngressService ingress, PackageRouter router) {
return new UniTcpServer.FrameHandler(ingress, new TcpAckRegistry(), router,
- new ConcurrentHashMap<>(), event -> null);
+ new ConcurrentHashMap<>());
}
private static final class InMemoryStorage implements MeshStoragePlugin {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@eventmesh.apache.org
For additional commands, e-mail: commits-help@eventmesh.apache.org