fix subject check

Fungx · Dec 16, 2023 · ed8787f · ed8787f
1 parent b85a58c
commit ed8787f
Showing 1 changed file with 8 additions and 2 deletions.
diff --git a/...c/main/java/org/apache/eventmesh/connector/http/source/connector/HttpSourceConnector.java b/...c/main/java/org/apache/eventmesh/connector/http/source/connector/HttpSourceConnector.java
@@ -83,15 +83,21 @@ private void doInit() {
             .handler(LoggerHandler.create())
             .handler(ctx -> {
                 VertxMessageFactory.createReader(ctx.request())
-                    .map(MessageReader::toEvent)
+                    .map(r -> {
+                        CloudEvent event = r.toEvent();
+                        if (event.getSubject() == null) {
+                            throw new IllegalStateException("attribute 'subject' cannot be null");
+                        }
+                        return event;
+                    })
                     .onSuccess(event -> {
                         queue.add(event);
                         log.info("[HttpSourceConnector] Succeed to convert payload into CloudEvent. StatusCode={}", HttpResponseStatus.OK.code());
                         ctx.response().setStatusCode(HttpResponseStatus.OK.code()).end();
                     })
                     .onFailure(t -> {
                         log.error("[HttpSourceConnector] Malformed request. StatusCode={}", HttpResponseStatus.BAD_REQUEST.code(), t);
-                        ctx.response().setStatusCode(HttpResponseStatus.BAD_REQUEST.code()).setStatusMessage(t.getMessage()).end();
+                        ctx.response().setStatusCode(HttpResponseStatus.BAD_REQUEST.code()).end();
                     });
             });
         this.server = vertx.createHttpServer(new HttpServerOptions()