Merge branch 'feat_1.10_release_4.0.x_restfulapi' into 1.10_release

Author: yanghuaiGit

flinkx-restapi/flinkx-restapi-core/src/main/java/com/dtstack/flinkx/restapi/common/ConstantVarible.java

@@ -3,12 +3,10 @@
 public class ConstantVarible<T> implements Paramitem {
     private final T object;
     private final String name;
-//    private ParamDefinition paramDefinition;
 
     public ConstantVarible(T object, String name) {
         this.object = object;
         this.name = name;
-//        this.paramDefinition = paramDefinition;
     }
 
     @Override
@@ -21,8 +19,4 @@ public String getName() {
         return name;
     }
 
-//    @Override
-//    public ParamDefinition getParamDefinition() {
-//        return paramDefinition;
-//    }
 }

flinkx-restapi/flinkx-restapi-core/src/main/java/com/dtstack/flinkx/restapi/common/UuidVarible.java

@@ -8,10 +8,6 @@ public class UuidVarible implements Paramitem<String> {
     public UuidVarible() {
     }
 
-//    public UuidVarible(ParamDefinition paramDefinition) {
-//        this.paramDefinition = paramDefinition;
-//    }
-
     @Override
     public String getValue(RestContext restContext) {
         return UUID.randomUUID().toString();
@@ -22,10 +18,4 @@ public String getName() {
         return "currentTime";
     }
 
-//    @Override
-//    public ParamDefinition getParamDefinition() {
-//        return paramDefinition;
-//    }
-
-
 }

flinkx-restapi/flinkx-restapi-reader/src/main/java/com/dtstack/flinkx/restapi/inputformat/HttpClient.java

@@ -56,7 +56,6 @@ public class HttpClient {
     private final long intervalTime;
     private BlockingQueue<Row> queue;
     private RestContext restContext;
-    private AtomicInteger atomicInteger = new AtomicInteger(0);
     private static final String THREAD_NAME = "restApiReader-thread";
     private List<MetaColumn> metaColumns;
     private List<DataHandler> handlers;
@@ -96,7 +95,6 @@ public Row takeEvent() {
     }
 
     public void execute() {
-        int i = atomicInteger.incrementAndGet();
         httprequestApi.Httprequest build = restContext.build();
         doExecute(build, 2);
         restContext.updateValue();

flinkx-restapi/flinkx-restapi-reader/src/main/java/com/dtstack/flinkx/restapi/inputformat/RestapiInputFormat.java

@@ -38,24 +38,17 @@
  */
 public class RestapiInputFormat extends BaseRichInputFormat {
 
-    protected String url;
-
-    protected String method;
-
-
     protected HttpClient myHttpClient;
 
     protected RestContext restContext;
 
-    protected Map<String, Object> header;
-
     protected Long intervalTime;
 
-    private HttpRestConfig httpRestConfig;
+    protected HttpRestConfig httpRestConfig;
 
-    private List<MetaColumn> metaColumns ;
+    protected List<MetaColumn> metaColumns ;
 
-    private List<DataHandler> handlers;
+    protected List<DataHandler> handlers;
 
     @Override
     public void openInputFormat() throws IOException {

flinkx-restapi/flinkx-restapi-reader/src/main/java/com/dtstack/flinkx/restapi/inputformat/RestapiInputFormatBuilder.java

@@ -21,6 +21,7 @@
 import com.dtstack.flinkx.reader.MetaColumn;
 import com.dtstack.flinkx.restapi.common.handler.DataHandler;
 import com.dtstack.flinkx.restapi.reader.HttpRestConfig;
+import org.apache.commons.lang3.StringUtils;
 
 import java.util.List;
 import java.util.Map;
@@ -37,18 +38,6 @@ public RestapiInputFormatBuilder() {
         super.format = format = new RestapiInputFormat();
     }
 
-    public void setUrl(String url) {
-        this.format.url = url;
-    }
-
-    public void setHeader(Map<String, Object> header) {
-        this.format.header = header;
-    }
-
-    public void setMethod(String method) {
-        this.format.method = method;
-    }
-
     public void setHttpRestConfig(HttpRestConfig httpRestConfig) {
         this.format.setHttpRestConfig(httpRestConfig);
     }
@@ -67,11 +56,11 @@ public void setHandlers(List<DataHandler> handlers) {
 
     @Override
     protected void checkFormat() {
-//        if(format.url.isEmpty()){
-//            throw new IllegalArgumentException("缺少url");
-//        }
-//        if (format.method.isEmpty()) {
-//            throw new IllegalArgumentException("缺少method");
-//        }
+        if(StringUtils.isBlank(format.httpRestConfig.getUrl())){
+            throw new IllegalArgumentException("缺少url");
+        }
+        if (StringUtils.isBlank(format.httpRestConfig.getType())) {
+            throw new IllegalArgumentException("缺少method");
+        }
     }
 }