Quarkus Vert.x EventBus中通过Uni<Class>返回自定义HTTP状态码

问题背景 #

使用Quarkus 3.9.2 + Vert.x全响应式框架，通过@ConsumeEvent调用EventBus时，服务端通过message.fail()返回错误时，接口始终返回默认500状态码及框架内置错误体，无法返回自定义的HTTP状态码（如400）和自定义响应结构。

当前控制器代码：

@Inject
EventBus bus;

@GET
@Produces(MediaType.APPLICATION_JSON)
@Path("/path/{id}")
public Uni<People> getPeopleById(@PathParam("id") String id){
   JsonObject req = new JsonObject();
   req.put("id", id);
   return bus.<People>request("getPeopleById", req, new DeliveryOptions().setSendTimeout(12000L))
         .onItem().transform(Message::body)
         .ifNoItem().after(Duration.ofMillis(12000L))
         .failWith(new Exception("Error Service"));
}

当前服务端代码：

@ConsumeEvent("getPeopleById")
public void getPeopleById(Message message){
   JsonObject req = (JsonObject) message.body();
   if("1".equalsIgnoreCase(req.getString("id"))){
      message.reply(new JsonObject(new People()));
   }else {
      FailObject failObject = new FailObject();
      failObject.setStatus(400);
      failObject.setMessage("Error no data found");
      failObject.setField1("data field1");
      failObject.setField2("data field2");
      message.fail(500, Json.encode(failObject));
   }
}

期望结果：无数据时返回400状态码及自定义响应体：

{"status":400, "message":"Error no data found", "field1":"data field1", "field2":"data field2"}

解决方案 #

方案1：正常回复+控制器端判断处理 #

message.fail()会触发EventBus的RECIPIENT_FAILURE事件，框架默认映射为500错误。可改为服务端直接回复错误对象，控制器端判断后设置对应HTTP状态码。

修改服务端代码

@ConsumeEvent("getPeopleById")
public void getPeopleById(Message message){
   JsonObject req = (JsonObject) message.body();
   if("1".equalsIgnoreCase(req.getString("id"))){
      message.reply(new People());
   }else {
      FailObject failObject = new FailObject();
      failObject.setStatus(400);
      failObject.setMessage("Error no data found");
      failObject.setField1("data field1");
      failObject.setField2("data field2");
      // 直接回复错误对象，替代message.fail()
      message.reply(failObject);
   }
}

修改控制器代码

将返回类型改为Uni<Response>，通过Response对象自定义状态码和响应体：

@Inject
EventBus bus;

@GET
@Produces(MediaType.APPLICATION_JSON)
@Path("/path/{id}")
public Uni<Response> getPeopleById(@PathParam("id") String id){
   JsonObject req = new JsonObject();
   req.put("id", id);
   return bus.request("getPeopleById", req, new DeliveryOptions().setSendTimeout(12000L))
         .onItem().transform(Message::body)
         .onItem().transformToUni(body -> {
             if (body instanceof People) {
                 // 成功场景：返回200状态码+People对象
                 return Uni.createFrom().item(Response.ok(body).build());
             } else if (body instanceof FailObject) {
                 FailObject failObj = (FailObject) body;
                 // 错误场景：返回自定义状态码+错误对象
                 return Uni.createFrom().item(Response.status(failObj.getStatus()).entity(failObj).build());
             } else {
                 // 未知场景：返回500默认错误
                 return Uni.createFrom().item(Response.status(Response.Status.INTERNAL_SERVER_ERROR).build());
             }
         })
         .ifNoItem().after(Duration.ofMillis(12000L))
         .failWith(new Exception("Error Service"))
         // 全局异常处理：将超时等异常转为自定义响应
         .onFailure().recoverWithItem(throwable -> {
             FailObject failObj = new FailObject();
             failObj.setStatus(500);
             failObj.setMessage("Service timeout or unexpected error");
             return Response.status(500).entity(failObj).build();
         });
}

方案2：自定义异常+ExceptionMapper统一处理 #

通过自定义异常传递错误信息，配合ExceptionMapper自动将异常转换为HTTP响应，实现全局统一错误处理。

1. 定义自定义异常

public class CustomServiceException extends RuntimeException {
    private final int statusCode;
    private final FailObject failObject;

    public CustomServiceException(int statusCode, FailObject failObject) {
        super(failObject.getMessage());
        this.statusCode = statusCode;
        this.failObject = failObject;
    }

    // Getter方法
    public int getStatusCode() { return statusCode; }
    public FailObject getFailObject() { return failObject; }
}

2. 修改服务端代码

服务端方法返回Uni<People>，抛出自定义异常替代message.fail()：

@ConsumeEvent("getPeopleById")
public Uni<People> getPeopleById(JsonObject req){
   if("1".equalsIgnoreCase(req.getString("id"))){
      return Uni.createFrom().item(new People());
   }else {
      FailObject failObject = new FailObject();
      failObject.setStatus(400);
      failObject.setMessage("Error no data found");
      failObject.setField1("data field1");
      failObject.setField2("data field2");
      // 抛出自定义异常
      return Uni.createFrom().failure(new CustomServiceException(400, failObject));
   }
}

3. 实现ExceptionMapper

@Provider
public class CustomServiceExceptionMapper implements ExceptionMapper<CustomServiceException> {
    @Override
    public Response toResponse(CustomServiceException exception) {
        return Response.status(exception.getStatusCode())
                .entity(exception.getFailObject())
                .type(MediaType.APPLICATION_JSON)
                .build();
    }
}

4. 简化控制器代码

控制器无需额外判断，异常会被ExceptionMapper自动处理：

@Inject
EventBus bus;

@GET
@Produces(MediaType.APPLICATION_JSON)
@Path("/path/{id}")
public Uni<People> getPeopleById(@PathParam("id") String id){
   JsonObject req = new JsonObject();
   req.put("id", id);
   return bus.<People>request("getPeopleById", req, new DeliveryOptions().setSendTimeout(12000L))
         .onItem().transform(Message::body)
         .ifNoItem().after(Duration.ofMillis(12000L))
         .failWith(new CustomServiceException(500, new FailObject(){{
             setStatus(500);
             setMessage("Error Service");
         }}));
}

内容的提问来源于stack exchange，提问作者David Vincent