24.0.0.2におけるMicroProfile Reactive Messagingの否定応答など

@Incoming("data")
@Outgoing("out")
public Message<String> process(Message<String> m) {
    String s = m.getPayload();
    if (s.equalsIgnoreCase("b")) {
        // we cannot fail, we must nack explicitly.
        m.nack(new IllegalArgumentException("b"));
        return null;
    }
    return m.withPayload(s.toUpperCase());
}

@Incoming("data")
@Outgoing("out")
public String process(String s) {
    if (s.equalsIgnoreCase("b")) {
        throw new IllegalArgumentException("b");
    }
    return s.toUpperCase();
}

@Inject
@Channel(CHANNEL_NAME)
Emitter<String> emitter;

@POST
@Path("/payload")
public CompletionStage<Void> emitPayload(String payload){
    CompletionStage<Void> cs = emitter.send(payload);
    return cs;
}

@POST
@Path("/message")
public CompletionStage<Void> emitPayload(String payload){
    CompletableFuture<Void> ackCf = new CompletableFuture<>();
    emitter.send(Message.of(payload,
        () -> {
            ackCf.complete(null);
            return CompletableFuture.completedFuture(null);
        },
        t -> {
            ackCf.completeExceptionally(t);
            return CompletableFuture.completedFuture(null);
        }));
    return ackCf;
}

@Inject @Channel("myChannel")
@OnOverflow(value=OnOverflow.Strategy.BUFFER, bufferSize=300)
private Emitter<String> emitter;

public void publishMessage() {
  emitter.send("a");
  emitter.send("b");
  emitter.complete();
}