githublaohu commented on code in PR #914:
URL:
https://github.com/apache/incubator-eventmesh/pull/914#discussion_r897634619
##########
eventmesh-webhook/eventmesh-webhook-receive/src/main/java/org/apache/eventmesh/webhook/receive/WebHookController.java:
##########
@@ -81,14 +93,44 @@ public void execute(String path, Map<String, String>
header, byte[] body) throws
cloudEventId = webHookRequest.getManufacturerEventId();
}
String eventType = manufacturerName + "." +
webHookConfig.getManufacturerEventName();
- CloudEvent event = CloudEventBuilder.v1()
- .withId(cloudEventId)
- .withSubject(webHookConfig.getCloudEventName())
- .withSource(URI.create(webHookConfig.getCloudEventSource()))
- .withDataContentType(webHookConfig.getDataContentType())
- .withType(eventType)
- .withData(body)
+
+ // 4. send cloudEvent
+ Properties properties;
+ try (final InputStream inputStream =
WebHookController.class.getClassLoader().getResourceAsStream("eventmesh.properties"))
{
+ properties = new Properties();
+ properties.load(inputStream);
+ }
Review Comment:
每次调用读一次吗？
##########
eventmesh-webhook/eventmesh-webhook-receive/src/main/java/org/apache/eventmesh/webhook/receive/WebHookController.java:
##########
@@ -81,14 +93,44 @@ public void execute(String path, Map<String, String>
header, byte[] body) throws
cloudEventId = webHookRequest.getManufacturerEventId();
}
String eventType = manufacturerName + "." +
webHookConfig.getManufacturerEventName();
- CloudEvent event = CloudEventBuilder.v1()
- .withId(cloudEventId)
- .withSubject(webHookConfig.getCloudEventName())
- .withSource(URI.create(webHookConfig.getCloudEventSource()))
- .withDataContentType(webHookConfig.getDataContentType())
- .withType(eventType)
- .withData(body)
+
+ // 4. send cloudEvent
+ Properties properties;
+ try (final InputStream inputStream =
WebHookController.class.getClassLoader().getResourceAsStream("eventmesh.properties"))
{
+ properties = new Properties();
+ properties.load(inputStream);
+ }
+ final String eventMeshIp =
properties.getProperty("eventMesh.server.ip");
+ final String eventMeshHttpPort =
properties.getProperty("eventMesh.server.http.port");
+
+ String eventMeshIPPort = eventMeshIp + ":" + eventMeshHttpPort;
+
+ // Both the producer and consumer require an instance of
EventMeshHttpClientConfig class
+ // that specifies the configuration of EventMesh HTTP client.
+ EventMeshHttpClientConfig eventMeshClientConfig =
EventMeshHttpClientConfig.builder()
+ .liteEventMeshAddr(eventMeshIPPort)
+ .producerGroup("eventMesh.webhook.producerGroup")
+ .env(properties.getProperty("eventMesh.server.env"))
+ .idc(properties.getProperty("eventMesh.server.idc"))
+ .ip(properties.getProperty("eventMesh.server.ip"))
+ .sys(properties.getProperty("eventMesh.sysid"))
+ .pid(String.valueOf(ThreadUtils.getPID()))
+
.userName(properties.getProperty("eventMesh.server.http.username"))
+
.password(properties.getProperty("eventMesh.server.http.password"))
.build();
+
+ try (EventMeshHttpProducer eventMeshHttpProducer = new
EventMeshHttpProducer(eventMeshClientConfig)) {
+
+ CloudEvent event = CloudEventBuilder.v1()
+ .withId(cloudEventId)
+ .withSubject(webHookConfig.getCloudEventName())
+
.withSource(URI.create(webHookConfig.getCloudEventSource()))
+ .withDataContentType(webHookConfig.getDataContentType())
+ .withType(eventType)
+ .withData(body)
+ .build();
+ eventMeshHttpProducer.publish(event);
Review Comment:
不可能只支持http
##########
eventmesh-webhook/eventmesh-webhook-receive/src/main/java/org/apache/eventmesh/webhook/receive/WebHookController.java:
##########
@@ -81,14 +93,44 @@ public void execute(String path, Map<String, String>
header, byte[] body) throws
cloudEventId = webHookRequest.getManufacturerEventId();
}
String eventType = manufacturerName + "." +
webHookConfig.getManufacturerEventName();
- CloudEvent event = CloudEventBuilder.v1()
- .withId(cloudEventId)
- .withSubject(webHookConfig.getCloudEventName())
- .withSource(URI.create(webHookConfig.getCloudEventSource()))
- .withDataContentType(webHookConfig.getDataContentType())
- .withType(eventType)
- .withData(body)
+
+ // 4. send cloudEvent
+ Properties properties;
+ try (final InputStream inputStream =
WebHookController.class.getClassLoader().getResourceAsStream("eventmesh.properties"))
{
+ properties = new Properties();
+ properties.load(inputStream);
+ }
Review Comment:
注意 producer的缓存，不能反复创建
##########
eventmesh-webhook/eventmesh-webhook-receive/src/main/java/org/apache/eventmesh/webhook/receive/WebHookController.java:
##########
@@ -81,14 +93,44 @@ public void execute(String path, Map<String, String>
header, byte[] body) throws
cloudEventId = webHookRequest.getManufacturerEventId();
}
String eventType = manufacturerName + "." +
webHookConfig.getManufacturerEventName();
- CloudEvent event = CloudEventBuilder.v1()
- .withId(cloudEventId)
- .withSubject(webHookConfig.getCloudEventName())
- .withSource(URI.create(webHookConfig.getCloudEventSource()))
- .withDataContentType(webHookConfig.getDataContentType())
- .withType(eventType)
- .withData(body)
+
+ // 4. send cloudEvent
+ Properties properties;
+ try (final InputStream inputStream =
WebHookController.class.getClassLoader().getResourceAsStream("eventmesh.properties"))
{
+ properties = new Properties();
+ properties.load(inputStream);
+ }
Review Comment:
cloudEvent Producer 支持http，rocketmq等。具体是那种WebHookConfig
有配置，里面有cloudevent服务的地址，端口，topic
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
