xwm1992 · xwm1992 · Sep 9, 2022 · Sep 9, 2022 · Sep 9, 2022 · Sep 9, 2022
diff --git a/.gitignore b/.gitignore
@@ -29,3 +29,5 @@ h2/db.mv.db
 all-dependencies.txt
 self-modules.txt
 third-party-dependencies.txt
+/bin/
+*/**/bin/
diff --git a/eventmesh-common/src/main/java/org/apache/eventmesh/common/config/ConfigurationWrapper.java b/eventmesh-common/src/main/java/org/apache/eventmesh/common/config/ConfigurationWrapper.java
@@ -64,12 +64,12 @@ public boolean support(FileChangeContext changeContext) {
 
     public ConfigurationWrapper(String directoryPath, String fileName, boolean reload) {
         this.directoryPath = directoryPath
-                .replace('/', File.separator.charAt(0))
-                .replace('\\', File.separator.charAt(0));
+            .replace('/', File.separator.charAt(0))
+            .replace('\\', File.separator.charAt(0));
         this.fileName = fileName;
         this.file = (directoryPath + File.separator + fileName)
-                .replace('/', File.separator.charAt(0))
-                .replace('\\', File.separator.charAt(0));
+            .replace('/', File.separator.charAt(0))
+            .replace('\\', File.separator.charAt(0));
         this.reload = reload;
         init();
     }
@@ -104,7 +104,7 @@ public int getIntProp(String configKey, int defaultValue) {
             return defaultValue;
         }
         Preconditions.checkState(StringUtils.isNumeric(configValue),
-                String.format("key:%s, value:%s error", configKey, configValue));
+            String.format("key:%s, value:%s error", configKey, configValue));
         return Integer.parseInt(configValue);
     }
 
@@ -138,4 +138,8 @@ public <T> T getPropertiesByConfig(String prefix, Class<?> clazz, boolean remove
         return (T) objectMapper.convertValue(getPropertiesByConfig(prefix, removePrefix), clazz);
     }
 
+    public Properties getProperties() {
+        return this.properties;
+    }
+
 }
diff --git a/eventmesh-connector-plugin/eventmesh-connector-pulsar/gradle.properties b/eventmesh-connector-plugin/eventmesh-connector-pulsar/gradle.properties
@@ -15,4 +15,4 @@
 #
 
 pluginType=connector
-pluginName=rocketmq
+pluginName=pulsar
diff --git a/...l-webhook/src/main/java/org/apache/eventmesh/protocol/webhook/WebHookProtocolAdaptor.java b/...l-webhook/src/main/java/org/apache/eventmesh/protocol/webhook/WebHookProtocolAdaptor.java
@@ -17,14 +17,20 @@
 
 package org.apache.eventmesh.protocol.webhook;
 
+import org.apache.eventmesh.common.Constants;
 import org.apache.eventmesh.common.protocol.ProtocolTransportObject;
+import org.apache.eventmesh.common.protocol.http.HttpEventWrapper;
 import org.apache.eventmesh.common.protocol.http.WebhookProtocolTransportObject;
 import org.apache.eventmesh.protocol.api.ProtocolAdaptor;
 import org.apache.eventmesh.protocol.api.exception.ProtocolHandleException;
 
+
 import java.net.URI;
 import java.util.ArrayList;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
+import java.util.Set;
 
 import io.cloudevents.CloudEvent;
 import io.cloudevents.core.builder.CloudEventBuilder;
@@ -34,13 +40,14 @@ public class WebHookProtocolAdaptor implements ProtocolAdaptor<WebhookProtocolTr
     @Override
     public CloudEvent toCloudEvent(WebhookProtocolTransportObject protocol) throws ProtocolHandleException {
         return CloudEventBuilder.v1()
-                .withId(protocol.getCloudEventId())
-                .withSubject(protocol.getCloudEventName())
-                .withSource(URI.create(protocol.getCloudEventSource()))
-                .withDataContentType(protocol.getDataContentType())
-                .withType(protocol.getEventType())
-                .withData(protocol.getBody())
-                .build();
+            .withId(protocol.getCloudEventId())
+            .withSubject(protocol.getCloudEventName())
+            .withSource(URI.create(protocol.getCloudEventSource()))
+            .withDataContentType(protocol.getDataContentType())
+            .withType(protocol.getEventType())
+            .withData(protocol.getBody())
+            .withExtension(Constants.PROTOCOL_TYPE, "webhookProtocolAdaptor")
+            .build();
     }
 
     @Override
