Adding slack sink connector

README.md

@@ -64,6 +64,10 @@ A Hazelcast Jet connector for consuming data from Twitter stream sources in Jet
 Tests to check compatibility of the XA support in your JMS broker or
 JDBC database with Jet's fault tolerance.
 
+### [Slack Connector](slack) 
+
+A Hazelcast Jet slack connector to post processed messages to slack channels using slack sink in the Jet pipelines.
+
 ## Snapshot Releases
 
 To access snapshot builds add the following `dependency` and `repository` declarations

settings.gradle

@@ -10,6 +10,7 @@ include 'mongodb'
 include 'debezium'
 include 'twitter'
 include 'xa-test'
+include 'slack'
 
 project(':elasticsearch-5').projectDir = file('elasticsearch/elasticsearch-5')
 project(':elasticsearch-6').projectDir = file('elasticsearch/elasticsearch-6')

slack/build.gradle

@@ -0,0 +1,15 @@
+plugins {
+    id 'java'
+}
+
+
+sourceCompatibility = 1.8
+
+repositories {
+    mavenCentral()
+}
+
+dependencies {
+    compile group: 'org.apache.httpcomponents', name: 'httpclient', version: '4.5.12'
+    testCompile group: 'junit', name: 'junit', version: '4.12'
+}

slack/src/main/java/com/hazelcast/jet/contrib/slack/SlackSinks.java

@@ -0,0 +1,90 @@
+/*
+ * Copyright (c) 2008-2020, Hazelcast, Inc. All Rights Reserved.
+ *
+ * Licensed 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 com.hazelcast.jet.contrib.slack;
+
+import com.hazelcast.function.SupplierEx;
+import com.hazelcast.jet.pipeline.Sink;
+import com.hazelcast.jet.pipeline.SinkBuilder;
+import org.apache.http.NameValuePair;
+import org.apache.http.client.entity.UrlEncodedFormEntity;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpPost;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.message.BasicNameValuePair;
+import org.apache.http.util.EntityUtils;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Contains the methods to create Slack sinks.
+ */
+public final class SlackSinks {
+
+    private static final String URL = "https://slack.com/api/chat.postMessage";
+
+    private SlackSinks() {
+    }
+
+    /**
+     * Creates a sink to send messages to the slack channel>
+     *
+     * @param accessToken String Bearer token to authenticate the slack web api requests
+     * @param channel String Unique channel id to send messages to the slack channel.
+     * @return
+     */
+    public static Sink sink(String accessToken, String channel) {
+        return SinkBuilder.sinkBuilder("slack-sink",
+                ctx -> new SlackContext(() -> HttpClients.createDefault(), accessToken, channel))
+                .<String>receiveFn((ctx, item) -> ctx.receiveFn(item)).destroyFn(ctx -> ctx.destroy()).build();
+    }
+
+    private static final class SlackContext {
+
+       private final SupplierEx<CloseableHttpClient> httpClientSupplierEx;
+       private final CloseableHttpClient closeableHttpClient;
+       private final String accessToken;
+       private final String channel;
+
+        private SlackContext(SupplierEx<CloseableHttpClient> bulkRequestSupplier, String accessToken, String channel) {
+            this.httpClientSupplierEx = bulkRequestSupplier;
+            this.closeableHttpClient = bulkRequestSupplier.get();
+            this.accessToken = accessToken;
+            this.channel = channel;
+        }
+
+        private String receiveFn(String message) throws IOException {
+            HttpPost request = new HttpPost(URL);
+            // add request headers
+            request.addHeader("Authorization", String.format("Bearer %s", accessToken));
+            List<NameValuePair> urlParameters = new ArrayList<>();
+            urlParameters.add(new BasicNameValuePair("channel",  channel));
+            urlParameters.add(new BasicNameValuePair("text", message));
+            request.setEntity(new UrlEncodedFormEntity(urlParameters));
+            CloseableHttpResponse response = (CloseableHttpResponse) closeableHttpClient.execute(request);
+            String result = EntityUtils.toString(response.getEntity());
+            request.releaseConnection();
+            return result;
+        }
+
+        private void destroy() throws IOException {
+            closeableHttpClient.close();
+        }
+    }
+}

slack/src/main/java/com/hazelcast/jet/contrib/slack/package-info.java

@@ -0,0 +1,20 @@
+/*
+ * Copyright (c) 2008-2020, Hazelcast, Inc. All Rights Reserved.
+ *
+ * Licensed 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.
+ */
+
+/**
+ * Connector for Hazelcast Jet to push messages to Slack sink.
+ */
+package com.hazelcast.jet.contrib.slack;

slack/src/test/java/com/hazelcast/jet/contrib/slack/SlackSInksTest.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright (c) 2008-2020, Hazelcast, Inc. All Rights Reserved.
+ *
+ * Licensed 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 com.hazelcast.jet.contrib.slack;
+
+import com.hazelcast.function.Functions;
+import com.hazelcast.jet.JetInstance;
+import com.hazelcast.jet.Traversers;
+import com.hazelcast.jet.aggregate.AggregateOperations;
+import com.hazelcast.jet.core.JetTestSupport;
+import com.hazelcast.jet.pipeline.BatchStage;
+import com.hazelcast.jet.pipeline.Pipeline;
+import com.hazelcast.jet.pipeline.Sources;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+public final class SlackSInksTest extends JetTestSupport {
+
+    private JetInstance jet;
+    private Properties credentials;
+
+    @Before
+    public void setup() {
+        jet = createJetMember();
+    }
+
+    @Test
+    public void testSlackSink() {
+        List<String> text = jet.getList("text");
+        text.add("hello world hello world hazelcast");
+        text.add("sample message to slack channel");
+
+        Pipeline pipeline = Pipeline.create();
+        BatchStage<Map.Entry<String, Long>> tweets = pipeline
+                .readFrom(Sources.<String>list("text"))
+                .flatMap(line -> Traversers.traverseArray(line.toLowerCase().split("\\W+")))
+                .filter(word -> !word.isEmpty())
+                .groupingKey(Functions.wholeItem())
+                .aggregate(AggregateOperations.counting());
+
+        Pipeline p = Pipeline.create();
+        p.readFrom(Sources.<String>list("text"))
+                .writeTo(SlackSinks.sink(System.getenv("ACCESS_TOKEN"), System.getenv("CHANNEL_ID")));
+        jet.newJob(p).join();
+    }
+}