xwm1992 commented on code in PR #937:
URL:
https://github.com/apache/incubator-eventmesh/pull/937#discussion_r916525171
##########
docs/cn/features/webhook.md:
##########
@@ -0,0 +1,158 @@
+
+
+## webhook使用流程
+#### 第一步：在eventmesh配置webhook相关信息并且启动
+
+##### 配置说明
+```
+# 是否启动webhook admin服务
+eventMesh.webHook.admin.start=true
+
+# webhook事件配置存储模式。目前只支持file与nacos
+eventMesh.webHook.operationMode=file
+# 文件存储模式的文件存放路径，如果写上#{eventMeshHome}，在eventMesh根目录
+eventMesh.webHook.fileMode.filePath= #{eventMeshHome}/webhook
+
+# nacos存储模式，配置命名规则是eventMesh.webHook.nacosMode.{nacos 原生配置key} 具体的配置请看 [nacos
github
api](https://github.com/alibaba/nacos/blob/develop/api/src/main/java/com/alibaba/nacos/api/SystemPropertyKeyConst.java)
+## nacos的地址
+eventMesh.webHook.nacosMode.serverAddr=127.0.0.1:8848
+
+# webhook eventcloud 发送模式。与eventMesh.connector.plugin.type 配置一样
+eventMesh.webHook.producer.connector=standalone
+```
+
+#### 第二步：添加webhook配置信息
+配置信息说明
+```java
+ /**
+ * 厂商调用的path。厂商事件调用地址、 [http or https ]://[域名 or IP
【厂商可以被调用】]:[端口]/webhook/[callbackPath]
+ * 比如：http://127.0.0.1:10504/webhook/test/event 需要把全完url填入厂商调用输入中
+ * callbackPath 是唯一
+ * manufacturer callback path
+ */
+ private String callbackPath;
+
+ /**
+ * 厂商的名字
+ * manufacturer name ,like github
+ */
+ private String manufacturerName;
+
+ /**
+ * 厂商的事件名
+ * webhook event name ,like rep-push
+ */
+ private String manufacturerEventName;
+
+ /**
+ *
+ * http header content type
+ */
+ private String contentType = "application/json";
+
+ /**
+ * 说明
+ * description of this WebHookConfig
+ */
+ private String description;
+
+ /**
+ * 有一些厂商使用验签方式，
+ * secret key ,for authentication
+ */
+ private String secret;
+
+ /**
+ * 有一些厂商使用验签方式，使用账户密码方式
+ * userName ,for HTTP authentication
+ */
+ private String userName;
+
+ /**
+ * 有一些厂商使用验签方式，使用账户密码方式
+ * password ,for HTTP authentication
+ */
+ private String password;
+
+
+
+ /**
+ * 事件发送到那个topic
+ * roll out event name ,like topic to mq
+ */
+ private String cloudEventName;
+
+ /**
+ * roll out data format -> CloudEvent serialization mode
+ * If HTTP protocol is used, the request header contentType needs to be
marked
+ */
+ private String dataContentType = "application/json";;
+
+ /**
+ * source of event
+ */
+ private String cloudEventSource;
+
+ /**
+ * cloudEvent事件对象唯一标识符识别方式，uuid或者manufacturerEventId(厂商id)
+ * id of cloudEvent ,like uuid/manufacturerEventId
+ */
+ private String cloudEventIdGenerateMode;
+
+```
+
+##### 添加接口
+路径： /webhook/insertWebHookConfig
+方法： POST
+contentType： application/json
+
+输入参数：
+| 字段 | 说明 | 类型 | 必须 | 默认值 |
+| -- | -- | -- | -- | -- |
+| callbackPath | 调用地址，唯一地址 | string | 是 | null |
+| manufacturerName | 厂商名 | string | 是 | null |
+| manufacturerEventName | 厂商事件名 | string | 是 | null |
+| contentType | http connettype | string | 否 | application/json |
+| description | 配置说明 | string | 否 | null |
+| secret | 验签密钥 | string | 否 | null |
+| userName | 用户名 | string | 否 | null |
+| password | 用户密码 | string | 否 | null |
+| cloudEventName | 事件名（） | string | 是 | null |
+| cloudEventSource | 事件来源可以填写 | string | 是 | null |
+| cloudEventIdGenerateMode |
cloudEvent事件对象唯一标识符识别方式，uuid或者manufacturerEventId(厂商id) | string | 否 |
manufacturerEventId |
+
+输出参数：１ 成功，０失败
+
+
+
+
+#### 第三步：查看配置是否成功
+1. file存储模式。请到eventMesh.webHook.fileMode.filePath 目录下查看。文件名为callbackPath转移后的
+2. nacos存储模式。请到eventMesh.webHook.nacosMode.serverAddr 配置的nacos服务去看
+
+#### 第四步：配置cloudevent的消费者
+
+
+#### 第五步：在厂商配置webhook相关信息
+> 厂商操作请看【厂商webhook操作说明】
+
+
+## 厂商webhook操作说明
+### github 注册
+#### 第一步：进入对应的项目
+#### 第二步：点击setting
+
Review Comment:
these images not display may be url is not correct
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HandlerService.java:
##########
@@ -0,0 +1,233 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.runtime.core.protocol.http.processor;
+
+import org.apache.eventmesh.runtime.boot.HTTPTrace;
+import org.apache.eventmesh.runtime.boot.HTTPTrace.TraceOperation;
+import org.apache.eventmesh.runtime.metrics.http.HTTPMetricsServer;
+import org.apache.eventmesh.runtime.util.HttpResponseUtils;
+import org.apache.eventmesh.runtime.util.RemotingHelper;
+
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ThreadPoolExecutor;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.HttpRequest;
+import io.netty.handler.codec.http.HttpResponse;
+
+import lombok.Setter;
+
+
+public class HandlerService {
+
+ private Logger httpServerLogger = LoggerFactory.getLogger(this.getClass());
+
+ private Logger httpLogger = LoggerFactory.getLogger("http");
+
+ private Map<String, ProcessorWrapper> httpProcessorMap = new
ConcurrentHashMap<>();
+
+ @Setter
+ private HTTPMetricsServer metrics;
+
+ @Setter
+ private HTTPTrace httpTrace;
+
+
+ public void init() {
+ httpServerLogger.info("HandlerService start ");
+ }
+
+ public void register(HttpProcessor httpProcessor, ThreadPoolExecutor
threadPoolExecutor) {
+ for (String path : httpProcessor.paths()) {
+ this.register(path, httpProcessor, threadPoolExecutor);
+ }
+ }
+
+ public void register(String path, HttpProcessor httpProcessor,
ThreadPoolExecutor threadPoolExecutor) {
+
+ if (httpProcessorMap.containsKey(path)) {
+ throw new RuntimeException(String.format("HandlerService path %s
repeat, repeat processor is %s ",
+ path, httpProcessor.getClass().getSimpleName()));
+ }
+ ProcessorWrapper processorWrapper = new ProcessorWrapper();
+ processorWrapper.threadPoolExecutor = threadPoolExecutor;
+ processorWrapper.httpProcessor = httpProcessor;
+ if (httpProcessor instanceof AsynHttpProcessor) {
+ processorWrapper.asyn = (AsynHttpProcessor) httpProcessor;
+ } else {
+ processorWrapper.httpProcessor = httpProcessor;
+ }
+ httpProcessorMap.put(path, processorWrapper);
+ httpServerLogger.info("path is {} proocessor name is", path,
httpProcessor.getClass().getSimpleName());
Review Comment:
change -> httpServerLogger.info("path is {} processor name is {}", path,
httpProcessor.getClass().getSimpleName());
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HandlerService.java:
##########
@@ -0,0 +1,233 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.runtime.core.protocol.http.processor;
+
+import org.apache.eventmesh.runtime.boot.HTTPTrace;
+import org.apache.eventmesh.runtime.boot.HTTPTrace.TraceOperation;
+import org.apache.eventmesh.runtime.metrics.http.HTTPMetricsServer;
+import org.apache.eventmesh.runtime.util.HttpResponseUtils;
+import org.apache.eventmesh.runtime.util.RemotingHelper;
+
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ThreadPoolExecutor;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.HttpRequest;
+import io.netty.handler.codec.http.HttpResponse;
+
+import lombok.Setter;
+
+
+public class HandlerService {
+
+ private Logger httpServerLogger = LoggerFactory.getLogger(this.getClass());
+
+ private Logger httpLogger = LoggerFactory.getLogger("http");
+
+ private Map<String, ProcessorWrapper> httpProcessorMap = new
ConcurrentHashMap<>();
+
+ @Setter
+ private HTTPMetricsServer metrics;
+
+ @Setter
+ private HTTPTrace httpTrace;
+
+
+ public void init() {
+ httpServerLogger.info("HandlerService start ");
+ }
+
+ public void register(HttpProcessor httpProcessor, ThreadPoolExecutor
threadPoolExecutor) {
+ for (String path : httpProcessor.paths()) {
+ this.register(path, httpProcessor, threadPoolExecutor);
+ }
+ }
+
+ public void register(String path, HttpProcessor httpProcessor,
ThreadPoolExecutor threadPoolExecutor) {
+
+ if (httpProcessorMap.containsKey(path)) {
+ throw new RuntimeException(String.format("HandlerService path %s
repeat, repeat processor is %s ",
+ path, httpProcessor.getClass().getSimpleName()));
+ }
+ ProcessorWrapper processorWrapper = new ProcessorWrapper();
+ processorWrapper.threadPoolExecutor = threadPoolExecutor;
+ processorWrapper.httpProcessor = httpProcessor;
+ if (httpProcessor instanceof AsynHttpProcessor) {
+ processorWrapper.asyn = (AsynHttpProcessor) httpProcessor;
+ } else {
+ processorWrapper.httpProcessor = httpProcessor;
+ }
+ httpProcessorMap.put(path, processorWrapper);
+ httpServerLogger.info("path is {} proocessor name is", path,
httpProcessor.getClass().getSimpleName());
+ }
+
+ public boolean isProcessorWrapper(HttpRequest httpRequest) {
+ return this.getProcessorWrapper(httpRequest) == null ? false : true;
Review Comment:
change -> return this.getProcessorWrapper(httpRequest) != null;
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HandlerService.java:
##########
@@ -0,0 +1,233 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.runtime.core.protocol.http.processor;
+
+import org.apache.eventmesh.runtime.boot.HTTPTrace;
+import org.apache.eventmesh.runtime.boot.HTTPTrace.TraceOperation;
+import org.apache.eventmesh.runtime.metrics.http.HTTPMetricsServer;
+import org.apache.eventmesh.runtime.util.HttpResponseUtils;
+import org.apache.eventmesh.runtime.util.RemotingHelper;
+
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ThreadPoolExecutor;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.HttpRequest;
+import io.netty.handler.codec.http.HttpResponse;
+
+import lombok.Setter;
+
+
+public class HandlerService {
+
+ private Logger httpServerLogger = LoggerFactory.getLogger(this.getClass());
+
+ private Logger httpLogger = LoggerFactory.getLogger("http");
+
+ private Map<String, ProcessorWrapper> httpProcessorMap = new
ConcurrentHashMap<>();
+
+ @Setter
+ private HTTPMetricsServer metrics;
+
+ @Setter
+ private HTTPTrace httpTrace;
+
+
+ public void init() {
+ httpServerLogger.info("HandlerService start ");
+ }
+
+ public void register(HttpProcessor httpProcessor, ThreadPoolExecutor
threadPoolExecutor) {
+ for (String path : httpProcessor.paths()) {
+ this.register(path, httpProcessor, threadPoolExecutor);
+ }
+ }
+
+ public void register(String path, HttpProcessor httpProcessor,
ThreadPoolExecutor threadPoolExecutor) {
+
+ if (httpProcessorMap.containsKey(path)) {
+ throw new RuntimeException(String.format("HandlerService path %s
repeat, repeat processor is %s ",
+ path, httpProcessor.getClass().getSimpleName()));
+ }
+ ProcessorWrapper processorWrapper = new ProcessorWrapper();
+ processorWrapper.threadPoolExecutor = threadPoolExecutor;
+ processorWrapper.httpProcessor = httpProcessor;
+ if (httpProcessor instanceof AsynHttpProcessor) {
+ processorWrapper.asyn = (AsynHttpProcessor) httpProcessor;
+ } else {
+ processorWrapper.httpProcessor = httpProcessor;
+ }
+ httpProcessorMap.put(path, processorWrapper);
+ httpServerLogger.info("path is {} proocessor name is", path,
httpProcessor.getClass().getSimpleName());
+ }
+
+ public boolean isProcessorWrapper(HttpRequest httpRequest) {
+ return this.getProcessorWrapper(httpRequest) == null ? false : true;
+ }
+
+ private ProcessorWrapper getProcessorWrapper(HttpRequest httpRequest) {
+ String uri = httpRequest.uri();
+ for (Entry<String, ProcessorWrapper> e : httpProcessorMap.entrySet()) {
+ if (e.getKey().startsWith(uri)) {
+ return e.getValue();
+ }
+ }
+ return null;
+ }
+
+ /**
+ * @param httpRequest
+ */
+ public void handler(ChannelHandlerContext ctx, HttpRequest httpRequest) {
+
+
+ TraceOperation traceOperation =
httpTrace.getTraceOperation(httpRequest, ctx.channel());
+
+ ProcessorWrapper processorWrapper = getProcessorWrapper(httpRequest);
+ if (Objects.isNull(processorWrapper)) {
+ this.sendResponse(ctx, HttpResponseUtils.createNotFound());
+ }
+ try {
+ HandlerSpecific handlerSpecific = new HandlerSpecific();
+ handlerSpecific.httpRequest = httpRequest;
+ handlerSpecific.ctx = ctx;
+ handlerSpecific.traceOperation = traceOperation;
+ processorWrapper.threadPoolExecutor.execute(handlerSpecific);
Review Comment:
`processorWrapper` may produce NullPointerException
##########
eventmesh-runtime/src/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HandlerService.java:
##########
@@ -0,0 +1,233 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.eventmesh.runtime.core.protocol.http.processor;
+
+import org.apache.eventmesh.runtime.boot.HTTPTrace;
+import org.apache.eventmesh.runtime.boot.HTTPTrace.TraceOperation;
+import org.apache.eventmesh.runtime.metrics.http.HTTPMetricsServer;
+import org.apache.eventmesh.runtime.util.HttpResponseUtils;
+import org.apache.eventmesh.runtime.util.RemotingHelper;
+
+import java.util.Map;
+import java.util.Map.Entry;
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ThreadPoolExecutor;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import io.netty.channel.ChannelFutureListener;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.handler.codec.http.HttpRequest;
+import io.netty.handler.codec.http.HttpResponse;
+
+import lombok.Setter;
+
+
+public class HandlerService {
+
+ private Logger httpServerLogger = LoggerFactory.getLogger(this.getClass());
+
+ private Logger httpLogger = LoggerFactory.getLogger("http");
+
+ private Map<String, ProcessorWrapper> httpProcessorMap = new
ConcurrentHashMap<>();
+
+ @Setter
+ private HTTPMetricsServer metrics;
+
+ @Setter
+ private HTTPTrace httpTrace;
+
+
+ public void init() {
+ httpServerLogger.info("HandlerService start ");
+ }
+
+ public void register(HttpProcessor httpProcessor, ThreadPoolExecutor
threadPoolExecutor) {
+ for (String path : httpProcessor.paths()) {
+ this.register(path, httpProcessor, threadPoolExecutor);
+ }
+ }
+
+ public void register(String path, HttpProcessor httpProcessor,
ThreadPoolExecutor threadPoolExecutor) {
+
+ if (httpProcessorMap.containsKey(path)) {
+ throw new RuntimeException(String.format("HandlerService path %s
repeat, repeat processor is %s ",
+ path, httpProcessor.getClass().getSimpleName()));
+ }
+ ProcessorWrapper processorWrapper = new ProcessorWrapper();
+ processorWrapper.threadPoolExecutor = threadPoolExecutor;
+ processorWrapper.httpProcessor = httpProcessor;
+ if (httpProcessor instanceof AsynHttpProcessor) {
+ processorWrapper.asyn = (AsynHttpProcessor) httpProcessor;
+ } else {
+ processorWrapper.httpProcessor = httpProcessor;
+ }
+ httpProcessorMap.put(path, processorWrapper);
+ httpServerLogger.info("path is {} proocessor name is", path,
httpProcessor.getClass().getSimpleName());
+ }
+
+ public boolean isProcessorWrapper(HttpRequest httpRequest) {
+ return this.getProcessorWrapper(httpRequest) == null ? false : true;
+ }
+
+ private ProcessorWrapper getProcessorWrapper(HttpRequest httpRequest) {
+ String uri = httpRequest.uri();
+ for (Entry<String, ProcessorWrapper> e : httpProcessorMap.entrySet()) {
+ if (e.getKey().startsWith(uri)) {
+ return e.getValue();
+ }
+ }
+ return null;
+ }
+
+ /**
+ * @param httpRequest
+ */
+ public void handler(ChannelHandlerContext ctx, HttpRequest httpRequest) {
+
+
+ TraceOperation traceOperation =
httpTrace.getTraceOperation(httpRequest, ctx.channel());
+
+ ProcessorWrapper processorWrapper = getProcessorWrapper(httpRequest);
+ if (Objects.isNull(processorWrapper)) {
+ this.sendResponse(ctx, HttpResponseUtils.createNotFound());
+ }
+ try {
+ HandlerSpecific handlerSpecific = new HandlerSpecific();
+ handlerSpecific.httpRequest = httpRequest;
+ handlerSpecific.ctx = ctx;
+ handlerSpecific.traceOperation = traceOperation;
+ processorWrapper.threadPoolExecutor.execute(handlerSpecific);
+ } catch (Exception e) {
+ httpServerLogger.error(e.getMessage(), e);
+ this.sendResponse(ctx,
HttpResponseUtils.createInternalServerError());
+ }
+ }
+
+ private void sendResponse(ChannelHandlerContext ctx, HttpResponse
response) {
+ this.sendResponse(ctx, response, true);
+ }
+
+ private void sendResponse(ChannelHandlerContext ctx, HttpResponse
response, boolean isClose) {
+ ctx.writeAndFlush(response).addListener((ChannelFutureListener) f -> {
+ if (!f.isSuccess()) {
+ httpLogger.warn("send response to [{}] fail, will close this
channel",
+ RemotingHelper.parseChannelRemoteAddr(f.channel()));
+ if (isClose) {
+ f.channel().close();
+ }
+ }
+ });
+ }
+
+ class HandlerSpecific implements Runnable {
+
+ private TraceOperation traceOperation;
+
+ private ChannelHandlerContext ctx;
+
+ private HttpRequest httpRequest;
+
+ private HttpResponse response;
+
+ private Throwable exception;
+
+ long requestTime = System.currentTimeMillis();
+
+
+ public void run() {
+ ProcessorWrapper processorWrapper =
HandlerService.this.httpProcessorMap.get(httpRequest.uri());
+ try {
+ this.postHandler();
+ if (Objects.isNull(processorWrapper.httpProcessor)) {
+ processorWrapper.asyn.handler(this, httpRequest);
+ return;
+ }
+ response = processorWrapper.httpProcessor.handler(httpRequest);
+
+ this.preHandler();
+ } catch (Throwable e) {
+ httpServerLogger.error("{},{}");
Review Comment:
this line is no use.
--
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]