@@ -51,12 +58,30 @@ public List<CloudEvent> toBatchCloudEvent(WebhookProtocolTransportObject protoco
 
     @Override
     public ProtocolTransportObject fromCloudEvent(CloudEvent cloudEvent) throws ProtocolHandleException {
-        return null;
+        final HttpEventWrapper httpEventWrapper = new HttpEventWrapper();
+        Map<String, Object> sysHeaderMap = new HashMap<>();
+        // ce attributes
+        Set<String> attributeNames = cloudEvent.getAttributeNames();
+        // ce extensions
+        Set<String> extensionNames = cloudEvent.getExtensionNames();
+        for (String attributeName : attributeNames) {
+            sysHeaderMap.put(attributeName, cloudEvent.getAttribute(attributeName));
+        }
+        for (String extensionName : extensionNames) {
+            sysHeaderMap.put(extensionName, cloudEvent.getExtension(extensionName));
+        }
+        sysHeaderMap.put("cloudEventId", cloudEvent.getId());
+        sysHeaderMap.put("cloudEventName", cloudEvent.getSubject());
+        sysHeaderMap.put("cloudEventSource", cloudEvent.getSource().toString());
+        sysHeaderMap.put("type", cloudEvent.getType());
+        httpEventWrapper.setSysHeaderMap(sysHeaderMap);
+        httpEventWrapper.setBody(cloudEvent.getData().toBytes());
+        return httpEventWrapper;
     }
 
     @Override
     public String getProtocolType() {
-        return "webhook";
+        return "webhookProtocolAdaptor";
     }
 
 }
diff --git a/...c/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HandlerService.java b/...c/main/java/org/apache/eventmesh/runtime/core/protocol/http/processor/HandlerService.java
@@ -94,7 +94,7 @@ public void register(String path, HttpProcessor httpProcessor, ThreadPoolExecuto
 
         if (httpProcessorMap.containsKey(path)) {
             throw new RuntimeException(String.format("HandlerService path %s repeat, repeat processor is %s ",
-                    path, httpProcessor.getClass().getSimpleName()));
+                path, httpProcessor.getClass().getSimpleName()));
         }
         ProcessorWrapper processorWrapper = new ProcessorWrapper();
         processorWrapper.threadPoolExecutor = threadPoolExecutor;
@@ -154,10 +154,10 @@ private void sendResponse(ChannelHandlerContext ctx, HttpRequest httpRequest, Ht
         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();
-                }
+                    RemotingHelper.parseChannelRemoteAddr(f.channel()));
+            }
+            if (isClose) {
+                f.channel().close();
             }
         });
     }

diff --git a/...e/src/test/java/org/apache/eventmesh/runtime/protocol/processor/WebHookProcessorTest.java b/...e/src/test/java/org/apache/eventmesh/runtime/protocol/processor/WebHookProcessorTest.java
@@ -82,7 +82,7 @@ public void testHandler() {
             CloudEvent msgSendToMq = captor.getValue();
             Assert.assertNotNull(msgSendToMq);
             Assert.assertTrue(StringUtils.isNoneBlank(msgSendToMq.getId()));
-            Assert.assertEquals("github", msgSendToMq.getSource().getPath());
+            Assert.assertEquals("www.github.com", msgSendToMq.getSource().getPath());
             Assert.assertEquals("github.ForkEvent", msgSendToMq.getType());
             Assert.assertEquals(BytesCloudEventData.wrap("\"mock_data\":0".getBytes()), msgSendToMq.getData());
         } catch (Exception e) {
@@ -97,7 +97,7 @@ private HttpRequest buildMockWebhookRequest() {
         FullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, "/webhook/github/eventmesh/all", buffer);
         request.headers().set("content-type", "application/json");
         // encrypt method see: GithubProtocol
-        request.headers().set("X-Hub-Signature-256", "sha256=ddb62e1182e2e6d364c0b5d03f2413fd5d1f68d99d1a4b3873e0d6850650d4b3");
+        request.headers().set("x-hub-signature-256", "sha256=ddb62e1182e2e6d364c0b5d03f2413fd5d1f68d99d1a4b3873e0d6850650d4b3");
         return request;
     }
 

