[ISSUE#4419] Add slack sink connector.

eventmesh-connectors/README.md

@@ -18,50 +18,52 @@ Add a new connector by implementing the source/sink interface using :
 ## Connector Status
 
 
-|                  Connector Name                  |       Type        |  Status  |
-|:------------------------------------------------:|:-----------------:|:--------:|
-|     [RocketMQ](eventmesh-connector-rocketmq)     |      Source       |    ✅     |
-|     [RocketMQ](eventmesh-connector-rocketmq)     |       Sink        |    ✅     |
-|                     ChatGPT                      |      Source       |    ⬜     |
-|                     ChatGPT                      |       Sink        |    ⬜     |
-|                    ClickHouse                    |      Source       |    ⬜     |
-|                    ClickHouse                    |       Sink        |    ⬜     |
-|                     DingDing                     |      Source       |    ⬜     |
-|     [DingDing](eventmesh-connector-dingding)     |       Sink        |    ✅     |
-|                      Email                       |      Source       |    ⬜     |
-|                      Email                       |       Sink        |    ⬜     |
-|                      FeiShu                      |      Source       |    ⬜     |
-|                      FeiShu                      |       Sink        |    ⬜     |
-|         [File](eventmesh-connector-file)         |      Source       |    ✅     |
-|         [File](eventmesh-connector-file)         |       Sink        |    ✅     |
-|                      Github                      |      Source       |    ⬜     |
-|                      Github                      |       Sink        |    ⬜     |
-|                       Http                       |      Source       |    ⬜     |
-|                       Http                       |       Sink        |    ⬜     |
-|                       Jdbc                       |      Source       |    ⬜     |
-|         [Jdbc](eventmesh-connector-jdbc)         |       Sink        |    ✅     |
-|        [Kafka](eventmesh-connector-kafka)        |      Source       |    ✅     |
-|        [Kafka](eventmesh-connector-kafka)        |       Sink        |    ✅     |
-|      [Knative](eventmesh-connector-knative)      |      Source       |    ✅     |
-|      [Knative](eventmesh-connector-knative)      |       Sink        |    ✅     |
-|      [MongoDB](eventmesh-connector-mongodb)      |      Source       |    ✅     |
-|      [MongoDB](eventmesh-connector-mongodb)      |       Sink        |    ✅     |
-| [OpenFunction](eventmesh-connector-openfunction) |      Source       |    ✅     |
-| [OpenFunction](eventmesh-connector-openfunction) |       Sink        |    ✅     |
-|      [Pravega](eventmesh-connector-pravega)      |      Source       |    ✅     |
-|      [Pravega](eventmesh-connector-pravega)      |       Sink        |    ✅     |
-|   [Promethues](eventmesh-connector-prometheus)   |      Source       |    ✅     |
-|   [Promethues](eventmesh-connector-prometheus)   |       Sink        |    ⬜     |
-|       [Pulsar](eventmesh-connector-pulsar)       |      Source       |    ✅     |
-|       [Pulsar](eventmesh-connector-pulsar)       |       Sink        |    ✅     |
-|     [Rabbitmq](eventmesh-connector-rabbitmq)     |      Source       |    ✅     |
-|     [Rabbitmq](eventmesh-connector-rabbitmq)     |       Sink        |    ✅     |
-|        [Redis](eventmesh-connector-redis)        |      Source       |    ✅     |
-|        [Redis](eventmesh-connector-redis)        |       Sink        |    ✅     |
-|                      S3File                      |      Source       |    ⬜     |
-|         [S3File](eventmesh-connector-s3)         |       Sink        |    ✅     |
-|       [Spring](eventmesh-connector-spring)       |      Source       |    ✅     |
-|       [Spring](eventmesh-connector-spring)       |       Sink        |    ✅     |
-|                      WeCom                       |      Source       |    ⬜     |
-|        [WeCom](eventmesh-connector-wecom)        |       Sink        |    ✅     |
-|         More connectors will be added...         |    Source/Sink    |   N/A    |
+|                  Connector Name                  |    Type     | Status  |
+|:------------------------------------------------:|:-----------:|:-------:|
+|     [RocketMQ](eventmesh-connector-rocketmq)     |   Source    |    ✅    |
+|     [RocketMQ](eventmesh-connector-rocketmq)     |    Sink     |    ✅    |
+|                     ChatGPT                      |   Source    |    ⬜    |
+|                     ChatGPT                      |    Sink     |    ⬜    |
+|                    ClickHouse                    |   Source    |    ⬜    |
+|                    ClickHouse                    |    Sink     |    ⬜    |
+|                     DingDing                     |   Source    |    ⬜    |
+|     [DingDing](eventmesh-connector-dingding)     |    Sink     |    ✅    |
+|                      Email                       |   Source    |    ⬜    |
+|                      Email                       |    Sink     |    ⬜    |
+|                      FeiShu                      |   Source    |    ⬜    |
+|                      FeiShu                      |    Sink     |    ⬜    |
+|         [File](eventmesh-connector-file)         |   Source    |    ✅    |
+|         [File](eventmesh-connector-file)         |    Sink     |    ✅    |
+|                      Github                      |   Source    |    ⬜    |
+|                      Github                      |    Sink     |    ⬜    |
+|                       Http                       |   Source    |    ⬜    |
+|                       Http                       |    Sink     |    ⬜    |
+|                       Jdbc                       |   Source    |    ⬜    |
+|         [Jdbc](eventmesh-connector-jdbc)         |    Sink     |    ✅    |
+|        [Kafka](eventmesh-connector-kafka)        |   Source    |    ✅    |
+|        [Kafka](eventmesh-connector-kafka)        |    Sink     |    ✅    |
+|      [Knative](eventmesh-connector-knative)      |   Source    |    ✅    |
+|      [Knative](eventmesh-connector-knative)      |    Sink     |    ✅    |
+|      [MongoDB](eventmesh-connector-mongodb)      |   Source    |    ✅    |
+|      [MongoDB](eventmesh-connector-mongodb)      |    Sink     |    ✅    |
+| [OpenFunction](eventmesh-connector-openfunction) |   Source    |    ✅    |
+| [OpenFunction](eventmesh-connector-openfunction) |    Sink     |    ✅    |
+|      [Pravega](eventmesh-connector-pravega)      |   Source    |    ✅    |
+|      [Pravega](eventmesh-connector-pravega)      |    Sink     |    ✅    |
+|   [Promethues](eventmesh-connector-prometheus)   |   Source    |    ✅    |
+|   [Promethues](eventmesh-connector-prometheus)   |    Sink     |    ⬜    |
+|       [Pulsar](eventmesh-connector-pulsar)       |   Source    |    ✅    |
+|       [Pulsar](eventmesh-connector-pulsar)       |    Sink     |    ✅    |
+|     [Rabbitmq](eventmesh-connector-rabbitmq)     |   Source    |    ✅    |
+|     [Rabbitmq](eventmesh-connector-rabbitmq)     |    Sink     |    ✅    |
+|        [Redis](eventmesh-connector-redis)        |   Source    |    ✅    |
+|        [Redis](eventmesh-connector-redis)        |    Sink     |    ✅    |
+|                      S3File                      |   Source    |    ⬜    |
+|         [S3File](eventmesh-connector-s3)         |    Sink     |    ✅    |
+|        [Slack](eventmesh-connector-slack)        |   Source    |    ⬜    |
+|        [Slack](eventmesh-connector-slack)        |    Sink     |    ✅    |
+|       [Spring](eventmesh-connector-spring)       |   Source    |    ✅    |
+|       [Spring](eventmesh-connector-spring)       |    Sink     |    ✅    |
+|                      WeCom                       |   Source    |    ⬜    |
+|        [WeCom](eventmesh-connector-wecom)        |    Sink     |    ✅    |
+|         More connectors will be added...         | Source/Sink |   N/A   |

eventmesh-connectors/eventmesh-connector-slack/build.gradle

@@ -0,0 +1,37 @@
+/*
+ * 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.
+ */
+
+configurations {
+    implementation.exclude group: 'ch.qos.logback', module: 'logback-classic'
+    implementation.exclude group: 'log4j', module: 'log4j'
+    testImplementation.exclude group: 'org.apache.logging.log4j', module: 'log4j-to-slf4j'
+}
+
+dependencies {
+    implementation project(":eventmesh-common")
+    implementation project(":eventmesh-sdks:eventmesh-sdk-java")
+    implementation project(":eventmesh-openconnect:eventmesh-openconnect-java")
+
+    implementation "com.slack.api:bolt:1.1.+"
+    implementation 'com.google.guava:guava'
+
+    compileOnly 'org.projectlombok:lombok'
+    annotationProcessor 'org.projectlombok:lombok'
+
+    testImplementation "org.mockito:mockito-core"
+    testImplementation "org.mockito:mockito-junit-jupiter"
+}

eventmesh-connectors/eventmesh-connector-slack/src/main/java/org/apache/eventmesh/connector/slack/config/SlackConnectServerConfig.java

@@ -0,0 +1,30 @@
+/*
+ * 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.connector.slack.config;
+
+import org.apache.eventmesh.openconnect.api.config.Config;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+@Data
+@EqualsAndHashCode(callSuper = true)
+public class SlackConnectServerConfig extends Config {
+
+    private boolean sinkEnable;
+}

eventmesh-connectors/eventmesh-connector-slack/src/main/java/org/apache/eventmesh/connector/slack/server/SlackConnectServer.java

@@ -0,0 +1,38 @@
+/*
+ * 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.connector.slack.server;
+
+import org.apache.eventmesh.common.Constants;
+import org.apache.eventmesh.connector.slack.config.SlackConnectServerConfig;
+import org.apache.eventmesh.connector.slack.sink.connector.SlackSinkConnector;
+import org.apache.eventmesh.openconnect.Application;
+import org.apache.eventmesh.openconnect.util.ConfigUtil;
+
+public class SlackConnectServer {
+
+    public static void main(String[] args) throws Exception {
+
+        SlackConnectServerConfig slackConnectServerConfig = ConfigUtil.parse(SlackConnectServerConfig.class,
+            Constants.CONNECT_SERVER_CONFIG_FILE_NAME);
+
+        if (slackConnectServerConfig.isSinkEnable()) {
+            Application application = new Application();
+            application.run(SlackSinkConnector.class);
+        }
+    }
+}

eventmesh-connectors/eventmesh-connector-slack/src/main/java/org/apache/eventmesh/connector/slack/sink/config/SinkConnectorConfig.java

@@ -0,0 +1,30 @@
+/*
+ * 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.connector.slack.sink.config;
+
+import lombok.Data;
+
+@Data
+public class SinkConnectorConfig {
+
+    private String connectorName;
+
+    private String appToken;
+
+    private String channelId;
+}

eventmesh-connectors/eventmesh-connector-slack/src/main/java/org/apache/eventmesh/connector/slack/sink/config/SlackSinkConfig.java

@@ -0,0 +1,30 @@
+/*
+ * 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.connector.slack.sink.config;
+
+import org.apache.eventmesh.openconnect.api.config.SinkConfig;
+
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+
+@Data
+@EqualsAndHashCode(callSuper = true)
+public class SlackSinkConfig extends SinkConfig {
+
+    private SinkConnectorConfig sinkConnectorConfig;
+}

eventmesh-connectors/eventmesh-connector-slack/src/main/java/org/apache/eventmesh/connector/slack/sink/connector/SlackSinkConnector.java

@@ -0,0 +1,117 @@
+/*
+ * 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.connector.slack.sink.connector;
+
+import org.apache.eventmesh.connector.slack.sink.config.SlackSinkConfig;
+import org.apache.eventmesh.openconnect.api.config.Config;
+import org.apache.eventmesh.openconnect.api.connector.ConnectorContext;
+import org.apache.eventmesh.openconnect.api.connector.SinkConnectorContext;
+import org.apache.eventmesh.openconnect.api.sink.Sink;
+import org.apache.eventmesh.openconnect.offsetmgmt.api.data.ConnectRecord;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.List;
+import java.util.Objects;
+
+import com.slack.api.Slack;
+import com.slack.api.methods.MethodsClient;
+import com.slack.api.methods.response.chat.ChatPostMessageResponse;
+
+import lombok.SneakyThrows;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * Slack sink connector.
+ * Slack doc: <a href="https://api.slack.com/messaging/sending">...</a>
+ */
+@Slf4j
+public class SlackSinkConnector implements Sink {
+
+    private SlackSinkConfig sinkConfig;
+
+    private volatile boolean isRunning = false;
+
+    private MethodsClient client;
+
+    @Override
+    public Class<? extends Config> configClass() {
+        return SlackSinkConfig.class;
+    }
+
+    @Override
+    public void init(Config config) throws Exception {
+        // init config for dingding sink connector
+        this.sinkConfig = (SlackSinkConfig) config;
+        this.client = Slack.getInstance().methods();
+    }
+
+    @Override
+    public void init(ConnectorContext connectorContext) throws Exception {
+        // init config for dingding source connector
+        SinkConnectorContext sinkConnectorContext = (SinkConnectorContext) connectorContext;
+        this.sinkConfig = (SlackSinkConfig) sinkConnectorContext.getSinkConfig();
+        this.client = Slack.getInstance().methods();
+    }
+
+    @Override
+    public void start() {
+        isRunning = true;
+    }
+
+    @Override
+    public void commit(ConnectRecord record) {
+
+    }
+
+    @Override
+    public String name() {
+        return this.sinkConfig.getSinkConnectorConfig().getConnectorName();
+    }
+
+    @Override
+    public void stop() {
+        isRunning = false;
+    }
+
+    public boolean isRunning() {
+        return isRunning;
+    }
+
+    @SneakyThrows
+    @Override
+    public void put(List<ConnectRecord> sinkRecords) {
+        for (ConnectRecord record : sinkRecords) {
+            publishMessage(record);
+        }
+    }
+
+    private void publishMessage(ConnectRecord record) {
+        try {
+            ChatPostMessageResponse response = client.chatPostMessage(r -> r
+                .token(sinkConfig.getSinkConnectorConfig().getAppToken())
+                .channel(sinkConfig.getSinkConnectorConfig().getChannelId())
+                .text(new String((byte[]) record.getData())));
+            if (Objects.nonNull(response) && StringUtils.isNotBlank(response.getError())) {
+                throw new IllegalAccessException(response.getError());
+            }
+        } catch (Exception e) {
+            log.error("Send message to slack error.", e);
+        }
+    }
+}