Skip to content

Support event-bus interceptor chain failure. #6257

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
vietj merged 1 commit into from
Jul 24, 2026
Merged

Support event-bus interceptor chain failure. #6257

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 15 additions & 0 deletions vertx-core/src/main/java/io/vertx/core/eventbus/DeliveryContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,21 @@ public interface DeliveryContext<T> {
*/
void next();

/**
* {@link #fail(ReplyFailure, int, String)} with {@link ReplyFailure#RECIPIENT_FAILURE}
*/
boolean fail(int failureCode, String message);

/**
* Terminate the interception chain immediately and fail the message.
*
* @param failure the failure
* @param failureCode the failure code
* @param message the message
* @return true if the operation succeeded
*/
boolean fail(ReplyFailure failure, int failureCode, String message);

/**
* @return true if the message is being sent (point to point) or False if the message is being published
*/
Expand Down
28 changes: 25 additions & 3 deletions vertx-core/src/main/java/io/vertx/core/eventbus/impl/DeliveryContextImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,8 @@
import io.vertx.core.Handler;
import io.vertx.core.eventbus.DeliveryContext;
import io.vertx.core.eventbus.Message;
import io.vertx.core.eventbus.ReplyException;
import io.vertx.core.eventbus.ReplyFailure;
import io.vertx.core.internal.ContextInternal;

import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
Expand All @@ -24,12 +26,12 @@ class DeliveryContextImpl<T> implements DeliveryContext<T> {
private final MessageImpl<?, T> message;
private final ContextInternal context;
private final Object body;
private final Runnable dispatch;
private final Handler<ReplyException> dispatch;
private final Handler<DeliveryContext<?>>[] interceptors;
private volatile int interceptorIdx;

protected DeliveryContextImpl(MessageImpl<?, T> message, Handler<DeliveryContext<?>>[] interceptors,
ContextInternal context, Object body, Runnable dispatch) {
ContextInternal context, Object body, Handler<ReplyException> dispatch) {
this.message = message;
this.interceptors = interceptors;
this.context = context;
Expand All @@ -53,6 +55,26 @@ public Object body() {
return body;
}

@Override
public boolean fail(int failureCode, String message) {
return fail(ReplyFailure.RECIPIENT_FAILURE, failureCode, message);
}

@Override
public boolean fail(ReplyFailure failure, int failureCode, String message) {
while (true) {
int idx = UPDATER.get(this);
if (idx > interceptors.length) {
return false;
}
if (UPDATER.compareAndSet(this, idx, interceptors.length + 1)) {
break;
}
}
dispatch.handle(new ReplyException(failure, failureCode, message));
return true;
}

@Override
public void next() {
int idx = UPDATER.getAndIncrement(this);
Expand All @@ -68,7 +90,7 @@ public void next() {
}
}
} else if (idx == interceptors.length) {
dispatch.run();
dispatch.handle(null);
} else {
throw new IllegalStateException();
}
Expand Down
8 changes: 7 additions & 1 deletion vertx-core/src/main/java/io/vertx/core/eventbus/impl/HandlerRegistration.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -95,7 +95,13 @@ public Future<Void> unregister() {
void dispatchMessage(Function<Message<T>, Future<?>> processor, MessageImpl<?, T> message, ContextInternal context) {
Handler<DeliveryContext<?>>[] interceptors = message.bus.inboundInterceptors();
if (interceptors.length > 0) {
Runnable dispatch = () -> dispatch(context, message, processor);
Handler<ReplyException> dispatch = failure -> {
if (failure == null) {
dispatch(context, message, processor);
} else {
message.reply(failure);
}
};
DeliveryContextImpl<T> deliveryCtx = new DeliveryContextImpl<>(message, interceptors, context, message.receivedBody, dispatch);
deliveryCtx.next();
} else {
Expand Down
43 changes: 26 additions & 17 deletions vertx-core/src/main/java/io/vertx/core/eventbus/impl/SendContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -49,10 +49,10 @@ public class SendContext<T> implements Promise<Void> {
void send() {
Handler<DeliveryContext<?>>[] interceptors = message.bus.outboundInterceptors();
if (interceptors.length > 0) {
DeliveryContextImpl<T> deliveryContext = new DeliveryContextImpl<>(message, interceptors, ctx, message.sentBody, this::sendOrPub);
DeliveryContextImpl<T> deliveryContext = new DeliveryContextImpl<>(message, interceptors, ctx, message.sentBody, this::sendOrPubOrFail);
deliveryContext.next();
} else {
sendOrPub();
sendOrPubOrFail(null);
}
}

Expand Down Expand Up @@ -113,23 +113,32 @@ private void written(Throwable failure) {
}
}

private void sendOrPub() {
VertxTracer tracer = ctx.tracer();
if (tracer != null) {
if (message.trace == null) {
src = true;
BiConsumer<String, String> biConsumer = (String key, String val) -> message.headers().set(key, val);
TracingPolicy tracingPolicy = options.getTracingPolicy();
if (tracingPolicy == null) {
tracingPolicy = TracingPolicy.PROPAGATE;
private void sendOrPubOrFail(ReplyException failure) {
if (failure == null) {
VertxTracer tracer = ctx.tracer();
if (tracer != null) {
if (message.trace == null) {
src = true;
BiConsumer<String, String> biConsumer = (String key, String val) -> message.headers().set(key, val);
TracingPolicy tracingPolicy = options.getTracingPolicy();
if (tracingPolicy == null) {
tracingPolicy = TracingPolicy.PROPAGATE;
}
message.trace = tracer.sendRequest(ctx, SpanKind.RPC, tracingPolicy, message, message.send ? "send" : "publish", biConsumer, MessageTagExtractor.INSTANCE);
} else {
// Handle failure here
tracer.sendResponse(ctx, null, message.trace, null, TagExtractor.empty());
}
message.trace = tracer.sendRequest(ctx, SpanKind.RPC, tracingPolicy, message, message.send ? "send" : "publish", biConsumer, MessageTagExtractor.INSTANCE);
} else {
// Handle failure here
tracer.sendResponse(ctx, null, message.trace, null, TagExtractor.empty());
}
bus.sendOrPub(this);
} else {
// Fail write promise
writePromise.fail(failure);

// Fail reply handler when it exists
if (replyHandler != null) {
replyHandler.fail(failure);
}
}
bus.sendOrPub(this);
}

}
103 changes: 101 additions & 2 deletions vertx-core/src/test/java/io/vertx/tests/eventbus/EventBusInterceptorTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,10 +12,10 @@
package io.vertx.tests.eventbus;

import io.vertx.core.Context;
import io.vertx.core.Future;
import io.vertx.core.Handler;
import io.vertx.core.Vertx;
import io.vertx.core.eventbus.DeliveryContext;
import io.vertx.core.eventbus.EventBus;
import io.vertx.core.eventbus.*;
import io.vertx.test.core.VertxTestBase;
import org.junit.Test;

Expand Down Expand Up @@ -425,6 +425,105 @@ public void testInboundInterceptorFromNonVertxThreadFailure() {
assertSame(expected, caught.get());
}

@Test
public void testOutboundReplySendFailure() {
eb.addOutboundInterceptor(sc -> {
assertTrue(sc.fail(3, "test"));
try {
sc.next();
fail();
} catch (IllegalStateException expected) {
}
});
eb.consumer("some-address", msg -> {
fail();
});
MessageProducer<String> sender = eb.sender("some-address");
Future<Void> fut = sender.write("armadillo");
fut.onComplete(onFailure(expected -> {
testComplete();
}));
await();
}

@Test
public void testOutboundReplyRequestFailure() {
eb.addOutboundInterceptor(sc -> {
assertTrue(sc.fail(ReplyFailure.NO_HANDLERS, 3, "test"));
try {
sc.next();
fail();
} catch (IllegalStateException expected) {
}
});
eb.consumer("some-address", msg -> {
fail();
});
Future<Message<Object>> fut = eb.request("some-address", "armadillo");
fut.onComplete(onFailure(expected -> {
ReplyException replyException = (ReplyException) expected;
assertEquals(3, replyException.failureCode());
assertEquals(ReplyFailure.NO_HANDLERS, replyException.failureType());
testComplete();
}));
await();
}

@Test
public void testInboundReplySendFailure() {
eb.addInboundInterceptor(sc -> {
assertTrue(sc.fail(3, "test"));
try {
sc.next();
fail();
} catch (IllegalStateException expected) {
}
});
eb.consumer("some-address", msg -> {
fail();
});
MessageProducer<String> sender = eb.sender("some-address");
Future<Void> fut = sender.write("armadillo");
fut.onComplete(onSuccess(expected -> {
testComplete();
}));
await();
}

@Test
public void testInboundReplyRequestFailure() {
eb.addInboundInterceptor(dc -> {
if (dc.message().address().equals("some-address")) {
assertTrue(dc.fail(3, "test"));
try {
dc.next();
fail();
} catch (IllegalStateException expected) {
}
} else {
dc.next();
}
});
eb.addInboundInterceptor(dc -> {
if (dc.message().address().equals("some-address")) {
fail();
} else {
dc.next();
}
});
eb.consumer("some-address", msg -> {
fail();
});
Future<Message<Object>> fut = eb.request("some-address", "armadillo");
fut.onComplete(onFailure(expected -> {
ReplyException replyException = (ReplyException) expected;
assertEquals(3, replyException.failureCode());
assertEquals(ReplyFailure.RECIPIENT_FAILURE, replyException.failureType());
testComplete();
}));
await();
}

@Override
public void setUp() throws Exception {
super.setUp();
Expand Down
Loading