diff --git a/eventmesh-webhook/build.gradle b/eventmesh-webhook/build.gradle
@@ -25,7 +25,7 @@ task copyEventMeshAdmin(dependsOn: ['jar']) {
             into('../eventmesh-webhook/dist/apps/')
             from project.jar.getArchivePath()
             exclude {
-                "eventmesh-webhook-${version}.jar"                
+                "eventmesh-webhook-${version}.jar"
             }
         }
         copy {

diff --git a/eventmesh-webhook/eventmesh-webhook-admin/build.gradle b/eventmesh-webhook/eventmesh-webhook-admin/build.gradle
@@ -18,15 +18,15 @@
 dependencies {
 
     implementation project(":eventmesh-common")
-	implementation project(":eventmesh-webhook:eventmesh-webhook-api")
+    implementation project(":eventmesh-webhook:eventmesh-webhook-api")
 
     implementation 'org.slf4j:slf4j-api'
     implementation "com.alibaba.nacos:nacos-client:2.0.4"
     implementation "com.fasterxml.jackson.core:jackson-databind"
     implementation "com.fasterxml.jackson.core:jackson-core"
     implementation "com.fasterxml.jackson.core:jackson-annotations"
 
-    testImplementation project(":eventmesh-webhook:eventmesh-webhook-api")   
+    testImplementation project(":eventmesh-webhook:eventmesh-webhook-api")
 
 
 }
diff --git a/...n/src/main/java/org/apache/eventmesh/webhook/admin/AdminWebHookConfigOperationManage.java b/...n/src/main/java/org/apache/eventmesh/webhook/admin/AdminWebHookConfigOperationManage.java
@@ -33,15 +33,14 @@
 
 public class AdminWebHookConfigOperationManage {
 
-    public Logger logger = LoggerFactory.getLogger(this.getClass());
-
     private static final Map<String, Class<? extends WebHookConfigOperation>> map = new HashMap<>();
 
     static {
         map.put("file", FileWebHookConfigOperation.class);
         map.put("nacos", NacosWebHookConfigOperation.class);
     }
 
+    public Logger logger = LoggerFactory.getLogger(this.getClass());
     private ConfigurationWrapper configurationWrapper;
 
     private WebHookConfigOperation webHookConfigOperation;

diff --git a/...ok-admin/src/main/java/org/apache/eventmesh/webhook/admin/FileWebHookConfigOperation.java b/...ok-admin/src/main/java/org/apache/eventmesh/webhook/admin/FileWebHookConfigOperation.java
@@ -21,6 +21,7 @@
 import org.apache.eventmesh.webhook.api.WebHookConfig;
 import org.apache.eventmesh.webhook.api.WebHookConfigOperation;
 import org.apache.eventmesh.webhook.api.WebHookOperationConstant;
+import org.apache.eventmesh.webhook.api.utils.StringUtils;
 
 import java.io.BufferedReader;
 import java.io.BufferedWriter;
@@ -31,8 +32,6 @@
 import java.io.IOException;
 import java.io.InputStreamReader;
 import java.io.OutputStreamWriter;
-import java.io.UnsupportedEncodingException;
-import java.net.URLEncoder;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Properties;
@@ -166,14 +165,9 @@ private String getWebhookConfigManuDir(WebHookConfig webHookConfig) {
 
     private File getWebhookConfigFile(WebHookConfig webHookConfig) {
         String webhookConfigFilePath = null;
-        try {
-            // use URLEncoder.encode before, because the path may contain some speacial char like '/', which is illegal as a file name.
-            webhookConfigFilePath = this.getWebhookConfigManuDir(webHookConfig)
-                + WebHookOperationConstant.FILE_SEPARATOR + URLEncoder.encode(webHookConfig.getCallbackPath(), "UTF-8")
-                + WebHookOperationConstant.FILE_EXTENSION;
-        } catch (UnsupportedEncodingException e) {
-            logger.error("get webhookConfig file path {} failed", webHookConfig.getCallbackPath(), e);
-        }
+        webhookConfigFilePath = this.getWebhookConfigManuDir(webHookConfig) + WebHookOperationConstant.FILE_SEPARATOR
+            + StringUtils.getFileName(webHookConfig.getCallbackPath());
+
         assert webhookConfigFilePath != null;
         return new File(webhookConfigFilePath);
     }

diff --git a/...k-admin/src/main/java/org/apache/eventmesh/webhook/admin/NacosWebHookConfigOperation.java b/...k-admin/src/main/java/org/apache/eventmesh/webhook/admin/NacosWebHookConfigOperation.java
@@ -23,7 +23,6 @@
 import static org.apache.eventmesh.webhook.api.WebHookOperationConstant.TIMEOUT_MS;
 
 
-
 import org.apache.eventmesh.common.utils.JsonUtils;
 import org.apache.eventmesh.webhook.api.ManufacturerObject;
 import org.apache.eventmesh.webhook.api.WebHookConfig;
@@ -80,7 +79,7 @@ public Integer insertWebHookConfig(WebHookConfig webHookConfig) {
                 return 0;
             }
             result = configService.publishConfig(getWebHookConfigDataId(webHookConfig), getManuGroupId(webHookConfig),
-                    JsonUtils.serialize(webHookConfig), ConfigType.JSON.getType());
+                JsonUtils.serialize(webHookConfig), ConfigType.JSON.getType());
         } catch (NacosException e) {
             logger.error("insertWebHookConfig failed", e);
             return 0;
@@ -92,7 +91,7 @@ public Integer insertWebHookConfig(WebHookConfig webHookConfig) {
                 manufacturerObject.addManufacturer(manufacturerName);
                 manufacturerObject.getManufacturerEvents(manufacturerName).add(getWebHookConfigDataId(webHookConfig));
                 configService.publishConfig(MANUFACTURERS_DATA_ID, CONSTANTS_WEBHOOK,
-                        JsonUtils.serialize(manufacturerObject), ConfigType.JSON.getType());
+                    JsonUtils.serialize(manufacturerObject), ConfigType.JSON.getType());
             } catch (NacosException e) {
                 logger.error("update manufacturersInfo error", e);
                 //rollback insert
@@ -115,7 +114,7 @@ public Integer updateWebHookConfig(WebHookConfig webHookConfig) {
                 return 0;
             }
             result = configService.publishConfig(getWebHookConfigDataId(webHookConfig),
-                    getManuGroupId(webHookConfig), JsonUtils.serialize(webHookConfig), ConfigType.JSON.getType());
+                getManuGroupId(webHookConfig), JsonUtils.serialize(webHookConfig), ConfigType.JSON.getType());
         } catch (NacosException e) {
             logger.error("updateWebHookConfig failed", e);
         }
