[Feature][Connector2] Add Dingtalk Sink apache#2257

MRYOG · Aug 1, 2022 · 7fc71a4 · 7fc71a4
1 parent fd698eb
commit 7fc71a4
Show file tree

Hide file tree

Showing 2 changed files with 118 additions and 118 deletions.
diff --git a/...r-dingtalk/src/main/java/org/apache/seatunnel/connectors/seatunnel/sink/DingTalkSink.java b/...r-dingtalk/src/main/java/org/apache/seatunnel/connectors/seatunnel/sink/DingTalkSink.java
@@ -1,58 +1,61 @@
 /*
-  * 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.seatunnel.connectors.seatunnel.sink;
-
- import org.apache.seatunnel.api.common.PrepareFailException;
- import org.apache.seatunnel.api.sink.SeaTunnelSink;
- import org.apache.seatunnel.api.sink.SinkWriter;
- import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
- import org.apache.seatunnel.api.table.type.SeaTunnelRow;
- import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
- import org.apache.seatunnel.common.constants.PluginType;
- import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSimpleSink;
- import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSinkWriter;
-
- import org.apache.seatunnel.shade.com.typesafe.config.Config;
-
- import com.google.auto.service.AutoService;
-
- import java.io.IOException;
-
- /**
-  * DingTalk sink class
-  */
- @AutoService(SeaTunnelSink.class)
- public class DingTalkSink extends AbstractSimpleSink<SeaTunnelRow, Void> {
-   private Config pluginConfig;
-   private SeaTunnelRowType seaTunnelRowType;
-   private final String CONSTANT_URL="url";
-   private final String CONSTANT_SECRET="secret";
-   @Override
-   public String getPluginName() {
-     return "DingTalk";
-   }
-
-   @Override
-   public void prepare(Config pluginConfig) throws PrepareFailException {
-     if(pluginConfig.getIsNull(CONSTANT_URL)){
-       throw new PrepareFailException(getPluginName(), PluginType.SINK,  String.format("Config must include column : %s", CONSTANT_URL));
-     }
-     if(pluginConfig.getIsNull(CONSTANT_SECRET)){
-       throw new PrepareFailException(getPluginName(), PluginType.SINK,  String.format("Config must include column : %s", CONSTANT_SECRET));
-     }
-     this.pluginConfig = pluginConfig;
+ * 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.seatunnel.connectors.seatunnel.sink;
+
+import org.apache.seatunnel.api.common.PrepareFailException;
+import org.apache.seatunnel.api.sink.SeaTunnelSink;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
+import org.apache.seatunnel.common.constants.PluginType;
+import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSimpleSink;
+
+import org.apache.seatunnel.shade.com.typesafe.config.Config;
+
+import com.google.auto.service.AutoService;
+
+
+/**
+ * DingTalk sink class
+ */
+@AutoService(SeaTunnelSink.class)
+public class DingTalkSink extends AbstractSimpleSink<SeaTunnelRow, Void> {
+
+    private Config pluginConfig;
+    private SeaTunnelRowType seaTunnelRowType;
+    private final String dtURL = "url";
+    private final String dtSecret = "secret";
+
+    @Override
+    public String getPluginName() {
+        return "DingTalk";
+    }
+
+    @Override
+    public void prepare(Config pluginConfig) throws PrepareFailException {
+        if (pluginConfig.getIsNull(dtURL)) {
+            throw new PrepareFailException(getPluginName(), PluginType.SINK,
+                String.format("Config must include column : %s", dtURL));
+        }
+        if (pluginConfig.getIsNull(dtSecret)) {
+            throw new PrepareFailException(getPluginName(), PluginType.SINK,
+                String.format("Config must include column : %s", dtSecret));
+        }
+        this.pluginConfig = pluginConfig;
+    }
+
+}
diff --git a/...dingtalk/src/main/java/org/apache/seatunnel/connectors/seatunnel/sink/DingTalkWriter.java b/...dingtalk/src/main/java/org/apache/seatunnel/connectors/seatunnel/sink/DingTalkWriter.java
@@ -17,90 +17,87 @@
 
 package org.apache.seatunnel.connectors.seatunnel.sink;
 
+import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSinkWriter;
 import java.nio.charset.StandardCharsets;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
-import org.apache.seatunnel.connectors.seatunnel.common.sink.AbstractSinkWriter;
-
 import com.dingtalk.api.DefaultDingTalkClient;
 import com.dingtalk.api.request.OapiRobotSendRequest;
 import com.dingtalk.api.response.OapiRobotSendResponse;
 import com.taobao.api.ApiException;
-
 import javax.crypto.Mac;
 import javax.crypto.spec.SecretKeySpec;
-
 import java.io.IOException;
 import java.io.Serializable;
 import java.net.URLEncoder;
 import java.util.Base64;
 
 /**
- *  DingTalk write class
+ * DingTalk write class
  */
 public class DingTalkWriter extends AbstractSinkWriter<SeaTunnelRow, Void> {
 
-  private RobotClient robotClient;
-
-  public DingTalkWriter(String url, String secret) {
-    this.robotClient = new RobotClient(url, secret);
-  }
-
-  @Override
-  public void write(SeaTunnelRow element) throws IOException {
-    robotClient.send(element.toString());
-  }
+    private RobotClient robotClient;
 
-  @Override
-  public void close() throws IOException {
-
-  }
-
-  private static class RobotClient implements Serializable {
-
-    private String url;
-
-    private String secret;
-
-    private DefaultDingTalkClient client;
-
-    public RobotClient(String url, String secret) {
-      this.url = url;
-      this.secret = secret;
+    public DingTalkWriter(String url, String secret) {
+        this.robotClient = new RobotClient(url, secret);
     }
 
-    public OapiRobotSendResponse send(String message) throws IOException {
-      if (null == client) {
-        client = new DefaultDingTalkClient(getUrl());
-      }
-      OapiRobotSendRequest request = new OapiRobotSendRequest();
-      request.setMsgtype("text");
-      OapiRobotSendRequest.Text text = new OapiRobotSendRequest.Text();
-      text.setContent(message);
-      request.setText(text);
-      try {
-        return this.client.execute(request);
-      } catch (ApiException e) {
-        throw new IOException(e);
-      }
+    @Override
+    public void write(SeaTunnelRow element) throws IOException {
+        robotClient.send(element.toString());
     }
 
-    public String getUrl() {
-      Long timestamp = System.currentTimeMillis();
-      String sign = getSign(timestamp);
-      return url + "&timestamp=" + timestamp + "&sign=" + sign;
+    @Override
+    public void close() throws IOException {
+
     }
 
-    public String getSign(Long timestamp) {
-      try {
-        String stringToSign = timestamp + "\n" + secret;
-        Mac mac = Mac.getInstance("HmacSHA256");
-        mac.init(new SecretKeySpec(secret.getBytes(StandardCharsets.UTF_8), "HmacSHA256"));
-        byte[] signData = mac.doFinal(stringToSign.getBytes(StandardCharsets.UTF_8));
-        return URLEncoder.encode(Base64.getEncoder().encodeToString(signData), "UTF-8");
-      } catch (Exception e) {
-        return null;
-      }
+    private static class RobotClient implements Serializable {
+
+        private String url;
+
+        private String secret;
+
+        private DefaultDingTalkClient client;
+
+        public RobotClient(String url, String secret) {
+            this.url = url;
+            this.secret = secret;
+        }
+
+        public OapiRobotSendResponse send(String message) throws IOException {
+            if (null == client) {
+                client = new DefaultDingTalkClient(getUrl());
+            }
+            OapiRobotSendRequest request = new OapiRobotSendRequest();
+            request.setMsgtype("text");
+            OapiRobotSendRequest.Text text = new OapiRobotSendRequest.Text();
+            text.setContent(message);
+            request.setText(text);
+            try {
+                return this.client.execute(request);
+            } catch (ApiException e) {
+                throw new IOException(e);
+            }
+        }
+
+        public String getUrl() throws IOException {
+            Long timestamp = System.currentTimeMillis();
+            String sign = getSign(timestamp);
+            return url + "&timestamp=" + timestamp + "&sign=" + sign;
+        }
+
+        public String getSign(Long timestamp) throws IOException {
+            try {
+                String stringToSign = timestamp + "\n" + secret;
+                Mac mac = Mac.getInstance("HmacSHA256");
+                mac.init(new SecretKeySpec(secret.getBytes(StandardCharsets.UTF_8), "HmacSHA256"));
+                byte[] signData = mac.doFinal(stringToSign.getBytes(StandardCharsets.UTF_8));
+                return URLEncoder.encode(Base64.getEncoder().encodeToString(signData), "UTF-8");
+            } catch (Exception e) {
+                throw new IOException(e);
+            }
+        }
     }
-  }
 
-}
+}