Support HTTP job (#1339)

elasticjob-executor/elasticjob-executor-type/elasticjob-http-executor/pom.xml

@@ -0,0 +1,48 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+  ~ 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.
+  -->
+
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>org.apache.shardingsphere.elasticjob</groupId>
+        <artifactId>elasticjob-executor-type</artifactId>
+        <version>3.0.0-beta-SNAPSHOT</version>
+    </parent>
+    <artifactId>elasticjob-http-executor</artifactId>
+    <name>${project.artifactId}</name>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.shardingsphere.elasticjob</groupId>
+            <artifactId>elasticjob-executor-kernel</artifactId>
+            <version>${project.parent.version}</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+        </dependency>
+    </dependencies>
+</project>

elasticjob-executor/elasticjob-executor-type/elasticjob-http-executor/src/main/java/org/apache/shardingsphere/elasticjob/http/executor/HttpJobExecutor.java

@@ -0,0 +1,124 @@
+/*
+ * 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.shardingsphere.elasticjob.http.executor;
+
+import com.google.common.base.Strings;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.shardingsphere.elasticjob.api.ElasticJob;
+import org.apache.shardingsphere.elasticjob.api.JobConfiguration;
+import org.apache.shardingsphere.elasticjob.api.ShardingContext;
+import org.apache.shardingsphere.elasticjob.executor.JobFacade;
+import org.apache.shardingsphere.elasticjob.executor.item.impl.TypedJobItemExecutor;
+import org.apache.shardingsphere.elasticjob.http.pojo.HttpParam;
+import org.apache.shardingsphere.elasticjob.http.props.HttpJobProperties;
+import org.apache.shardingsphere.elasticjob.infra.exception.JobConfigurationException;
+import org.apache.shardingsphere.elasticjob.infra.exception.JobExecutionException;
+import org.apache.shardingsphere.elasticjob.infra.json.GsonFactory;
+
+import java.io.BufferedReader;
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.net.HttpURLConnection;
+import java.net.URL;
+import java.nio.charset.StandardCharsets;
+import java.util.Properties;
+
+/**
+ * Http job executor.
+ */
+@Slf4j
+public final class HttpJobExecutor implements TypedJobItemExecutor {
+
+    @Override
+    public void process(final ElasticJob elasticJob, final JobConfiguration jobConfig, final JobFacade jobFacade, final ShardingContext shardingContext) {
+        HttpParam httpParam = getHttpParam(jobConfig.getProps());
+        HttpURLConnection connection = null;
+        BufferedReader bufferedReader = null;
+        try {
+            URL url = new URL(httpParam.getUrl());
+            connection = (HttpURLConnection) url.openConnection();
+            connection.setRequestMethod(httpParam.getMethod());
+            connection.setDoOutput(true);
+            connection.setConnectTimeout(httpParam.getConnectTimeout());
+            connection.setReadTimeout(httpParam.getReadTimeout());
+            if (!Strings.isNullOrEmpty(httpParam.getContentType())) {
+                connection.setRequestProperty("Content-Type", httpParam.getContentType());
+            }
+            connection.connect();
+            String data = httpParam.getData();
+            if (!Strings.isNullOrEmpty(data)) {
+                DataOutputStream dataOutputStream = new DataOutputStream(connection.getOutputStream());
+                if (httpParam.isEnableTransparentShardingContext()) {
+                    StringBuilder builder = new StringBuilder(data);
+                    builder.append("&").append(HttpJobProperties.TRANSPARENT_SHARDING_CONTEXT_KEY);
+                    builder.append("=").append(GsonFactory.getGson().toJson(shardingContext));
+                    data = builder.toString();
+                }
+                dataOutputStream.write(data.getBytes(StandardCharsets.UTF_8));
+                dataOutputStream.flush();
+                dataOutputStream.close();
+            }
+            int code = connection.getResponseCode();
+            if (code != 200) {
+                throw new JobExecutionException("Http job %s executed with response code %d", jobConfig.getJobName(), code);
+            }
+            bufferedReader = new BufferedReader(new InputStreamReader(connection.getInputStream(), StandardCharsets.UTF_8));
+            StringBuilder result = new StringBuilder();
+            String line;
+            while ((line = bufferedReader.readLine()) != null) {
+                result.append(line);
+            }
+            log.debug("http job execute result : {}", result.toString());
+        } catch (final IOException ex) {
+            throw new JobExecutionException(ex);
+        } finally {
+            try {
+                if (null != bufferedReader) {
+                    bufferedReader.close();
+                }
+                if (null != connection) {
+                    connection.disconnect();
+                }
+            } catch (final IOException ignore) {
+            }
+        }
+    }
+
+    private HttpParam getHttpParam(final Properties props) {
+        String url = props.getProperty(HttpJobProperties.URI_KEY);
+        if (Strings.isNullOrEmpty(url)) {
+            throw new JobConfigurationException("Cannot find http url, job is not executed.");
+        }
+        String method = props.getProperty(HttpJobProperties.METHOD_KEY);
+        if (Strings.isNullOrEmpty(method)) {
+            throw new JobConfigurationException("Cannot find http method, job is not executed.");
+        }
+        String data = props.getProperty(HttpJobProperties.DATA_KEY);
+        int connectTimeout = Integer.parseInt(props.getProperty(HttpJobProperties.CONNECT_TIMEOUT_KEY, "3000"));
+        int readTimeout = Integer.parseInt(props.getProperty(HttpJobProperties.READ_TIMEOUT_KEY, "5000"));
+        String contentType = props.getProperty(HttpJobProperties.CONTENT_TYPE_KEY);
+        boolean enableTransparentShardingContext = Boolean.parseBoolean(props.getProperty(HttpJobProperties.ENABLE_TRANSPARENT_SHARDING_CONTEXT_KEY, "false"));
+        return new HttpParam(url, method, data, connectTimeout, readTimeout, contentType, enableTransparentShardingContext);
+    }
+
+    @Override
+    public String getType() {
+        return "HTTP";
+    }
+}

elasticjob-executor/elasticjob-executor-type/elasticjob-http-executor/src/main/java/org/apache/shardingsphere/elasticjob/http/pojo/HttpParam.java

@@ -0,0 +1,43 @@
+/*
+ * 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.shardingsphere.elasticjob.http.pojo;
+
+import lombok.Getter;
+import lombok.RequiredArgsConstructor;
+
+/**
+ * Http job param.
+ */
+@RequiredArgsConstructor
+@Getter
+public final class HttpParam {
+
+    private final String url;
+
+    private final String method;
+
+    private final String data;
+
+    private final int connectTimeout;
+
+    private final int readTimeout;
+
+    private final String contentType;
+
+    private final boolean enableTransparentShardingContext;
+}

elasticjob-executor/elasticjob-executor-type/elasticjob-http-executor/src/main/java/org/apache/shardingsphere/elasticjob/http/props/HttpJobProperties.java

@@ -0,0 +1,64 @@
+/*
+ * 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.shardingsphere.elasticjob.http.props;
+
+/**
+ * Http job properties.
+ */
+public final class HttpJobProperties {
+
+    /**
+     * Http request uri.
+     */
+    public static final String URI_KEY = "http.uri";
+
+    /**
+     * Http request method.
+     */
+    public static final String METHOD_KEY = "http.method";
+
+    /**
+     * Http request data.
+     */
+    public static final String DATA_KEY = "http.data";
+
+    /**
+     * Http connect timeout.
+     */
+    public static final String CONNECT_TIMEOUT_KEY = "http.connect.timeout";
+
+    /**
+     * Http read timeout.
+     */
+    public static final String READ_TIMEOUT_KEY = "http.read.timeout";
+
+    /**
+     * Http content type.
+     */
+    public static final String CONTENT_TYPE_KEY = "http.content.type";
+
+    /**
+     * Http enable transparent sharding context.
+     */
+    public static final String ENABLE_TRANSPARENT_SHARDING_CONTEXT_KEY = "http.enable.transparent.sharding.context";
+
+    /**
+     * Http transparent sharding context.
+     */
+    public static final String TRANSPARENT_SHARDING_CONTEXT_KEY = "shardingContext";
+}

elasticjob-executor/elasticjob-executor-type/elasticjob-http-executor/src/main/resources/META-INF/services/org.apache.shardingsphere.elasticjob.executor.item.JobItemExecutor

@@ -0,0 +1,18 @@
+#
+# 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.
+#
+
+org.apache.shardingsphere.elasticjob.http.executor.HttpJobExecutor