@@ -136,7 +135,7 @@ public Integer deleteWebHookConfig(WebHookConfig webHookConfig) {
                 ManufacturerObject manufacturerObject = getManufacturersInfo();
                 manufacturerObject.getManufacturerEvents(manufacturerName).remove(getWebHookConfigDataId(webHookConfig));
                 configService.publishConfig(MANUFACTURERS_DATA_ID, CONSTANTS_WEBHOOK,
-                        JsonUtils.serialize(manufacturerObject), ConfigType.JSON.getType());
+                    JsonUtils.serialize(manufacturerObject), ConfigType.JSON.getType());
             } catch (NacosException e) {
                 logger.error("update manufacturersInfo error", e);
             }
@@ -170,7 +169,7 @@ public List<WebHookConfig> queryWebHookConfigByManufacturer(WebHookConfig webHoo
                 // nacos API is not able to get all config, so use foreach
                 for (int i = startIndex; i < endIndex && i < manufacturerEvents.size(); i++) {
                     String content = configService.getConfig(manufacturerEvents.get(i) + DATA_ID_EXTENSION,
-                            getManuGroupId(webHookConfig), TIMEOUT_MS);
+                        getManuGroupId(webHookConfig), TIMEOUT_MS);
                     webHookConfigs.add(JsonUtils.deserialize(content, WebHookConfig.class));
                 }
             }
@@ -201,7 +200,7 @@ private String getManuGroupId(WebHookConfig webHookConfig) {
     private ManufacturerObject getManufacturersInfo() throws NacosException {
         String manufacturersContent = configService.getConfig(MANUFACTURERS_DATA_ID, CONSTANTS_WEBHOOK, TIMEOUT_MS);
         return StringUtil.isNullOrEmpty(manufacturersContent)
-                ? new ManufacturerObject() : JsonUtils.deserialize(manufacturersContent, ManufacturerObject.class);
+            ? new ManufacturerObject() : JsonUtils.deserialize(manufacturersContent, ManufacturerObject.class);
     }
 
 }
