This is an automated email from the ASF dual-hosted git repository.
mxsm pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/eventmesh.git
The following commit(s) were added to refs/heads/master by this push:
new d62579074 [ISSUE #4659] Fix HTTP source connector stop after receiving
an invalid request (#4666)
d62579074 is described below
commit d62579074d782b344179bcf25781f99f17c146d0
Author: Fungx <[email protected]>
AuthorDate: Sat Dec 16 21:11:14 2023 +0800
[ISSUE #4659] Fix HTTP source connector stop after receiving an invalid
request (#4666)
---
.../http/source/connector/HttpSourceConnector.java | 17 ++++++++++++++---
1 file changed, 14 insertions(+), 3 deletions(-)
diff --git
a/eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/source/connector/HttpSourceConnector.java
b/eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/source/connector/HttpSourceConnector.java
index b4b58f9f8..f9a6c568e 100644
---
a/eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/source/connector/HttpSourceConnector.java
+++
b/eventmesh-connectors/eventmesh-connector-http/src/main/java/org/apache/eventmesh/connector/http/source/connector/HttpSourceConnector.java
@@ -33,7 +33,6 @@ import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import io.cloudevents.CloudEvent;
-import io.cloudevents.core.message.MessageReader;
import io.cloudevents.http.vertx.VertxMessageFactory;
import io.netty.handler.codec.http.HttpResponseStatus;
import io.vertx.core.Vertx;
@@ -83,7 +82,19 @@ public class HttpSourceConnector implements Source {
.handler(LoggerHandler.create())
.handler(ctx -> {
VertxMessageFactory.createReader(ctx.request())
- .map(MessageReader::toEvent)
+ .map(reader -> {
+ CloudEvent event = reader.toEvent();
+ if (event.getSubject() == null) {
+ throw new IllegalStateException("attribute
'subject' cannot be null");
+ }
+ if (event.getDataContentType() == null) {
+ throw new IllegalStateException("attribute
'datacontenttype' cannot be null");
+ }
+ if (event.getData() == null) {
+ throw new IllegalStateException("attribute 'data'
cannot be null");
+ }
+ return event;
+ })
.onSuccess(event -> {
queue.add(event);
log.info("[HttpSourceConnector] Succeed to convert
payload into CloudEvent. StatusCode={}", HttpResponseStatus.OK.code());
@@ -91,7 +102,7 @@ public class HttpSourceConnector implements Source {
})
.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()
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