diff --git a/...-webhook-api/src/main/java/org/apache/eventmesh/webhook/api/WebHookOperationConstant.java b/...-webhook-api/src/main/java/org/apache/eventmesh/webhook/api/WebHookOperationConstant.java
@@ -42,8 +42,8 @@ public class WebHookOperationConstant {
     public static final String getFilePath(String filePath) {
         if (filePath.startsWith("#{eventMeshHome}")) {
             String configPath = System.getProperty("confPath", System.getenv("confPath"));
-           
-            filePath = filePath.replace("#{eventMeshHome}", configPath.substring(0,  configPath.lastIndexOf(FILE_SEPARATOR)));
+
+            filePath = filePath.replace("#{eventMeshHome}", configPath.substring(0, configPath.lastIndexOf(FILE_SEPARATOR)));
         }
         return filePath;
     }

diff --git a/...entmesh-webhook-api/src/main/java/org/apache/eventmesh/webhook/api/utils/StringUtils.java b/...entmesh-webhook-api/src/main/java/org/apache/eventmesh/webhook/api/utils/StringUtils.java
@@ -0,0 +1,25 @@
+/*
+ * 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.webhook.api.utils;
+
+public class StringUtils {
+
+    public static final String getFileName(String path) {
+        return path.substring(1).replace('/', '.');
+    }
+}
diff --git a/...webhook-receive/src/main/java/org/apache/eventmesh/webhook/receive/WebHookController.java b/...webhook-receive/src/main/java/org/apache/eventmesh/webhook/receive/WebHookController.java
@@ -41,13 +41,11 @@
 
 public class WebHookController {
 
-    public Logger logger = LoggerFactory.getLogger(this.getClass());
-
     /**
      * protocol pool
      */
     private final ProtocolManage protocolManage = new ProtocolManage();
-
+    public Logger logger = LoggerFactory.getLogger(this.getClass());
     /**
      * config pool
      */
@@ -61,7 +59,7 @@ public class WebHookController {
     private ConfigurationWrapper configurationWrapper;
 
     public void init() throws Exception {
-        this.webHookMQProducer = new WebHookMQProducer(
+        this.webHookMQProducer = new WebHookMQProducer(configurationWrapper.getProperties(),
             configurationWrapper.getProp("eventMesh.webHook.producer.connector"));
         this.hookConfigOperationManage = new HookConfigOperationManage(configurationWrapper);
         this.protocolAdaptor = ProtocolPluginFactory.getProtocolAdaptor("webhook");
@@ -98,7 +96,7 @@ public void execute(String path, Map<String, String> header, byte[] body) throws
         try {
             protocol.execute(webHookRequest, webHookConfig, header);
         } catch (Exception e) {
-            throw new Exception("Webhook Message Parse Failed.");
+            throw new Exception("Webhook Message Parse Failed. " + e.getMessage(), e);
         }
 
         // 3. convert to cloudEvent obj
@@ -107,13 +105,9 @@ public void execute(String path, Map<String, String> header, byte[] body) throws
         String eventType = manufacturerName + "." + webHookConfig.getManufacturerEventName();
 
         WebhookProtocolTransportObject webhookProtocolTransportObject = WebhookProtocolTransportObject.builder()
-            .cloudEventId(cloudEventId)
-            .eventType(eventType)
-            .cloudEventName(webHookConfig.getCloudEventName())
-            .cloudEventSource(webHookConfig.getCloudEventSource())
-            .dataContentType(webHookConfig.getDataContentType())
-            .body(body)
-            .build();
+            .cloudEventId(cloudEventId).eventType(eventType).cloudEventName(webHookConfig.getCloudEventName())
+            .cloudEventSource("www." + webHookConfig.getManufacturerName() + ".com")
+            .dataContentType(webHookConfig.getDataContentType()).body(body).build();
 
         // 4. send cloudEvent
         webHookMQProducer.send(this.protocolAdaptor.toCloudEvent(webhookProtocolTransportObject), new SendCallback() {