diff --git a/connectors/rocketmq-connect-http/README.md b/connectors/rocketmq-connect-http/README.md index 8b6d5c704..7cb845dd4 100644 --- a/connectors/rocketmq-connect-http/README.md +++ b/connectors/rocketmq-connect-http/README.md @@ -4,26 +4,48 @@ Be responsible for consuming messages from producer and writing data to another web service system. ``` -## rocketmq-connect-http 打包 -``` -mvn clean install -Dmaven.test.skip=true -``` +## rocketmq-connect-http使用方法 + +1. 进入想要使用的connectors目录下（以rocketmq-connect-http目录为例），使用以下指令将插件进行打包 + ```shell + mvn clean package -Dmaven.test.skip=true + ``` +2. 打包好的插件以jar包的模式出现在`rocketmq-connect-http/target/`目录下 + +3. 在`distribution/conf`目录下找的对应的配置文件进行更新，对于standalone的启动方式，更新`connect-standalone.conf`文件中的`pluginPaths`变量 + + ``` + pluginPaths=(you plugin path) + ``` + + 相应的，使用distributed启动方式，则更新`connect-distributed.conf`中的变量 +4. 创建并启动对应的`SourceConnector`以及`SinkConnector` + ## rocketmq-connect-http 启动 * **http-sink-connector** 启动 ``` -http://${runtime-ip}:${runtime-port}/connectors/${rocketmq-http-sink-connector-name} -?config={"source-rocketmq":"${runtime-ip}:${runtime-port}","source-cluster":"${broker-cluster}","connector-class":"org.apache.rocketmq.connect.http.sink.HttpSinkConnector","connect-topicname" : "${connect-topicname}","url":"${url}"} +POST http://${runtime-ip}:${runtime-port}/connectors/${rocketmq-jdbc-source-connector-name} +{ + "connector.class":"org.apache.rocketmq.connect.http.HttpSinkTask", + "url":"${url}", + "method":"${method}", + "connect.topicnames":"${connect.topicnames}" +} ``` -例子 +例子 +``` +http://localhost:8081/connectors/httpSinkConnector?config={"connector-class":"org.apache.rocketmq.connect.http.HttpSinkTask","connect-topicname" : "http-topic","url":"192.168.1.2"} ``` -http://localhost:8081/connectors/httpConnectorSink?config={"source-rocketmq":"localhost:9876","source-cluster":"DefaultCluster", -"connector-class":"org.apache.rocketmq.connect.http.sink.HttpSinkConnector","connect-topicname" : "http-topic","url":"192.168.1.2"} +```在请求中定义http header、query、body参数 +http://127.0.0.1:8082/connectors/httpSinkConnector?config={"connector.class":"org.apache.rocketmq.connect.http.HttpSinkConnector","url":"http://localhost:8080/api","timeout":"6000","connect.topicnames":"fileTopic","headerParameters":"{\"header1k\":\"header1v\"}","method":"POST","queryParameters":"{\"queryk1\":\"queryv1\"}"} ``` +更多参数见[rocketmq-connect-http 参数说明](#rocketmq-connect-http-参数说明) + >**注：** `rocketmq-http-connect` 的启动依赖于`rocketmq-connect-runtime`项目的启动，需将打好的所有`jar`包放置到`runtime`项目中`pluginPaths`配置的路径后再执行上面的启动请求,该值配置在`runtime`项目下的`connect.conf`文件中 ## rocketmq-connect-http 停止 @@ -35,8 +57,31 @@ http://${runtime-ip}:${runtime-port}/connectors/${rocketmq-http-connector-name}/ ## rocketmq-connect-http 参数说明 * **http-sink-connector 参数说明** -| KEY | TYPE | Must be filled | Description | Example -|-----|---------|----------------|-------------|------------------| -| url | String | YES | sink端 域名地址 | http://127.0.0.1 | -|connect-topicname | String | YES | sink需要处理数据消息topic | xxxx | - +| KEY | TYPE | Must be filled | Description | Example | +|-----------------------|--------|----------------|------------------------------------------------|---------------------------| +| connect-topicname | String | YES | sink需要处理数据消息topic | fileTopic | +| url | String | YES | 目标端url地址 | http://localhost:8080/api | +| method | String | YES | http请求方法 | POST | +| body | String | No | http请求body字段，不填时默认使用事件的Data字段 | POST | +| headerParameters | String | NO | http请求header map动态参数Json字符串 | {"key1":"value1"} | +| fixedHeaderParameters | String | NO | http请求header map静态参数Json字符串 | {"key1":"value1"} | +| queryParameters | String | NO | http请求query map动态参数Json字符串 | {"key1":"value1"} | +| fixedQueryParameters | String | NO | http请求query map静态参数Json字符串 | {"key1":"value1"} | +| socks5UserName | String | NO | sock5代理用户名 | ***** | +| socks5Password | String | NO | sock5代理密码 | ***** | +| socks5Endpoint | String | NO | sock5代理地址 | http://localhost:7000 | +| timeout | String | NO | http请求超时时间(毫秒) | 3000 | +| concurrency | String | NO | http请求并发数 | 1 | +| authType | String | NO | 认证方式 (BASIC_AUTH/OAUTH_AUTH/API_KEY_AUTH/NONE) | BASIC_AUTH | +| basicUsername | String | NO | basic auth username | ***** | +| basicPassword | String | NO | basic auth password | ***** | +| apiKeyUsername | String | NO | api key auth username | ***** | +| apiKeyPassword | String | NO | api key auth password | ***** | +| oAuthEndpoint | String | NO | oauth 地址 | http://localhost:7000 | +| oAuthHttpMethod | String | NO | oauth http请求方法 | GET | +| oAuthClientId | String | NO | oauth client id | xxxx | +| oAuthClientSecret | String | NO | oauth client secret | xxxx | +| oAuthHeaderParameters | String | NO | oauth header map参数Json字符串 | {"key1":"value1"} | +| oAuthQueryParameters | String | NO | oauth query map参数Json字符串 | {"key1":"value1"} | +| oAuthBody | String | NO | oauth body参数 | bodyData | +| token | String | NO | http请求token，如果非空，会添加到http请求的header中，key为token | xxxx | diff --git a/connectors/rocketmq-connect-http/pom.xml b/connectors/rocketmq-connect-http/pom.xml index ba8a8b85d..6fcd1d366 100644 --- a/connectors/rocketmq-connect-http/pom.xml +++ b/connectors/rocketmq-connect-http/pom.xml @@ -141,6 +141,49 @@ + + org.apache.maven.plugins + maven-checkstyle-plugin + 3.1.2 + + + verify + verify + + ../../style/rmq_checkstyle.xml + UTF-8 + true + true + false + false + + + check + + + + + + maven-checkstyle-plugin + 3.1.2 + + + verify + verify + + ../../style/rmq_checkstyle.xml + UTF-8 + true + true + false + false + + + check + + + + @@ -197,6 +240,41 @@ commons-lang3 ${commons-lang3.version} + + org.apache.httpcomponents + httpclient + 4.5.13 + compile + + + com.google.code.gson + gson + 2.8.9 + compile + + + io.netty + netty-all + 4.1.25.Final + compile + + + com.google.guava + guava + 31.1-jre + compile + + + org.apache.httpcomponents + httpmime + 4.4.1 + + + org.projectlombok + lombok + 1.18.20 + compile + \ No newline at end of file diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/HttpConfig.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/HttpConfig.java new file mode 100644 index 000000000..39e14b911 --- /dev/null +++ b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/HttpConfig.java @@ -0,0 +1,31 @@ +/* + * 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.rocketmq.connect.http; + +import org.apache.rocketmq.connect.http.constant.HttpConstant; + +import java.util.HashSet; +import java.util.Set; + +public class HttpConfig { + public static final Set REQUEST_CONFIG = new HashSet() { + { + add(HttpConstant.URL); + add(HttpConstant.METHOD); + } + }; +} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/HttpSinkConnector.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/HttpSinkConnector.java new file mode 100644 index 000000000..8db9a99e9 --- /dev/null +++ b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/HttpSinkConnector.java @@ -0,0 +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.rocketmq.connect.http; + +import io.openmessaging.KeyValue; +import io.openmessaging.connector.api.component.task.Task; +import io.openmessaging.connector.api.component.task.sink.SinkConnector; +import io.openmessaging.connector.api.errors.ConnectException; + +import java.util.ArrayList; +import java.util.List; + +/** + * http sink connector + */ +public class HttpSinkConnector extends SinkConnector { + + @Override + public Class taskClass() { + return HttpSinkTask.class; + } + + private KeyValue config; + + @Override public void validate(KeyValue config) { + for (String requestKey : HttpConfig.REQUEST_CONFIG) { + if (!config.containsKey(requestKey)) { + throw new ConnectException("Request config key: " + requestKey); + } + } + } + + @Override public void start(KeyValue config) { + this.config = config; + } + + @Override public void stop() { + this.config = null; + } + + + @Override public List taskConfigs(int maxTasks) { + List config = new ArrayList<>(); + config.add(this.config); + return config; + } +} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/HttpSinkTask.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/HttpSinkTask.java new file mode 100644 index 000000000..d8a220c97 --- /dev/null +++ b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/HttpSinkTask.java @@ -0,0 +1,335 @@ +/* + * 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.rocketmq.connect.http; + +import com.alibaba.fastjson.JSONObject; +import com.google.gson.Gson; +import com.google.gson.reflect.TypeToken; +import io.openmessaging.KeyValue; +import io.openmessaging.connector.api.component.task.sink.SinkTask; +import io.openmessaging.connector.api.data.ConnectRecord; +import io.openmessaging.connector.api.errors.ConnectException; +import io.openmessaging.connector.api.errors.RetriableException; +import org.apache.commons.lang3.StringUtils; +import org.apache.rocketmq.connect.http.auth.Auth; +import org.apache.rocketmq.connect.http.auth.BasicAuthImpl; +import org.apache.rocketmq.connect.http.auth.ApiKeyAuthImpl; +import org.apache.rocketmq.connect.http.auth.OAuthClientImpl; +import org.apache.rocketmq.connect.http.constant.AuthTypeEnum; +import org.apache.rocketmq.connect.http.constant.HttpConstant; +import org.apache.rocketmq.connect.http.auth.ApacheHttpClientImpl; +import org.apache.rocketmq.connect.http.auth.AbstractHttpClient; +import org.apache.rocketmq.connect.http.auth.HttpCallback; +import org.apache.rocketmq.connect.http.entity.HttpBasicAuthParameters; +import org.apache.rocketmq.connect.http.entity.HttpApiKeyAuthParameters; +import org.apache.rocketmq.connect.http.entity.HttpAuthParameters; +import org.apache.rocketmq.connect.http.entity.HttpOAuthParameters; +import org.apache.rocketmq.connect.http.entity.HttpRequest; +import org.apache.rocketmq.connect.http.entity.ProxyConfig; +import org.apache.rocketmq.connect.http.util.CheckUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import java.io.UnsupportedEncodingException; +import java.net.URLEncoder; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.TimeUnit; + +public class HttpSinkTask extends SinkTask { + private static final Logger log = LoggerFactory.getLogger(HttpSinkTask.class); + protected static final int DEFAULT_CONSUMER_TIMEOUT_SECONDS = 30; + protected static final String DEFAULT_REQUEST_TIMEOUT_MILL_SECONDS = "3000"; + protected static final int DEFAULT_OAUTH_DELAY_SECONDS = 1; + protected static final int DEFAULT_CONCURRENCY = 1; + + private String url; + private String method; + private String headerParameters; + private String fixedHeaderParameters; + private String queryParameters; + private String fixedQueryParameters; + private String body; + private String socks5UserName; + private String socks5Password; + private String socks5Endpoint; + private String timeout; + private String concurrency; + private String authType; + private String basicUsername; + private String basicPassword; + private String apiKeyUsername; + private String apiKeyPassword; + private String oAuthEndpoint; + private String oAuthHttpMethod; + private String oAuthClientId; + private String oAuthClientSecret; + private String oAuthHeaderParameters; + private String oAuthQueryParameters; + private String oAuthBody; + private String token; + + private Long awaitTimeoutSeconds; + private Auth auth; + private HttpAuthParameters httpAuthParameters; + private ScheduledExecutorService scheduledExecutorService; + private AbstractHttpClient httpClient; + + @Override + public void put(List records) throws ConnectException { + try { + Long startTime = System.currentTimeMillis(); + CountDownLatch countDownLatch = new CountDownLatch(records.size()); + HttpCallback httpCallback = new HttpCallback(countDownLatch); + for (ConnectRecord connectRecord : records) { + // body + // header + Map headerMap = renderHeaderMap(headerParameters, fixedHeaderParameters, token); + if (auth != null) { + headerMap.putAll(auth.auth()); + } + // render query to url + String urlWithQueryParameters = renderQueryParametersToUrl(url, queryParameters, fixedQueryParameters); + HttpRequest httpRequest = new HttpRequest(); + httpRequest.setUrl(urlWithQueryParameters); + httpRequest.setMethod(method); + httpRequest.setHeaderMap(headerMap); + if (body != null) { + httpRequest.setBody(body); + } else { + httpRequest.setBody(new Gson().toJson(connectRecord.getData())); + } + httpRequest.setTimeout(StringUtils.isNotBlank(timeout) ? timeout : DEFAULT_REQUEST_TIMEOUT_MILL_SECONDS); + log.info("HttpSinkTask send request | url:{} | method: {} | headerMap: {} | body: {}", + httpRequest.getUrl(), httpRequest.getMethod(), httpRequest.getHeaderMap(), httpRequest.getBody()); + httpClient.executeNotReturn(httpRequest, httpCallback); + } + boolean consumeSucceed = Boolean.FALSE; + try { + consumeSucceed = countDownLatch.await(awaitTimeoutSeconds, TimeUnit.SECONDS); + } catch (Throwable e) { + log.error("count down latch failed.", e); + } + if (!consumeSucceed) { + throw new RetriableException("Request Timeout"); + } + if (httpCallback.isFailed()) { + throw new RetriableException(httpCallback.getMsg()); + } + log.info("HttpSinkTask put size:{},cost:{}", records.size(), System.currentTimeMillis() - startTime); + } catch (Exception e) { + log.error("HttpSinkTask | put | error => ", e); + throw new RuntimeException(e); + } + } + + private Map renderHeaderMap(String headerParameters, String fixedHeaderParameters, String token) { + Map headerMap = new HashMap<>(); + if (headerParameters != null) { + Map userDefinedHeaders = new Gson().fromJson(headerParameters, new TypeToken>() { + }.getType()); + headerMap.putAll(userDefinedHeaders); + } + if (fixedHeaderParameters != null) { + Map fixedHeaders = new Gson().fromJson(fixedHeaderParameters, new TypeToken>() { + }.getType()); + headerMap.putAll(fixedHeaders); + } + if (StringUtils.isNotBlank(token)) { + headerMap.put(HttpConstant.TOKEN, token); + } + return headerMap; + } + + private String renderQueryParametersToUrl(String url, String queryParameters, String fixedQueryParameters) throws UnsupportedEncodingException { + Map queryParametersMap = new HashMap<>(); + if (queryParameters != null) { + queryParametersMap.putAll(new Gson().fromJson(queryParameters, Map.class)); + } + if (fixedQueryParameters != null) { + queryParametersMap.putAll(new Gson().fromJson(fixedQueryParameters, Map.class)); + } + StringBuilder queryParametersStringBuilder = new StringBuilder(); + for (Map.Entry entry : queryParametersMap.entrySet()) { + if (entry.getValue() instanceof JSONObject) { + queryParametersStringBuilder.append(URLEncoder.encode(entry.getKey(), "UTF-8")).append("=").append(URLEncoder.encode(((JSONObject) entry.getValue()).toJSONString(), "UTF-8")).append("&"); + } else { + queryParametersStringBuilder.append(URLEncoder.encode(entry.getKey(), "UTF-8")).append("=").append(URLEncoder.encode((String) entry.getValue(), "UTF-8")).append("&"); + } + } + String path = queryParametersStringBuilder.toString(); + if (StringUtils.isNotBlank(path) && StringUtils.isNotBlank(url)) { + if (url.contains("?")) { + return url + "&" + path.substring(0, path.length() - 1); + } + return url + "?" + path.substring(0, path.length() - 1); + } + return url; + } + + @Override + public void validate(KeyValue config) { + if (CheckUtils.checkNull(config.getString(HttpConstant.URL)) || + CheckUtils.checkNull(config.getString(HttpConstant.METHOD))) { + throw new RuntimeException("http required parameter is null !"); + } + } + + @Override + public void start(KeyValue config) { + // http + url = CheckUtils.checkNullReturnDefault(config.getString(HttpConstant.URL)); + method = CheckUtils.checkNullReturnDefault(config.getString(HttpConstant.METHOD)); + headerParameters = CheckUtils.checkNullReturnDefault(config.getString(HttpConstant.HEADER_PARAMETERS)); + fixedHeaderParameters = CheckUtils.checkNullReturnDefault(config.getString(HttpConstant.FIXED_HEADER_PARAMETERS)); + queryParameters = CheckUtils.checkNullReturnDefault(config.getString(HttpConstant.QUERY_PARAMETERS)); + fixedQueryParameters = CheckUtils.checkNullReturnDefault(config.getString(HttpConstant.FIXED_QUERY_PARAMETERS)); + body = CheckUtils.checkNullReturnDefault(config.getString(HttpConstant.BODY)); + socks5Endpoint = CheckUtils.checkNullReturnDefault(config.getString(HttpConstant.SOCKS5_ENDPOINT)); + socks5UserName = CheckUtils.checkNullReturnDefault(config.getString(HttpConstant.SOCKS5_USERNAME)); + socks5Password = CheckUtils.checkNullReturnDefault(config.getString(HttpConstant.SOCKS5_PASSWORD)); + timeout = CheckUtils.checkNullReturnDefault(config.getString(HttpConstant.TIMEOUT)); + concurrency = CheckUtils.checkNullReturnDefault(config.getString(HttpConstant.CONCURRENCY)); + authType = CheckUtils.checkNullReturnDefault(config.getString(HttpConstant.AUTH_TYPE)); + basicUsername = CheckUtils.checkNullReturnDefault(config.getString(HttpConstant.BASIC_USERNAME)); + basicPassword = CheckUtils.checkNullReturnDefault(config.getString(HttpConstant.BASIC_PASSWORD)); + apiKeyUsername = CheckUtils.checkNullReturnDefault(config.getString(HttpConstant.API_KEY_USERNAME)); + apiKeyPassword = CheckUtils.checkNullReturnDefault(config.getString(HttpConstant.API_KEY_PASSWORD)); + oAuthEndpoint = CheckUtils.checkNullReturnDefault(config.getString(HttpConstant.OAUTH_ENDPOINT)); + oAuthHttpMethod = CheckUtils.checkNullReturnDefault(config.getString(HttpConstant.OAUTH_HTTP_METHOD)); + oAuthClientId = CheckUtils.checkNullReturnDefault(config.getString(HttpConstant.OAUTH_CLIENT_ID)); + oAuthClientSecret = CheckUtils.checkNullReturnDefault(config.getString(HttpConstant.OAUTH_CLIENT_SECRET)); + oAuthHeaderParameters = CheckUtils.checkNullReturnDefault(config.getString(HttpConstant.OAUTH_HEADER_PARAMETERS)); + oAuthQueryParameters = CheckUtils.checkNullReturnDefault(config.getString(HttpConstant.OAUTH_QUERY_PARAMETERS)); + oAuthBody = CheckUtils.checkNullReturnDefault(config.getString(HttpConstant.OAUTH_BODY)); + token = config.getString(HttpConstant.TOKEN); + try { + httpClient = new ApacheHttpClientImpl(StringUtils.isNotBlank(concurrency) ? Integer.parseInt(concurrency) : DEFAULT_CONCURRENCY); + ProxyConfig proxyConfig = new ProxyConfig(); + proxyConfig.setSocks5Endpoint(socks5Endpoint); + proxyConfig.setSocks5UserName(socks5UserName); + proxyConfig.setSocks5Password(socks5Password); + httpClient.init(proxyConfig); + initAuth(httpClient); + awaitTimeoutSeconds = Long.min(StringUtils.isNotBlank(timeout) ? Long.parseLong(timeout) + 5 : DEFAULT_CONSUMER_TIMEOUT_SECONDS, DEFAULT_CONSUMER_TIMEOUT_SECONDS); + } catch (Exception e) { + log.error("HttpSinkTask | start | error => ", e); + } + log.info("HttpSinkTask | start | success => config : {}", new Gson().toJson(config)); + } + + @Override + public void stop() { + httpClient.close(); + if (scheduledExecutorService != null) { + this.scheduledExecutorService.shutdownNow(); + this.scheduledExecutorService = null; + } + } + + private void initAuth(AbstractHttpClient httpClient) throws Exception { + httpAuthParameters = initAuthParameters(); + if (scheduledExecutorService != null) { + scheduledExecutorService.shutdownNow(); + scheduledExecutorService = null; + } + if (httpAuthParameters == null || StringUtils.isBlank(httpAuthParameters.getAuthType())) { + this.auth = null; + return; + } + if (httpAuthParameters.getAuthType().equals(AuthTypeEnum.BASIC.getAuthType())) { + this.auth = new BasicAuthImpl(httpAuthParameters); + } else if (httpAuthParameters.getAuthType().equals(AuthTypeEnum.API_KEY.getAuthType())) { + this.auth = new ApiKeyAuthImpl(httpAuthParameters); + } else if (httpAuthParameters.getAuthType().equals(AuthTypeEnum.OAUTH_CLIENT_CREDENTIALS.getAuthType())) { + if (httpClient == null) { + log.error("HttpSinkTask | initOrUpdateAuth | httpClient is null !"); + throw new RuntimeException("httpClient is null ! init httpClient before initOrUpdateAuth"); + } + OAuthClientImpl oAuthClient = new OAuthClientImpl(httpAuthParameters); + oAuthClient.setHttpClient(httpClient); + scheduledExecutorService = Executors.newSingleThreadScheduledExecutor(); + scheduledExecutorService.scheduleWithFixedDelay(oAuthClient, DEFAULT_OAUTH_DELAY_SECONDS, DEFAULT_OAUTH_DELAY_SECONDS, TimeUnit.SECONDS); + this.auth = oAuthClient; + } else { + this.auth = null; + } + } + + private HttpAuthParameters initAuthParameters() { + HttpBasicAuthParameters httpBasicAuthParameters = new HttpBasicAuthParameters(basicUsername, basicPassword); + HttpApiKeyAuthParameters httpApiKeyAuthParameters = new HttpApiKeyAuthParameters(apiKeyUsername, apiKeyPassword); + HttpOAuthParameters httpOAuthParameters = new HttpOAuthParameters(oAuthEndpoint, oAuthHttpMethod, oAuthClientId, oAuthClientSecret, oAuthHeaderParameters, oAuthQueryParameters, oAuthBody, DEFAULT_REQUEST_TIMEOUT_MILL_SECONDS); + return new HttpAuthParameters(authType, httpBasicAuthParameters, httpApiKeyAuthParameters, httpOAuthParameters); + } + + /* + * update proxy + * */ + protected void updateProxy(ProxyConfig proxyConfig) { + this.httpClient.updateProxyConfig(proxyConfig); + } + + protected void updateAuth(String authType, + String basicUsername, String basicPassword, + String apiKeyUsername, String apiKeyPassword, + String oAuthEndpoint, String oAuthHttpMethod, String oAuthClientId, String oAuthClientSecret, String oAuthHeaderParameters, String oAuthQueryParameters, String oAuthBody) { + if (AuthTypeEnum.parse(authType) == null) { + throw new RuntimeException("authType is null !"); + } + this.authType = authType; + this.basicUsername = basicUsername; + this.basicPassword = basicPassword; + this.apiKeyUsername = apiKeyUsername; + this.apiKeyPassword = apiKeyPassword; + this.oAuthEndpoint = oAuthEndpoint; + this.oAuthHttpMethod = oAuthHttpMethod; + this.oAuthClientId = oAuthClientId; + this.oAuthClientSecret = oAuthClientSecret; + this.oAuthHeaderParameters = oAuthHeaderParameters; + this.oAuthQueryParameters = oAuthQueryParameters; + this.oAuthBody = oAuthBody; + try { + initAuth(httpClient); + } catch (Exception e) { + log.error("HttpSinkTask | updateAuth | error => ", e); + } + } + + protected Auth getAuth() { + return auth; + } + + protected AbstractHttpClient getHttpClient() { + return httpClient; + } + + protected Map getCustomHeaders() throws Exception { + return renderHeaderMap(headerParameters, fixedHeaderParameters, token); + } + + protected String getCustomUrlWithQueryParameters() throws Exception { + return renderQueryParametersToUrl(url, queryParameters, fixedQueryParameters); + } + + protected String getCustomBody(ConnectRecord connectRecord) throws Exception { + return body; + } +} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/auth/AbstractHttpClient.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/auth/AbstractHttpClient.java new file mode 100644 index 000000000..319b51ddd --- /dev/null +++ b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/auth/AbstractHttpClient.java @@ -0,0 +1,48 @@ +/* + * 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.rocketmq.connect.http.auth; + +import org.apache.rocketmq.connect.http.entity.ProxyConfig; +import org.apache.rocketmq.connect.http.entity.HttpRequest; +import java.io.IOException; + +public interface AbstractHttpClient { + + /** + * + * @param config + */ + void init(ProxyConfig config); + + /** + * + * @return + * @throws IOException + */ + void executeNotReturn(HttpRequest httpRequest, HttpCallback httpCallback) throws Exception; + + /** + * + * @return + * @throws IOException + */ + String execute(HttpRequest httpRequest, HttpCallback httpCallback) throws Exception; + + void close(); + + void updateProxyConfig(ProxyConfig config); +} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/auth/ApacheHttpClientImpl.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/auth/ApacheHttpClientImpl.java new file mode 100644 index 000000000..6193d63bf --- /dev/null +++ b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/auth/ApacheHttpClientImpl.java @@ -0,0 +1,601 @@ +/* + * 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.rocketmq.connect.http.auth; + +import com.google.common.net.MediaType; +import com.google.gson.Gson; +import com.google.gson.reflect.TypeToken; +import io.netty.util.concurrent.DefaultThreadFactory; +import org.apache.commons.lang3.StringUtils; +import org.apache.http.HttpEntity; +import org.apache.http.HttpHost; +import org.apache.http.NameValuePair; +import org.apache.http.client.config.RequestConfig; +import org.apache.http.client.entity.UrlEncodedFormEntity; +import org.apache.http.client.methods.CloseableHttpResponse; +import org.apache.http.client.methods.HttpOptions; +import org.apache.http.client.methods.HttpPatch; +import org.apache.http.client.methods.HttpPost; +import org.apache.http.client.methods.HttpPut; +import org.apache.http.client.methods.HttpRequestBase; +import org.apache.http.client.protocol.HttpClientContext; +import org.apache.http.config.Registry; +import org.apache.http.config.RegistryBuilder; +import org.apache.http.conn.DnsResolver; +import org.apache.http.conn.socket.ConnectionSocketFactory; +import org.apache.http.conn.socket.PlainConnectionSocketFactory; +import org.apache.http.conn.ssl.SSLConnectionSocketFactory; +import org.apache.http.conn.ssl.TrustStrategy; +import org.apache.http.entity.ByteArrayEntity; +import org.apache.http.entity.ContentType; +import org.apache.http.entity.StringEntity; +import org.apache.http.entity.mime.MultipartEntityBuilder; +import org.apache.http.entity.mime.content.StringBody; +import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.impl.client.HttpClients; +import org.apache.http.impl.conn.PoolingHttpClientConnectionManager; +import org.apache.http.protocol.HTTP; +import org.apache.http.protocol.HttpContext; +import org.apache.http.ssl.SSLContextBuilder; +import org.apache.rocketmq.connect.http.constant.HttpConstant; +import org.apache.rocketmq.connect.http.entity.ProxyConfig; +import org.apache.rocketmq.connect.http.entity.HttpDeleteWithEntity; +import org.apache.rocketmq.connect.http.entity.HttpGetWithEntity; +import org.apache.rocketmq.connect.http.entity.HttpHeadWithEntity; +import org.apache.rocketmq.connect.http.entity.HttpRequest; +import org.apache.rocketmq.connect.http.entity.HttpTraceWithEntity; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import javax.net.ssl.SSLContext; +import javax.net.ssl.SSLSession; +import java.io.IOException; +import java.io.UnsupportedEncodingException; +import java.net.InetAddress; +import java.net.InetSocketAddress; +import java.net.Proxy; +import java.net.Socket; +import java.net.UnknownHostException; +import java.nio.charset.StandardCharsets; +import java.security.cert.X509Certificate; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Future; +import java.util.concurrent.LinkedBlockingDeque; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.TimeUnit; + +public class ApacheHttpClientImpl implements AbstractHttpClient { + private static final Logger log = LoggerFactory.getLogger(ApacheHttpClientImpl.class); + + private static ExecutorService executorServicePool; + private CloseableHttpClient httpClient = null; + private SocksProxyConfig socksProxyConfig; + private static final String SOCKS_ADDRESS_KEY = "socks.address"; + + public ApacheHttpClientImpl() { + executorServicePool = new ThreadPoolExecutor(200, 2000, 600, TimeUnit.SECONDS, + new LinkedBlockingDeque(1000), new DefaultThreadFactory("ApacheHttpClientRequestThread")); + } + + public ApacheHttpClientImpl(Integer concurrency) { + executorServicePool = new ThreadPoolExecutor(concurrency, concurrency, 600, TimeUnit.SECONDS, + new LinkedBlockingDeque(1000), new DefaultThreadFactory("ApacheHttpClientRequestThread")); + } + + @Override + public void init(ProxyConfig config) { + try { + SSLContextBuilder sslContextBuilder = new SSLContextBuilder().loadTrustMaterial(null, new TrustStrategy() { + @Override + public boolean isTrusted(X509Certificate[] chain, String authType) { + return true; + } + }); + Registry reg = RegistryBuilder.create().register("http", + new SocksPlainConnectionSocketFactory()) + .register("https", new SocksSSLConnectionSocketFactory(sslContextBuilder.build())) + .build(); + PoolingHttpClientConnectionManager connManager = new PoolingHttpClientConnectionManager(reg, + new FakeDnsResolver()); + connManager.setMaxTotal(400); + connManager.setDefaultMaxPerRoute(500); + httpClient = HttpClients.custom() + .setConnectionManager(connManager) + .build(); + + this.socksProxyConfig = new SocksProxyConfig(config.getSocks5Endpoint(), config.getSocks5UserName(), config.getSocks5Password()); + } catch (Exception e) { + log.error("ApacheHttpClientImpl | init | error => ", e); + throw new RuntimeException(e); + } + } + + @Override + public String execute(HttpRequest httpRequest, HttpCallback httpCallback) throws Exception { + HttpRequestCallable httpRequestCallable = getHttpRequestCallable(httpRequest, httpCallback); + Future submit = executorServicePool.submit(httpRequestCallable); + String result = submit.get(); + log.info("ApacheHttpClientImpl | execute | success | result : {}", result); + return result; + } + + @Override + public void executeNotReturn(HttpRequest httpRequest, HttpCallback httpCallback) throws Exception { + HttpRequestCallable httpRequestCallable = getHttpRequestCallable(httpRequest, httpCallback); + executorServicePool.submit(httpRequestCallable); + } + + private HttpRequestCallable getHttpRequestCallable(HttpRequest httpRequest, HttpCallback httpCallback) throws UnsupportedEncodingException { + CloseableHttpResponse response; + HttpRequestBase httpRequestBase = null; + if (httpRequest != null) { + if (!StringUtils.isEmpty(httpRequest.getBody())) { + httpRequestBase = extracted(httpRequest.getUrl(), httpRequest.getMethod(), httpRequest.getHeaderMap(), httpRequest.getBody()); + } else if (httpRequest.getBytesBody() != null && httpRequest.getBytesBody().length != 0) { + httpRequestBase = extracted(httpRequest.getUrl(), httpRequest.getMethod(), httpRequest.getHeaderMap(), httpRequest.getBytesBody()); + } + if (StringUtils.isNotBlank(httpRequest.getTimeout())) { + final RequestConfig requestConfig = RequestConfig.custom(). + setConnectionRequestTimeout(Integer.parseInt(httpRequest.getTimeout())). + setSocketTimeout(Integer.parseInt(httpRequest.getTimeout())). + setConnectTimeout(Integer.parseInt(httpRequest.getTimeout())).build(); + httpRequestBase.setConfig(requestConfig); + } + } + + HttpRequestCallable httpRequestCallable = new HttpRequestCallable(httpClient, httpRequestBase, + HttpClientContext.create(), this.socksProxyConfig, httpCallback); + return httpRequestCallable; + } + + private HttpRequestBase extracted(String url, String method, Map headerMap, String body) throws UnsupportedEncodingException { + switch (method) { + case "GET": + return getHttpGetWithEntity(url, headerMap, body); + case "POST": + return getHttpPost(url, headerMap, body); + case "DELETE": + return getHttpDeleteWithEntity(url, headerMap, body); + case "PUT": + return getHttpPut(url, headerMap, body); + case "HEAD": + return getHttpHeadWithEntity(url, headerMap, body); + case "TRACE": + return getHttpTraceWithEntity(url, headerMap, body); + case "PATCH": + return getHttpPatch(url, headerMap, body); + default: + } + HttpOptions httpOptions = new HttpOptions(url); + headerMap.forEach(httpOptions::addHeader); + httpOptions.addHeader(HTTP.CONTENT_TYPE, MediaType.JSON_UTF_8.toString()); + return httpOptions; + } + + private HttpRequestBase extracted(String url, String method, Map headerMap, byte[] bytesBody) throws UnsupportedEncodingException { + switch (method) { + case "POST": + return getHttpPostByte(url, headerMap, bytesBody); + default: + throw new UnsupportedEncodingException(); + } + } + + private HttpTraceWithEntity getHttpTraceWithEntity(String url, Map headerMap, String body) { + HttpTraceWithEntity httpTraceWithEntity = new HttpTraceWithEntity(url); + headerMap.forEach(httpTraceWithEntity::addHeader); + httpTraceWithEntity.addHeader(HTTP.CONN_DIRECTIVE, HTTP.CONN_KEEP_ALIVE); + String contentType = headerMap.get(HTTP.CONTENT_TYPE); + if (StringUtils.isNotBlank(body) + && StringUtils.isNotBlank(contentType) + && (contentType.contains(ContentType.APPLICATION_JSON.getMimeType()) + || contentType.contains(HttpConstant.CONTENT_TYPE_APPLICATION_JAVASCRIPT) + || contentType.contains(ContentType.APPLICATION_XML.getMimeType()) + || contentType.contains(ContentType.TEXT_PLAIN.getMimeType()) + || contentType.contains(ContentType.TEXT_XML.getMimeType()) + || contentType.contains(ContentType.TEXT_HTML.getMimeType()))) { + HttpEntity entityPot = new StringEntity(body, MediaType.JSON_UTF_8.charset() + .get()); + httpTraceWithEntity.setEntity(entityPot); + } + if (StringUtils.isNotBlank(body) + && StringUtils.isNotBlank(contentType) + && contentType.contains(ContentType.APPLICATION_FORM_URLENCODED.getMimeType())) { + UrlEncodedFormEntity urlEncodedFormEntity = getUrlEncodedFormEntity(body); + httpTraceWithEntity.setEntity(urlEncodedFormEntity); + } + if (StringUtils.isNotBlank(body) + && StringUtils.isNotBlank(contentType) + && contentType.contains(ContentType.MULTIPART_FORM_DATA.getMimeType())) { + MultipartEntityBuilder multipartEntityBuilder = getMultipartEntityBuilder(body); + httpTraceWithEntity.setEntity(multipartEntityBuilder.build()); + } + if (StringUtils.isNotBlank(body) + && StringUtils.isBlank(contentType)) { + HttpEntity entityPot = new StringEntity(body, MediaType.JSON_UTF_8.charset() + .get()); + httpTraceWithEntity.setEntity(entityPot); + } + return httpTraceWithEntity; + } + + private HttpHeadWithEntity getHttpHeadWithEntity(String url, Map headerMap, String body) { + HttpHeadWithEntity httpHeadWithEntity = new HttpHeadWithEntity(url); + headerMap.forEach(httpHeadWithEntity::addHeader); + httpHeadWithEntity.addHeader(HTTP.CONN_DIRECTIVE, HTTP.CONN_KEEP_ALIVE); + String contentType = headerMap.get(HTTP.CONTENT_TYPE); + if (StringUtils.isNotBlank(body) + && StringUtils.isNotBlank(contentType) + && (contentType.contains(ContentType.APPLICATION_JSON.getMimeType()) + || contentType.contains(HttpConstant.CONTENT_TYPE_APPLICATION_JAVASCRIPT) + || contentType.contains(ContentType.APPLICATION_XML.getMimeType()) + || contentType.contains(ContentType.TEXT_PLAIN.getMimeType()) + || contentType.contains(ContentType.TEXT_XML.getMimeType()) + || contentType.contains(ContentType.TEXT_HTML.getMimeType()))) { + HttpEntity entityPot = new StringEntity(body, MediaType.JSON_UTF_8.charset() + .get()); + httpHeadWithEntity.setEntity(entityPot); + } + if (StringUtils.isNotBlank(body) + && StringUtils.isNotBlank(contentType) + && contentType.contains(ContentType.APPLICATION_FORM_URLENCODED.getMimeType())) { + UrlEncodedFormEntity urlEncodedFormEntity = getUrlEncodedFormEntity(body); + httpHeadWithEntity.setEntity(urlEncodedFormEntity); + } + if (StringUtils.isNotBlank(body) + && StringUtils.isNotBlank(contentType) + && contentType.contains(ContentType.MULTIPART_FORM_DATA.getMimeType())) { + MultipartEntityBuilder multipartEntityBuilder = getMultipartEntityBuilder(body); + httpHeadWithEntity.setEntity(multipartEntityBuilder.build()); + } + if (StringUtils.isNotBlank(body) + && StringUtils.isBlank(contentType)) { + HttpEntity entityPot = new StringEntity(body, MediaType.JSON_UTF_8.charset() + .get()); + httpHeadWithEntity.setEntity(entityPot); + } + return httpHeadWithEntity; + } + + private HttpDeleteWithEntity getHttpDeleteWithEntity(String url, Map headerMap, String body) { + HttpDeleteWithEntity httpDeleteWithEntity = new HttpDeleteWithEntity(url); + headerMap.forEach(httpDeleteWithEntity::addHeader); + httpDeleteWithEntity.addHeader(HTTP.CONN_DIRECTIVE, HTTP.CONN_KEEP_ALIVE); + String contentType = headerMap.get(HTTP.CONTENT_TYPE); + if (StringUtils.isNotBlank(body) + && StringUtils.isNotBlank(contentType) + && (contentType.contains(ContentType.APPLICATION_JSON.getMimeType()) + || contentType.contains(HttpConstant.CONTENT_TYPE_APPLICATION_JAVASCRIPT) + || contentType.contains(ContentType.APPLICATION_XML.getMimeType()) + || contentType.contains(ContentType.TEXT_PLAIN.getMimeType()) + || contentType.contains(ContentType.TEXT_XML.getMimeType()) + || contentType.contains(ContentType.TEXT_HTML.getMimeType()))) { + HttpEntity entityPot = new StringEntity(body, MediaType.JSON_UTF_8.charset() + .get()); + httpDeleteWithEntity.setEntity(entityPot); + } + if (StringUtils.isNotBlank(body) + && StringUtils.isNotBlank(contentType) + && contentType.contains(ContentType.APPLICATION_FORM_URLENCODED.getMimeType())) { + UrlEncodedFormEntity urlEncodedFormEntity = getUrlEncodedFormEntity(body); + httpDeleteWithEntity.setEntity(urlEncodedFormEntity); + } + if (StringUtils.isNotBlank(body) + && StringUtils.isNotBlank(contentType) + && contentType.contains(ContentType.MULTIPART_FORM_DATA.getMimeType())) { + MultipartEntityBuilder multipartEntityBuilder = getMultipartEntityBuilder(body); + httpDeleteWithEntity.setEntity(multipartEntityBuilder.build()); + } + if (StringUtils.isNotBlank(body) + && StringUtils.isBlank(contentType)) { + HttpEntity entityPot = new StringEntity(body, MediaType.JSON_UTF_8.charset() + .get()); + httpDeleteWithEntity.setEntity(entityPot); + } + return httpDeleteWithEntity; + } + + private HttpGetWithEntity getHttpGetWithEntity(String url, Map headerMap, String body) { + HttpGetWithEntity httpGetWithEntity = new HttpGetWithEntity(url); + headerMap.forEach(httpGetWithEntity::addHeader); + httpGetWithEntity.addHeader(HTTP.CONN_DIRECTIVE, HTTP.CONN_KEEP_ALIVE); + String contentType = headerMap.get(HTTP.CONTENT_TYPE); + if (StringUtils.isNotBlank(body) + && StringUtils.isNotBlank(contentType) + && (contentType.contains(ContentType.APPLICATION_JSON.getMimeType()) + || contentType.contains(HttpConstant.CONTENT_TYPE_APPLICATION_JAVASCRIPT) + || contentType.contains(ContentType.APPLICATION_XML.getMimeType()) + || contentType.contains(ContentType.TEXT_PLAIN.getMimeType()) + || contentType.contains(ContentType.TEXT_XML.getMimeType()) + || contentType.contains(ContentType.TEXT_HTML.getMimeType()))) { + HttpEntity entityPot = new StringEntity(body, MediaType.JSON_UTF_8.charset() + .get()); + httpGetWithEntity.setEntity(entityPot); + } + if (StringUtils.isNotBlank(body) + && StringUtils.isNotBlank(contentType) + && contentType.contains(ContentType.APPLICATION_FORM_URLENCODED.getMimeType())) { + UrlEncodedFormEntity urlEncodedFormEntity = getUrlEncodedFormEntity(body); + httpGetWithEntity.setEntity(urlEncodedFormEntity); + } + if (StringUtils.isNotBlank(body) + && StringUtils.isNotBlank(contentType) + && contentType.contains(ContentType.MULTIPART_FORM_DATA.getMimeType())) { + MultipartEntityBuilder multipartEntityBuilder = getMultipartEntityBuilder(body); + httpGetWithEntity.setEntity(multipartEntityBuilder.build()); + } + if (StringUtils.isNotBlank(body) + && StringUtils.isBlank(contentType)) { + HttpEntity entityPot = new StringEntity(body, MediaType.JSON_UTF_8.charset() + .get()); + httpGetWithEntity.setEntity(entityPot); + } + return httpGetWithEntity; + } + + private HttpPut getHttpPut(String url, Map headerMap, String body) { + HttpPut httpPut = new HttpPut(url); + headerMap.forEach(httpPut::addHeader); + httpPut.addHeader(HTTP.CONN_DIRECTIVE, HTTP.CONN_KEEP_ALIVE); + String contentType = headerMap.get(HTTP.CONTENT_TYPE); + if (StringUtils.isNotBlank(body) + && StringUtils.isNotBlank(contentType) + && (contentType.contains(ContentType.APPLICATION_JSON.getMimeType()) + || contentType.contains(HttpConstant.CONTENT_TYPE_APPLICATION_JAVASCRIPT) + || contentType.contains(ContentType.APPLICATION_XML.getMimeType()) + || contentType.contains(ContentType.TEXT_PLAIN.getMimeType()) + || contentType.contains(ContentType.TEXT_XML.getMimeType()) + || contentType.contains(ContentType.TEXT_HTML.getMimeType()))) { + HttpEntity entityPot = new StringEntity(body, MediaType.JSON_UTF_8.charset() + .get()); + httpPut.setEntity(entityPot); + } + if (StringUtils.isNotBlank(body) + && StringUtils.isNotBlank(contentType) + && contentType.contains(ContentType.APPLICATION_FORM_URLENCODED.getMimeType())) { + UrlEncodedFormEntity urlEncodedFormEntity = getUrlEncodedFormEntity(body); + httpPut.setEntity(urlEncodedFormEntity); + } + if (StringUtils.isNotBlank(body) + && StringUtils.isNotBlank(contentType) + && contentType.contains(ContentType.MULTIPART_FORM_DATA.getMimeType())) { + MultipartEntityBuilder multipartEntityBuilder = getMultipartEntityBuilder(body); + httpPut.setEntity(multipartEntityBuilder.build()); + } + if (StringUtils.isNotBlank(body) + && StringUtils.isBlank(contentType)) { + HttpEntity entityPot = new StringEntity(body, MediaType.JSON_UTF_8.charset() + .get()); + httpPut.setEntity(entityPot); + } + return httpPut; + } + + private HttpPatch getHttpPatch(String url, Map headerMap, String body) { + HttpPatch httpPatch = new HttpPatch(url); + headerMap.forEach(httpPatch::addHeader); + httpPatch.addHeader(HTTP.CONN_DIRECTIVE, HTTP.CONN_KEEP_ALIVE); + String contentType = headerMap.get(HTTP.CONTENT_TYPE); + if (StringUtils.isNotBlank(body) + && StringUtils.isNotBlank(contentType) + && (contentType.contains(ContentType.APPLICATION_JSON.getMimeType()) + || contentType.contains(HttpConstant.CONTENT_TYPE_APPLICATION_JAVASCRIPT) + || contentType.contains(ContentType.APPLICATION_XML.getMimeType()) + || contentType.contains(ContentType.TEXT_PLAIN.getMimeType()) + || contentType.contains(ContentType.TEXT_XML.getMimeType()) + || contentType.contains(ContentType.TEXT_HTML.getMimeType()))) { + HttpEntity entityPot = new StringEntity(body, MediaType.JSON_UTF_8.charset() + .get()); + httpPatch.setEntity(entityPot); + } + if (StringUtils.isNotBlank(body) + && StringUtils.isNotBlank(contentType) + && contentType.contains(ContentType.APPLICATION_FORM_URLENCODED.getMimeType())) { + UrlEncodedFormEntity urlEncodedFormEntity = getUrlEncodedFormEntity(body); + httpPatch.setEntity(urlEncodedFormEntity); + } + if (StringUtils.isNotBlank(body) + && StringUtils.isNotBlank(contentType) + && contentType.contains(ContentType.MULTIPART_FORM_DATA.getMimeType())) { + MultipartEntityBuilder multipartEntityBuilder = getMultipartEntityBuilder(body); + httpPatch.setEntity(multipartEntityBuilder.build()); + } + if (StringUtils.isNotBlank(body) + && StringUtils.isBlank(contentType)) { + HttpEntity entityPot = new StringEntity(body, MediaType.JSON_UTF_8.charset() + .get()); + httpPatch.setEntity(entityPot); + } + return httpPatch; + } + + private UrlEncodedFormEntity getUrlEncodedFormEntity(String body) { + List params = new ArrayList<>(); + Map bodyMap = new Gson().fromJson(body, new TypeToken>() { + }.getType()); + bodyMap.forEach((k, v) -> params.add(new NameValuePair() { + @Override + public String getName() { + return k; + } + + @Override + public String getValue() { + return v; + } + })); + return new UrlEncodedFormEntity(params, StandardCharsets.UTF_8); + } + + private HttpPost getHttpPost(String url, Map headerMap, String body) { + HttpPost httpPost = new HttpPost(url); + headerMap.forEach(httpPost::addHeader); + httpPost.addHeader(HTTP.CONN_DIRECTIVE, HTTP.CONN_KEEP_ALIVE); + String contentType = headerMap.get(HTTP.CONTENT_TYPE); + if (StringUtils.isNotBlank(body) + && StringUtils.isNotBlank(contentType) + && (contentType.contains(ContentType.APPLICATION_JSON.getMimeType()) + || contentType.contains(HttpConstant.CONTENT_TYPE_APPLICATION_JAVASCRIPT) + || contentType.contains(ContentType.APPLICATION_XML.getMimeType()) + || contentType.contains(ContentType.TEXT_PLAIN.getMimeType()) + || contentType.contains(ContentType.TEXT_XML.getMimeType()) + || contentType.contains(ContentType.TEXT_HTML.getMimeType()))) { + HttpEntity entityPot = new StringEntity(body, MediaType.JSON_UTF_8.charset() + .get()); + httpPost.setEntity(entityPot); + } + if (StringUtils.isNotBlank(body) + && StringUtils.isNotBlank(contentType) + && contentType.contains(ContentType.APPLICATION_FORM_URLENCODED.getMimeType())) { + UrlEncodedFormEntity urlEncodedFormEntity = getUrlEncodedFormEntity(body); + httpPost.setEntity(urlEncodedFormEntity); + } + if (StringUtils.isNotBlank(body) + && StringUtils.isNotBlank(contentType) + && contentType.contains(ContentType.MULTIPART_FORM_DATA.getMimeType())) { + MultipartEntityBuilder multipartEntityBuilder = getMultipartEntityBuilder(body); + httpPost.setEntity(multipartEntityBuilder.build()); + } + if (StringUtils.isNotBlank(body) + && StringUtils.isBlank(contentType)) { + HttpEntity entityPot = new StringEntity(body, MediaType.JSON_UTF_8.charset() + .get()); + httpPost.setEntity(entityPot); + } + return httpPost; + } + + private HttpPost getHttpPostByte(String url, Map headerMap, byte[] body) { + HttpPost httpPost = new HttpPost(url); + headerMap.forEach(httpPost::addHeader); + ByteArrayEntity byteEntity = new ByteArrayEntity(body); + httpPost.setEntity(byteEntity); + + return httpPost; + } + + private MultipartEntityBuilder getMultipartEntityBuilder(String body) { + MultipartEntityBuilder multipartEntityBuilder = MultipartEntityBuilder.create(); + multipartEntityBuilder.setCharset(StandardCharsets.UTF_8); + Map bodyMap = new Gson().fromJson(body, new TypeToken>() { + }.getType()); + bodyMap.forEach((k, v) -> multipartEntityBuilder.addPart(k, new StringBody(v, ContentType.MULTIPART_FORM_DATA))); + return multipartEntityBuilder; + } + + + @Override + public void close() { + try { + executorServicePool.shutdownNow(); + httpClient.close(); + } catch (IOException e) { + log.error("ApacheHttpClientImpl | close | error => ", e); + } + } + + @Override + public void updateProxyConfig(ProxyConfig config) { + this.socksProxyConfig = new SocksProxyConfig(config.getSocks5Endpoint(), config.getSocks5UserName(), config.getSocks5Password()); + } + + static class SocksPlainConnectionSocketFactory extends PlainConnectionSocketFactory { + + @Override + public Socket createSocket(final HttpContext context) throws IOException { + InetSocketAddress socksaddr = (InetSocketAddress) context.getAttribute(SOCKS_ADDRESS_KEY); + if (socksaddr != null) { + Proxy proxy = new Proxy(Proxy.Type.SOCKS, socksaddr); + return new Socket(proxy); + } else { + return new Socket(); + } + } + + @Override + public Socket connectSocket(int connectTimeout, Socket socket, HttpHost host, InetSocketAddress remoteAddress, + InetSocketAddress localAddress, HttpContext context) throws IOException { + InetSocketAddress socksaddr = (InetSocketAddress) context.getAttribute(SOCKS_ADDRESS_KEY); + if (socksaddr != null) { + //make proxy server to resolve host in http url + remoteAddress = InetSocketAddress.createUnresolved(host.getHostName(), host.getPort()); + } + return super.connectSocket(connectTimeout, socket, host, remoteAddress, localAddress, context); + } + } + + static class FakeDnsResolver implements DnsResolver { + @Override + public InetAddress[] resolve(String host) throws UnknownHostException { + // Return some fake DNS record for every request, we won't be using it + try { + return new InetAddress[]{InetAddress.getByName(host)}; + } catch (Throwable e) { + return new InetAddress[]{InetAddress.getByAddress(new byte[]{0, 0, 0, 0})}; + } + } + } + + static class SocksSSLConnectionSocketFactory extends SSLConnectionSocketFactory { + public SocksSSLConnectionSocketFactory(SSLContext sslContext) { + super(sslContext, NoopHostnameVerifier.INSTANCE); + } + + @Override + public Socket createSocket(final HttpContext context) throws IOException { + InetSocketAddress socksaddr = (InetSocketAddress) context.getAttribute(SOCKS_ADDRESS_KEY); + if (socksaddr != null) { + Proxy proxy = new Proxy(Proxy.Type.SOCKS, socksaddr); + return new Socket(proxy); + } else { + return new Socket(); + } + } + + @Override + public Socket connectSocket(int connectTimeout, Socket socket, HttpHost host, InetSocketAddress remoteAddress, + InetSocketAddress localAddress, HttpContext context) throws IOException { + InetSocketAddress socksaddr = (InetSocketAddress) context.getAttribute(SOCKS_ADDRESS_KEY); + if (socksaddr != null) { + remoteAddress = InetSocketAddress.createUnresolved(host.getHostName(), host.getPort()); + } + return super.connectSocket(connectTimeout, socket, host, remoteAddress, localAddress, context); + } + } + + static class NoopHostnameVerifier implements javax.net.ssl.HostnameVerifier { + public static final NoopHostnameVerifier INSTANCE = new NoopHostnameVerifier(); + + @Override + public boolean verify(String s, SSLSession sslSession) { + return true; + } + } + + @Override + public String toString() { + return new StringBuilder() + .append("Socks5Endpoint=").append(this.socksProxyConfig.getSocks5Endpoint()) + .append(", Socks5UserName=").append(this.socksProxyConfig.getSocks5UserName()) + .append(", Socks5Password=").append(this.socksProxyConfig.getSocks5Password()) + .toString(); + } +} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/auth/ApiKeyAuthImpl.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/auth/ApiKeyAuthImpl.java new file mode 100644 index 000000000..3958e3d9e --- /dev/null +++ b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/auth/ApiKeyAuthImpl.java @@ -0,0 +1,47 @@ +/* + * 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.rocketmq.connect.http.auth; + +import com.google.common.collect.Maps; +import org.apache.commons.lang3.StringUtils; +import org.apache.rocketmq.connect.http.entity.HttpApiKeyAuthParameters; +import org.apache.rocketmq.connect.http.entity.HttpAuthParameters; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Map; + +public class ApiKeyAuthImpl implements Auth { + private static final Logger log = LoggerFactory.getLogger(ApiKeyAuthImpl.class); + private HttpAuthParameters httpAuthParameters; + public ApiKeyAuthImpl(HttpAuthParameters httpAuthParameters) { + this.httpAuthParameters = httpAuthParameters; + } + @Override + public Map auth() { + Map headMap = Maps.newHashMap(); + try { + HttpApiKeyAuthParameters httpApiKeyAuthParameters = httpAuthParameters.getHttpApiKeyAuthParameters(); + if (httpApiKeyAuthParameters != null && StringUtils.isNotBlank(httpApiKeyAuthParameters.getApiKeyUsername()) && StringUtils.isNotBlank(httpApiKeyAuthParameters.getApiKeySecret())) { + headMap.put(httpApiKeyAuthParameters.getApiKeyUsername(), httpApiKeyAuthParameters.getApiKeySecret()); + } + } catch (Exception e) { + log.error("ApiKeyImpl | auth | error => ", e); + } + return headMap; + } +} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/auth/Auth.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/auth/Auth.java new file mode 100644 index 000000000..ff315fe65 --- /dev/null +++ b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/auth/Auth.java @@ -0,0 +1,23 @@ +/* + * 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.rocketmq.connect.http.auth; + +import java.util.Map; + +public interface Auth { + Map auth(); +} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/auth/BasicAuthImpl.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/auth/BasicAuthImpl.java new file mode 100644 index 000000000..e9c45ce5e --- /dev/null +++ b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/auth/BasicAuthImpl.java @@ -0,0 +1,53 @@ +/* + * 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.rocketmq.connect.http.auth; + +import com.google.common.collect.Maps; +import com.sun.org.apache.xerces.internal.impl.dv.util.Base64; +import org.apache.commons.lang3.StringUtils; +import org.apache.rocketmq.connect.http.constant.HttpHeaderConstant; +import org.apache.rocketmq.connect.http.entity.HttpAuthParameters; +import org.apache.rocketmq.connect.http.entity.HttpBasicAuthParameters; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.nio.charset.StandardCharsets; +import java.util.Map; + +public class BasicAuthImpl implements Auth { + private static final Logger log = LoggerFactory.getLogger(BasicAuthImpl.class); + private HttpAuthParameters httpAuthParameters; + + public BasicAuthImpl(HttpAuthParameters httpAuthParameters) { + this.httpAuthParameters = httpAuthParameters; + } + + @Override + public Map auth() { + Map headMap = Maps.newHashMap(); + try { + HttpBasicAuthParameters httpBasicAuthParameters = httpAuthParameters.getHttpBasicAuthParameters(); + if (httpBasicAuthParameters != null && StringUtils.isNotBlank(httpBasicAuthParameters.getUsername()) && StringUtils.isNotBlank(httpBasicAuthParameters.getPassword())) { + String authorizationValue = httpBasicAuthParameters.getUsername() + ":" + httpBasicAuthParameters.getPassword(); + headMap.put(HttpHeaderConstant.AUTHORIZATION, "Basic " + Base64.encode(authorizationValue.getBytes(StandardCharsets.UTF_8))); + } + } catch (Exception e) { + log.error("BasicAuthImpl | auth | error => ", e); + } + return headMap; + } +} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/auth/HttpCallback.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/auth/HttpCallback.java new file mode 100644 index 000000000..26469d44e --- /dev/null +++ b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/auth/HttpCallback.java @@ -0,0 +1,73 @@ +/* + * 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.rocketmq.connect.http.auth; + +import org.apache.http.concurrent.FutureCallback; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.concurrent.CountDownLatch; + +public class HttpCallback implements FutureCallback { + + private static final Logger log = LoggerFactory.getLogger(HttpCallback.class); + + private CountDownLatch countDownLatch; + + private boolean isFailed; + + private String msg; + + public HttpCallback(CountDownLatch countDownLatch) { + this.countDownLatch = countDownLatch; + } + + @Override + public void completed(String s) { + countDownLatch.countDown(); + } + + public void failed(final Exception ex) { + countDownLatch.countDown(); + isFailed = true; + log.error("http request failed.", ex); + } + + public void cancelled() { + countDownLatch.countDown(); + } + + public CountDownLatch getCountDownLatch() { + return countDownLatch; + } + + public boolean isFailed() { + return isFailed; + } + + public void setFailed(boolean failed) { + isFailed = failed; + } + + public String getMsg() { + return msg; + } + + public void setMsg(String msg) { + this.msg = msg; + } +} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/auth/HttpRequestCallable.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/auth/HttpRequestCallable.java new file mode 100644 index 000000000..309bbfc35 --- /dev/null +++ b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/auth/HttpRequestCallable.java @@ -0,0 +1,100 @@ +/* + * 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.rocketmq.connect.http.auth; + +import com.google.common.base.Strings; +import lombok.SneakyThrows; +import org.apache.http.client.methods.CloseableHttpResponse; +import org.apache.http.client.methods.HttpUriRequest; +import org.apache.http.client.protocol.HttpClientContext; +import org.apache.http.impl.client.CloseableHttpClient; +import org.apache.http.util.EntityUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import java.net.InetSocketAddress; +import java.nio.charset.StandardCharsets; +import java.text.MessageFormat; +import java.util.concurrent.Callable; + +public class HttpRequestCallable implements Callable { + + private static final Logger log = LoggerFactory.getLogger(HttpCallback.class); + + private CloseableHttpClient httpclient; + private HttpCallback httpCallback; + private HttpUriRequest httpUriRequest; + private HttpClientContext context; + private SocksProxyConfig socksProxyConfig; + private static final String SOCKS_ADDRESS_KEY = "socks.address"; + + public HttpRequestCallable(CloseableHttpClient httpclient, HttpUriRequest httpUriRequest, HttpClientContext context, + SocksProxyConfig socksProxyConfig, HttpCallback httpCallback) { + this.httpclient = httpclient; + this.httpUriRequest = httpUriRequest; + this.context = context; + this.socksProxyConfig = socksProxyConfig; + this.httpCallback = httpCallback; + } + + public void loadSocks5ProxyConfig() { + if (!Strings.isNullOrEmpty(socksProxyConfig.getSocks5Endpoint())) { + String[] socksAddrAndPor = socksProxyConfig.getSocks5Endpoint() + .split(":"); + InetSocketAddress socksaddr = new InetSocketAddress(socksAddrAndPor[0], + Integer.parseInt(socksAddrAndPor[1])); + context.setAttribute(SOCKS_ADDRESS_KEY, socksaddr); + ThreadLocalProxyAuthenticator.getInstance() + .setCredentials(socksProxyConfig.getSocks5UserName(), socksProxyConfig.getSocks5Password()); + } + } + + @SneakyThrows + @Override + public String call() throws Exception { + CloseableHttpResponse response = null; + try { + Long startTime = System.currentTimeMillis(); + loadSocks5ProxyConfig(); + response = httpclient.execute(httpUriRequest, context); + if (response != null && response.getEntity() != null) { + String result = EntityUtils.toString(response.getEntity(), StandardCharsets.UTF_8); + if (response.getStatusLine() + .getStatusCode() / 100 != 2) { + String msg = MessageFormat.format("Http Status:{0},Msg:{1}", response.getStatusLine() + .getStatusCode(), result); + httpCallback.setMsg(msg); + httpCallback.setFailed(Boolean.TRUE); + } + log.info("The cost of one http request:{}, Connection Connection={}, Keep-Alive={}", + System.currentTimeMillis() - startTime, response.getHeaders("Connection"), + response.getHeaders("Keep-Alive")); + return result; + } + } catch (Throwable e) { + log.error("http execute failed.", e); + httpCallback.setFailed(Boolean.TRUE); + httpCallback.setMsg(e.getLocalizedMessage()); + } finally { + httpCallback.getCountDownLatch() + .countDown(); + if (null != response.getEntity()) { + EntityUtils.consume(response.getEntity()); + } + } + return null; + } +} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/auth/OAuthClientImpl.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/auth/OAuthClientImpl.java new file mode 100644 index 000000000..f62217927 --- /dev/null +++ b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/auth/OAuthClientImpl.java @@ -0,0 +1,174 @@ +/* + * 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.rocketmq.connect.http.auth; + +import com.alibaba.fastjson.JSONObject; +import com.google.common.collect.Maps; +import org.apache.commons.lang3.StringUtils; +import org.apache.rocketmq.connect.http.constant.HttpHeaderConstant; +import org.apache.rocketmq.connect.http.entity.HttpAuthParameters; +import org.apache.rocketmq.connect.http.entity.HttpOAuthParameters; +import org.apache.rocketmq.connect.http.entity.HttpRequest; +import org.apache.rocketmq.connect.http.entity.TokenEntity; +import org.apache.rocketmq.connect.http.util.JsonUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Map; +import java.util.concurrent.CountDownLatch; + +public class OAuthClientImpl implements Auth, Runnable { + private static final Logger log = LoggerFactory.getLogger(OAuthClientImpl.class); + private volatile HttpOAuthParameters httpOAuthParameters; + private volatile TokenEntity tokenEntity; + private AbstractHttpClient httpClient; + private HttpAuthParameters httpAuthParameters; + + public OAuthClientImpl(HttpAuthParameters httpAuthParameters) { + this.httpAuthParameters = httpAuthParameters; + } + + @Override + public Map auth() { + try { + HttpOAuthParameters httpOAuthParameters = httpAuthParameters.getHttpOAuthParameters(); + String resultToken = ""; + if (httpOAuthParameters != null && StringUtils.isNotBlank(httpOAuthParameters.getEndpoint()) + && StringUtils.isNotBlank(httpOAuthParameters.getHttpMethod())) { + if (tokenEntity != null) { + Map headMap = Maps.newHashMap(); + headMap.put(HttpHeaderConstant.AUTHORIZATION, "Bearer " + tokenEntity.getAccessToken()); + return headMap; + } + this.httpOAuthParameters = httpOAuthParameters; + HttpRequest httpRequest = new HttpRequest(); + resultToken = getResultToken(httpOAuthParameters, httpRequest); + if (StringUtils.isNotBlank(resultToken)) { + final TokenEntity token = JSONObject.parseObject(resultToken, TokenEntity.class); + if (StringUtils.isNotBlank(token.getAccessToken())) { + Map tokenHeadMap = Maps.newHashMap(); + tokenHeadMap.put(HttpHeaderConstant.AUTHORIZATION, "Bearer " + token.getAccessToken()); + token.setTokenTimestamp(Long.toString(System.currentTimeMillis())); + tokenEntity = token; + return tokenHeadMap; + } else { + throw new RuntimeException(token.getError()); + } + } + } + } catch (Exception e) { + log.error("OAuthClientImpl | auth | error => ", e); + throw new RuntimeException(e); + } + return Maps.newHashMap(); + } + + public String getResultToken(HttpOAuthParameters httpOAuthParameters, HttpRequest httpRequest) throws Exception { + Map headMap = Maps.newHashMap(); + if (HttpHeaderConstant.POST_METHOD.equals(httpOAuthParameters.getHttpMethod()) + || HttpHeaderConstant.PUT_METHOD.equals(httpOAuthParameters.getHttpMethod()) + || HttpHeaderConstant.PATCH_METHOD.equals(httpOAuthParameters.getHttpMethod())) { + String headerParameters = httpOAuthParameters.getHeaderParameters(); + JSONObject jsonObject = JSONObject.parseObject(headerParameters); + if (jsonObject != null) { + for (Map.Entry entry : jsonObject.entrySet()) { + headMap.put(entry.getKey(), (String) entry.getValue()); + } + } + httpRequest.setBody(httpOAuthParameters.getBodyParameters()); + httpRequest.setUrl(JsonUtils.addQueryStringAndPathValueToUrl(httpOAuthParameters.getEndpoint(), getOAuthQueryParameters(httpOAuthParameters))); + httpRequest.setMethod(httpOAuthParameters.getHttpMethod()); + httpRequest.setTimeout(httpOAuthParameters.getTimeout()); + httpRequest.setHeaderMap(headMap); + } else { + httpRequest.setUrl(JsonUtils.addQueryStringAndPathValueToUrl(httpOAuthParameters.getEndpoint(), getOAuthQueryParameters(httpOAuthParameters))); + httpRequest.setTimeout(httpOAuthParameters.getTimeout()); + httpRequest.setMethod(httpOAuthParameters.getHttpMethod()); + httpRequest.setHeaderMap(headMap); + httpRequest.setBody(StringUtils.EMPTY); + } + CountDownLatch countDownLatch = new CountDownLatch(1); + HttpCallback httpCallback = new HttpCallback(countDownLatch); + return httpClient.execute(httpRequest, httpCallback); + } + + private String getOAuthQueryParameters(HttpOAuthParameters httpOAuthParameters) { + JSONObject queryParameters = StringUtils.isBlank(httpOAuthParameters.getQueryParameters()) ? + new JSONObject() : JSONObject.parseObject(httpOAuthParameters.getQueryParameters()); + if (httpOAuthParameters.getClientID() != null && httpOAuthParameters.getClientSecret() != null) { + JSONObject additionalQueryParameters = new JSONObject(); + additionalQueryParameters.put("client_id", httpOAuthParameters.getClientID()); + additionalQueryParameters.put("client_secret", httpOAuthParameters.getClientSecret()); + queryParameters = JsonUtils.mergeJson(additionalQueryParameters, queryParameters); + } + return queryParameters.toJSONString(); + } + + @Override + public void run() { + log.info("OAuthTokenRunnable | run"); + if (httpAuthParameters == null || httpOAuthParameters == null) { + log.info("httpAuthParameters or httpOAuthParameters is null | break"); + return; + } + String resultToken = ""; + long tokenTimestamp = Long.parseLong(tokenEntity.getTokenTimestamp()) + (tokenEntity.getExpiresIn() * 1000L); + log.info("OAuthTokenRunnable | run | tokenTimestamp : {} | system.currentTimeMillis : {} | boolean : {}", tokenTimestamp, System.currentTimeMillis(), System.currentTimeMillis() > tokenTimestamp); + if (System.currentTimeMillis() > tokenTimestamp) { + log.info("OAuthTokenRunnable | run | update token"); + HttpRequest httpRequest = new HttpRequest(); + try { + resultToken = this.getResultToken(httpOAuthParameters, httpRequest); + } catch (Exception e) { + log.error("OAuthTokenRunnable | update token | scheduledExecutorService | error => ", e); + throw new RuntimeException(e); + } + if (StringUtils.isNotBlank(resultToken)) { + final TokenEntity token = JSONObject.parseObject(resultToken, TokenEntity.class); + if (StringUtils.isNotBlank(token.getAccessToken())) { + updateTokenEntity(tokenEntity, token); + } else { + throw new RuntimeException(token.getError()); + } + } + } + } + + private void updateTokenEntity(TokenEntity oldTokenEntity, TokenEntity newTokenEntity) { + if (newTokenEntity != null) { + if (StringUtils.isNotBlank(newTokenEntity.getAccessToken())) { + oldTokenEntity.setAccessToken(newTokenEntity.getAccessToken()); + } + oldTokenEntity.setExpiresIn(newTokenEntity.getExpiresIn()); + if (StringUtils.isNotBlank(newTokenEntity.getScope())) { + oldTokenEntity.setScope(newTokenEntity.getScope()); + } + if (StringUtils.isNotBlank(newTokenEntity.getTokenType())) { + oldTokenEntity.setTokenType(newTokenEntity.getTokenType()); + } + } + oldTokenEntity.setTokenTimestamp(Long.toString(System.currentTimeMillis())); + } + + public AbstractHttpClient getHttpClient() { + return httpClient; + } + + public void setHttpClient(AbstractHttpClient httpClient) { + this.httpClient = httpClient; + } +} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/auth/SocksProxyConfig.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/auth/SocksProxyConfig.java new file mode 100644 index 000000000..56a51e3ca --- /dev/null +++ b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/auth/SocksProxyConfig.java @@ -0,0 +1,54 @@ +/* + * 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.rocketmq.connect.http.auth; + +public class SocksProxyConfig { + + private String socks5Endpoint; + private String socks5UserName; + private String socks5Password; + + public SocksProxyConfig(String socks5Endpoint, String socks5UserName, String socks5Password) { + this.socks5Endpoint = socks5Endpoint; + this.socks5UserName = socks5UserName; + this.socks5Password = socks5Password; + } + + public String getSocks5Endpoint() { + return socks5Endpoint; + } + + public void setSocks5Endpoint(String socks5Endpoint) { + this.socks5Endpoint = socks5Endpoint; + } + + public String getSocks5UserName() { + return socks5UserName; + } + + public void setSocks5UserName(String socks5UserName) { + this.socks5UserName = socks5UserName; + } + + public String getSocks5Password() { + return socks5Password; + } + + public void setSocks5Password(String socks5Password) { + this.socks5Password = socks5Password; + } +} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/auth/ThreadLocalProxyAuthenticator.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/auth/ThreadLocalProxyAuthenticator.java new file mode 100644 index 000000000..4cfb09cad --- /dev/null +++ b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/auth/ThreadLocalProxyAuthenticator.java @@ -0,0 +1,45 @@ +/* + * 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.rocketmq.connect.http.auth; + +import java.net.Authenticator; +import java.net.PasswordAuthentication; + +public class ThreadLocalProxyAuthenticator extends Authenticator { + + private ThreadLocal credential = new ThreadLocal(); + + private static class SingletonHolder { + private static final ThreadLocalProxyAuthenticator INSTANCE = new ThreadLocalProxyAuthenticator(); + } + + public static final ThreadLocalProxyAuthenticator getInstance() { + return SingletonHolder.INSTANCE; + } + + public void setCredentials(String user, String password) { + credential.set(new PasswordAuthentication(user, password.toCharArray())); + Authenticator.setDefault(this); + } + + + @Override + public PasswordAuthentication getPasswordAuthentication() { + return credential.get(); + } + +} \ No newline at end of file diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/constant/AuthTypeEnum.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/constant/AuthTypeEnum.java new file mode 100644 index 000000000..654094098 --- /dev/null +++ b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/constant/AuthTypeEnum.java @@ -0,0 +1,52 @@ +/* + * 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.rocketmq.connect.http.constant; + +public enum AuthTypeEnum { + + /** + * BASIC + */ + BASIC("BASIC_AUTH"), + /** + * OAUTH_CLIENT_CREDENTIALS + */ + OAUTH_CLIENT_CREDENTIALS("OAUTH_AUTH"), + /** + * API_KEY + */ + API_KEY("API_KEY_AUTH"), + NONE("NONE"); + private final String authType; + + AuthTypeEnum(String authType) { + this.authType = authType; + } + + public String getAuthType() { + return authType; + } + + public static AuthTypeEnum parse(String authType) { + for (AuthTypeEnum authTypeEnum : AuthTypeEnum.values()) { + if (authTypeEnum.getAuthType().equals(authType)) { + return authTypeEnum; + } + } + return null; + } +} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/constant/HttpConstant.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/constant/HttpConstant.java new file mode 100644 index 000000000..e66049526 --- /dev/null +++ b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/constant/HttpConstant.java @@ -0,0 +1,54 @@ +/* + * 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.rocketmq.connect.http.constant; + +public class HttpConstant { + + public static final String URL = "url"; + public static final String METHOD = "method"; + // JsonString + public static final String QUERY_PARAMETERS = "queryParameters"; + // JsonString + public static final String FIXED_QUERY_PARAMETERS = "fixedQueryParameters"; + // JsonString + public static final String HEADER_PARAMETERS = "headerParameters"; + // JsonString + public static final String FIXED_HEADER_PARAMETERS = "fixedHeaderParameters"; + public static final String BODY = "body"; + public static final String SOCKS5_ENDPOINT = "socks5Endpoint"; + public static final String SOCKS5_USERNAME = "socks5UserName"; + public static final String SOCKS5_PASSWORD = "socks5Password"; + public static final String TIMEOUT = "timeout"; + public static final String CONCURRENCY = "concurrency"; + public static final String CONTENT_TYPE_APPLICATION_JAVASCRIPT = "application/javascript"; + public static final String TOKEN = "token"; + public static final String AUTH_TYPE = "authType"; + public static final String USE_KMS = "useKMS"; + public static final String BASIC_USERNAME = "basicUsername"; + public static final String BASIC_PASSWORD = "basicPassword"; + public static final String API_KEY_USERNAME = "apiKeyUsername"; + public static final String API_KEY_PASSWORD = "apiKeyPassword"; + public static final String OAUTH_ENDPOINT = "oAuthEndpoint"; + public static final String OAUTH_HTTP_METHOD = "oAuthHttpMethod"; + public static final String OAUTH_CLIENT_ID = "oAuthClientId"; + public static final String OAUTH_CLIENT_SECRET = "oAuthClientSecret"; + public static final String OAUTH_HEADER_PARAMETERS = "oAuthHeaderParameters"; + public static final String OAUTH_QUERY_PARAMETERS = "oAuthQueryParameters"; + public static final String OAUTH_BODY = "oAuthBody"; + public static final String ACCOUNT_ID = "accountId"; + +} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/constant/HttpHeaderConstant.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/constant/HttpHeaderConstant.java new file mode 100644 index 000000000..1eb518f6f --- /dev/null +++ b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/constant/HttpHeaderConstant.java @@ -0,0 +1,27 @@ +/* + * 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.rocketmq.connect.http.constant; + +public class HttpHeaderConstant { + public static final String CONTENT_TYPE = "Content-Type"; + public static final String CONTENT_LENGTH = "Content-Length"; + public static final String CONTENT_ENCODING = "Content-Encoding"; + public static final String AUTHORIZATION = "Authorization"; + public static final String POST_METHOD = "POST"; + public static final String PUT_METHOD = "PUT"; + public static final String PATCH_METHOD = "PATCH"; +} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/constant/MethodEnum.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/constant/MethodEnum.java new file mode 100644 index 000000000..ef8fe606b --- /dev/null +++ b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/constant/MethodEnum.java @@ -0,0 +1,41 @@ +/* + * 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.rocketmq.connect.http.constant; + +public enum MethodEnum { + + /** + * GET + */ + GET("GET"), + POST("POST"), + DELETE("DELETE"), + PUT("PUT"), + HEAD("HEAD"), + TRACE("TRACE"), + PATCH("PATCH"), + OPTIONS("OPTIONS"); + private final String method; + + MethodEnum(String method) { + this.method = method; + } + + public String getMethod() { + return method; + } +} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/HttpApiKeyAuthParameters.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/HttpApiKeyAuthParameters.java new file mode 100644 index 000000000..258499cef --- /dev/null +++ b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/HttpApiKeyAuthParameters.java @@ -0,0 +1,45 @@ +/* + * 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.rocketmq.connect.http.entity; + +public class HttpApiKeyAuthParameters { + + private String apiKeyUsername; + + private String apiKeySecret; + + public HttpApiKeyAuthParameters(String apiKeyUsername, String apiKeySecret) { + this.apiKeyUsername = apiKeyUsername; + this.apiKeySecret = apiKeySecret; + } + + public String getApiKeyUsername() { + return apiKeyUsername; + } + + public void setApiKeyUsername(String apiKeyUsername) { + this.apiKeyUsername = apiKeyUsername; + } + + public String getApiKeySecret() { + return apiKeySecret; + } + + public void setApiKeySecret(String apiKeySecret) { + this.apiKeySecret = apiKeySecret; + } +} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/HttpAuthParameters.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/HttpAuthParameters.java new file mode 100644 index 000000000..161cab052 --- /dev/null +++ b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/HttpAuthParameters.java @@ -0,0 +1,75 @@ +/* + * 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.rocketmq.connect.http.entity; + +import org.apache.commons.lang3.builder.ToStringBuilder; + +public class HttpAuthParameters { + + private String authType; + private HttpBasicAuthParameters httpBasicAuthParameters; + private HttpApiKeyAuthParameters httpApiKeyAuthParameters; + private HttpOAuthParameters httpOAuthParameters; + public HttpAuthParameters(String authType, HttpBasicAuthParameters httpBasicAuthParameters, HttpApiKeyAuthParameters httpApiKeyAuthParameters, HttpOAuthParameters httpOAuthParameters) { + this.authType = authType; + this.httpApiKeyAuthParameters = httpApiKeyAuthParameters; + this.httpBasicAuthParameters = httpBasicAuthParameters; + this.httpOAuthParameters = httpOAuthParameters; + } + + public String getAuthType() { + return authType; + } + + public void setAuthType(String authType) { + this.authType = authType; + } + + public HttpBasicAuthParameters getHttpBasicAuthParameters() { + return httpBasicAuthParameters; + } + + public void setHttpBasicAuthParameters(HttpBasicAuthParameters httpBasicAuthParameters) { + this.httpBasicAuthParameters = httpBasicAuthParameters; + } + + public HttpApiKeyAuthParameters getHttpApiKeyAuthParameters() { + return httpApiKeyAuthParameters; + } + + public void setHttpApiKeyAuthParameters(HttpApiKeyAuthParameters httpApiKeyAuthParameters) { + this.httpApiKeyAuthParameters = httpApiKeyAuthParameters; + } + + public HttpOAuthParameters getHttpOAuthParameters() { + return httpOAuthParameters; + } + + public void setHttpOAuthParameters(HttpOAuthParameters httpOAuthParameters) { + this.httpOAuthParameters = httpOAuthParameters; + } + + @Override + public String toString() { + return new ToStringBuilder(this) + .append("authType", authType) + .append("httpApiKeyAuthParameters", httpApiKeyAuthParameters) + .append("httpBasicAuthParameters", httpBasicAuthParameters) + .append("httpOAuthParameters", httpOAuthParameters) + .toString(); + } +} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/HttpBasicAuthParameters.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/HttpBasicAuthParameters.java new file mode 100644 index 000000000..a784014d3 --- /dev/null +++ b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/HttpBasicAuthParameters.java @@ -0,0 +1,44 @@ +/* + * 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.rocketmq.connect.http.entity; + +public class HttpBasicAuthParameters { + private String username; + private String password; + + + public HttpBasicAuthParameters(String username, String password) { + this.username = username; + this.password = password; + } + + public String getPassword() { + return password; + } + + public void setPassword(String password) { + this.password = password; + } + + public String getUsername() { + return username; + } + + public void setUsername(String username) { + this.username = username; + } +} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/HttpDeleteWithEntity.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/HttpDeleteWithEntity.java new file mode 100644 index 000000000..65eba4bfa --- /dev/null +++ b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/HttpDeleteWithEntity.java @@ -0,0 +1,41 @@ +/* + * 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.rocketmq.connect.http.entity; + +import org.apache.http.client.methods.HttpEntityEnclosingRequestBase; + +import java.net.URI; + +public class HttpDeleteWithEntity extends HttpEntityEnclosingRequestBase { + + private final static String GET_METHOD = "DELETE"; + + public HttpDeleteWithEntity(final URI uri) { + super(); + setURI(uri); + } + + public HttpDeleteWithEntity(final String uri) { + super(); + setURI(URI.create(uri)); + } + + @Override + public String getMethod() { + return GET_METHOD; + } +} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/HttpGetWithEntity.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/HttpGetWithEntity.java new file mode 100644 index 000000000..cfc24ca4c --- /dev/null +++ b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/HttpGetWithEntity.java @@ -0,0 +1,41 @@ +/* + * 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.rocketmq.connect.http.entity; + +import org.apache.http.client.methods.HttpEntityEnclosingRequestBase; + +import java.net.URI; + +public class HttpGetWithEntity extends HttpEntityEnclosingRequestBase { + + private final static String GET_METHOD = "GET"; + + public HttpGetWithEntity(final URI uri) { + super(); + setURI(uri); + } + + public HttpGetWithEntity(final String uri) { + super(); + setURI(URI.create(uri)); + } + + @Override + public String getMethod() { + return GET_METHOD; + } +} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/HttpHeadWithEntity.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/HttpHeadWithEntity.java new file mode 100644 index 000000000..dee7145c8 --- /dev/null +++ b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/HttpHeadWithEntity.java @@ -0,0 +1,41 @@ +/* + * 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.rocketmq.connect.http.entity; + +import org.apache.http.client.methods.HttpEntityEnclosingRequestBase; + +import java.net.URI; + +public class HttpHeadWithEntity extends HttpEntityEnclosingRequestBase { + + private final static String GET_METHOD = "HEAD"; + + public HttpHeadWithEntity(final URI uri) { + super(); + setURI(uri); + } + + public HttpHeadWithEntity(final String uri) { + super(); + setURI(URI.create(uri)); + } + + @Override + public String getMethod() { + return GET_METHOD; + } +} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/HttpOAuthParameters.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/HttpOAuthParameters.java new file mode 100644 index 000000000..42ec90524 --- /dev/null +++ b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/HttpOAuthParameters.java @@ -0,0 +1,104 @@ +/* + * 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.rocketmq.connect.http.entity; + +public class HttpOAuthParameters { + + private String endpoint; + private String httpMethod; + private String clientID; + private String clientSecret; + private String headerParameters; + private String queryParameters; + private String bodyParameters; + private String timeout; + + public HttpOAuthParameters(String endpoint, String httpMethod, String clientID, String clientSecret, String headerParameters, String queryParameters, String bodyParameters, String timeout) { + this.endpoint = endpoint; + this.httpMethod = httpMethod; + this.clientID = clientID; + this.clientSecret = clientSecret; + this.headerParameters = headerParameters; + this.queryParameters = queryParameters; + this.bodyParameters = bodyParameters; + this.timeout = timeout; + } + + public String getEndpoint() { + return endpoint; + } + + public void setEndpoint(String oAuthEndpoint) { + this.endpoint = oAuthEndpoint; + } + + public String getHttpMethod() { + return httpMethod; + } + + public void setHttpMethod(String httpMethod) { + this.httpMethod = httpMethod; + } + + public String getClientID() { + return clientID; + } + + public void setClientID(String clientID) { + this.clientID = clientID; + } + + public String getClientSecret() { + return clientSecret; + } + + public void setClientSecret(String clientSecret) { + this.clientSecret = clientSecret; + } + + public String getHeaderParameters() { + return headerParameters; + } + + public void setHeaderParameters(String headerParameters) { + this.headerParameters = headerParameters; + } + + public String getQueryParameters() { + return queryParameters; + } + + public void setQueryParameters(String queryParameters) { + this.queryParameters = queryParameters; + } + + public String getBodyParameters() { + return bodyParameters; + } + + public void setBodyParameters(String bodyParameters) { + this.bodyParameters = bodyParameters; + } + + public String getTimeout() { + return timeout; + } + + public void setTimeout(String timeout) { + this.timeout = timeout; + } +} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/HttpRequest.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/HttpRequest.java new file mode 100644 index 000000000..5c6be8915 --- /dev/null +++ b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/HttpRequest.java @@ -0,0 +1,78 @@ +/* + * 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.rocketmq.connect.http.entity; + +import java.util.Map; + +public class HttpRequest { + + private String url; + private String method; + private Map headerMap; + private String body; + private String timeout; + + public byte[] getBytesBody() { + return bytesBody; + } + + public void setBytesBody(byte[] bytesBody) { + this.bytesBody = bytesBody; + } + + private byte[] bytesBody; + + public String getUrl() { + return url; + } + + public void setUrl(String url) { + this.url = url; + } + + public String getMethod() { + return method; + } + + public void setMethod(String method) { + this.method = method; + } + + public Map getHeaderMap() { + return headerMap; + } + + public void setHeaderMap(Map headerMap) { + this.headerMap = headerMap; + } + + public String getBody() { + return body; + } + + public void setBody(String body) { + this.body = body; + } + + public String getTimeout() { + return timeout; + } + + public void setTimeout(String timeout) { + this.timeout = timeout; + } +} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/HttpTraceWithEntity.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/HttpTraceWithEntity.java new file mode 100644 index 000000000..b2d81be6c --- /dev/null +++ b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/HttpTraceWithEntity.java @@ -0,0 +1,41 @@ +/* + * 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.rocketmq.connect.http.entity; + +import org.apache.http.client.methods.HttpEntityEnclosingRequestBase; + +import java.net.URI; + +public class HttpTraceWithEntity extends HttpEntityEnclosingRequestBase { + + private final static String GET_METHOD = "TRACE"; + + public HttpTraceWithEntity(final URI uri) { + super(); + setURI(uri); + } + + public HttpTraceWithEntity(final String uri) { + super(); + setURI(URI.create(uri)); + } + + @Override + public String getMethod() { + return GET_METHOD; + } +} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/ProxyConfig.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/ProxyConfig.java new file mode 100644 index 000000000..d9589c874 --- /dev/null +++ b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/ProxyConfig.java @@ -0,0 +1,46 @@ +/* + * 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.rocketmq.connect.http.entity; + +public class ProxyConfig { + private String socks5UserName; + private String socks5Password; + private String socks5Endpoint; + public String getSocks5UserName() { + return socks5UserName; + } + + public void setSocks5UserName(String socks5UserName) { + this.socks5UserName = socks5UserName; + } + + public String getSocks5Password() { + return socks5Password; + } + + public void setSocks5Password(String socks5Password) { + this.socks5Password = socks5Password; + } + + public String getSocks5Endpoint() { + return socks5Endpoint; + } + + public void setSocks5Endpoint(String socks5Endpoint) { + this.socks5Endpoint = socks5Endpoint; + } +} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/TokenEntity.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/TokenEntity.java new file mode 100644 index 000000000..8ea62fdef --- /dev/null +++ b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/entity/TokenEntity.java @@ -0,0 +1,140 @@ +/* + * 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.rocketmq.connect.http.entity; + +import org.apache.commons.lang3.builder.ToStringBuilder; + +public class TokenEntity { + + private String accessToken; + private String tokenType; + private int expiresIn; + private String exampleParameter; + private String timestamp; + private String status; + private String error; + private String message; + private String path; + private String tokenTimestamp; + + private String scope; + + public String getAccessToken() { + return accessToken; + } + + public void setAccessToken(String accessToken) { + this.accessToken = accessToken; + } + + public String getTokenType() { + return tokenType; + } + + public void setTokenType(String tokenType) { + this.tokenType = tokenType; + } + + public int getExpiresIn() { + return expiresIn; + } + + public void setExpiresIn(int expiresIn) { + this.expiresIn = expiresIn; + } + + public String getTokenTimestamp() { + return tokenTimestamp; + } + + public void setTokenTimestamp(String tokenTimestamp) { + this.tokenTimestamp = tokenTimestamp; + } + + public String getExampleParameter() { + return exampleParameter; + } + + public void setExampleParameter(String exampleParameter) { + this.exampleParameter = exampleParameter; + } + + public String getTimestamp() { + return timestamp; + } + + public void setTimestamp(String timestamp) { + this.timestamp = timestamp; + } + + public String getStatus() { + return status; + } + + public void setStatus(String status) { + this.status = status; + } + + public String getError() { + return error; + } + + public void setError(String error) { + this.error = error; + } + + public String getMessage() { + return message; + } + + public void setMessage(String message) { + this.message = message; + } + + public String getPath() { + return path; + } + + public void setPath(String path) { + this.path = path; + } + + public String getScope() { + return scope; + } + + public void setScope(String scope) { + this.scope = scope; + } + + @Override + public String toString() { + return new ToStringBuilder(this) + .append("accessToken", accessToken) + .append("tokenType", tokenType) + .append("expiresIn", expiresIn) + .append("exampleParameter", exampleParameter) + .append("timestamp", timestamp) + .append("status", status) + .append("error", error) + .append("message", message) + .append("path", path) + .append("tokenTimestamp", tokenTimestamp) + .append("scope", scope) + .toString(); + } +} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/sink/HttpSinkConnector.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/sink/HttpSinkConnector.java deleted file mode 100644 index 5a9193fca..000000000 --- a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/sink/HttpSinkConnector.java +++ /dev/null @@ -1,59 +0,0 @@ -package org.apache.rocketmq.connect.http.sink; - -import org.apache.rocketmq.connect.http.sink.constant.HttpConstant; -import io.openmessaging.KeyValue; -import io.openmessaging.connector.api.component.task.Task; -import io.openmessaging.connector.api.component.task.sink.SinkConnector; -import io.openmessaging.internal.DefaultKeyValue; -import org.apache.commons.lang3.StringUtils; - -import java.net.URL; -import java.net.URLConnection; -import java.util.ArrayList; -import java.util.List; - -public class HttpSinkConnector extends SinkConnector { - - private KeyValue connectConfig; - - @Override - public List taskConfigs(int maxTasks) { - List keyValueList = new ArrayList<>(11); - KeyValue keyValue = new DefaultKeyValue(); - for (String key : connectConfig.keySet()) { - keyValue.put(key, connectConfig.getString(key)); - } - keyValueList.add(keyValue); - return keyValueList; - } - - @Override - public Class taskClass() { - return HttpSinkTask.class; - } - - @Override - public void validate(KeyValue config) { - if (StringUtils.isBlank(config.getString(HttpConstant.URL_CONSTANT))) { - throw new RuntimeException("http required parameter is null !"); - } - try { - URL urlConnect = new URL(config.getString(HttpConstant.URL_CONSTANT)); - URLConnection urlConnection = urlConnect.openConnection(); - urlConnection.setConnectTimeout(5000); - urlConnection.connect(); - } catch (Exception e) { - throw new RuntimeException(e.getMessage()); - } - } - - @Override - public void start(KeyValue config) { - this.connectConfig = config; - } - - @Override - public void stop() { - - } -} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/sink/HttpSinkTask.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/sink/HttpSinkTask.java deleted file mode 100644 index f561b654c..000000000 --- a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/sink/HttpSinkTask.java +++ /dev/null @@ -1,41 +0,0 @@ -package org.apache.rocketmq.connect.http.sink; - -import io.openmessaging.KeyValue; -import io.openmessaging.connector.api.component.task.sink.SinkTask; -import io.openmessaging.connector.api.data.ConnectRecord; -import io.openmessaging.connector.api.errors.ConnectException; -import org.apache.rocketmq.connect.http.sink.common.OkHttpUtils; -import org.apache.rocketmq.connect.http.sink.constant.HttpConstant; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import java.util.List; - -public class HttpSinkTask extends SinkTask { - private static final Logger log = LoggerFactory.getLogger(HttpSinkTask.class); - - private String url; - - @Override - public void put(List sinkRecords) throws ConnectException { - try { - sinkRecords.forEach(connectRecord -> OkHttpUtils.builder() - .url(url) - .addParam(HttpConstant.DATA_CONSTANT, connectRecord.getData().toString()) - .post(true) - .sync()); - } catch (Exception e) { - log.error("HttpSinkTask | put | error => ", e); - } - } - - @Override - public void start(KeyValue keyValue) { - url = keyValue.getString(HttpConstant.URL_CONSTANT); - } - - @Override - public void stop() { - - } -} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/sink/common/OkHttpUtils.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/sink/common/OkHttpUtils.java deleted file mode 100644 index 2c3fe6325..000000000 --- a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/sink/common/OkHttpUtils.java +++ /dev/null @@ -1,274 +0,0 @@ -package org.apache.rocketmq.connect.http.sink.common; - -import com.alibaba.fastjson.JSON; -import okhttp3.*; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -import javax.net.ssl.SSLContext; -import javax.net.ssl.SSLSocketFactory; -import javax.net.ssl.TrustManager; -import javax.net.ssl.X509TrustManager; -import java.io.IOException; -import java.net.URLEncoder; -import java.security.SecureRandom; -import java.security.cert.X509Certificate; -import java.util.LinkedHashMap; -import java.util.Map; -import java.util.concurrent.Semaphore; -import java.util.concurrent.TimeUnit; - -public class OkHttpUtils { - private static final Logger log = LoggerFactory.getLogger(OkHttpUtils.class); - - private static volatile OkHttpClient okHttpClient = null; - private static volatile Semaphore semaphore = null; - private Map headerMap; - private Map paramMap; - private String url; - private Request.Builder request; - - private OkHttpUtils() { - if (okHttpClient == null) { - synchronized (OkHttpUtils.class) { - if (okHttpClient == null) { - TrustManager[] trustManagers = buildTrustManagers(); - okHttpClient = new OkHttpClient.Builder() - .connectTimeout(15, TimeUnit.SECONDS) - .writeTimeout(20, TimeUnit.SECONDS) - .readTimeout(20, TimeUnit.SECONDS) - .sslSocketFactory(createSSLSocketFactory(trustManagers), (X509TrustManager) trustManagers[0]) - .hostnameVerifier((hostName, session) -> true) - .retryOnConnectionFailure(true) - .build(); - addHeader("User-Agent", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_12_6) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/63.0.3239.132 Safari/537.36"); - } - } - } - } - - private static Semaphore getSemaphoreInstance() { - synchronized (OkHttpUtils.class) { - if (semaphore == null) { - semaphore = new Semaphore(0); - } - } - return semaphore; - } - - public static OkHttpUtils builder() { - return new OkHttpUtils(); - } - - public OkHttpUtils url(String url) { - this.url = url; - return this; - } - - /** - * 添加参数 - * - * @param key 参数名 - * @param value 参数值 - * @return - */ - public OkHttpUtils addParam(String key, String value) { - if (paramMap == null) { - paramMap = new LinkedHashMap<>(16); - } - paramMap.put(key, value); - return this; - } - - /** - * 添加请求头 - * - * @param key 参数名 - * @param value 参数值 - * @return - */ - public OkHttpUtils addHeader(String key, String value) { - if (headerMap == null) { - headerMap = new LinkedHashMap<>(16); - } - headerMap.put(key, value); - return this; - } - - public OkHttpUtils get() { - request = new Request.Builder().get(); - StringBuilder urlBuilder = new StringBuilder(url); - if (paramMap != null) { - urlBuilder.append("?"); - try { - for (Map.Entry entry : paramMap.entrySet()) { - urlBuilder.append(URLEncoder.encode(entry.getKey(), "utf-8")). - append("="). - append(URLEncoder.encode(entry.getValue(), "utf-8")). - append("&"); - } - } catch (Exception e) { - log.error("OkHttpUtils | get | error => ", e); - } - urlBuilder.deleteCharAt(urlBuilder.length() - 1); - } - request.url(urlBuilder.toString()); - return this; - } - - /** - * 初始化post方法 - * - * @param isJsonPost true等于json的方式提交数据，类似postman里post方法的raw - * false等于普通的表单提交 - * @return - */ - public OkHttpUtils post(boolean isJsonPost) { - RequestBody requestBody; - if (isJsonPost) { - String json = ""; - if (paramMap != null) { - json = JSON.toJSONString(paramMap); - } - requestBody = RequestBody.create(MediaType.parse("application/json; charset=utf-8"), json); - } else { - FormBody.Builder formBody = new FormBody.Builder(); - if (paramMap != null) { - paramMap.forEach(formBody::add); - } - requestBody = formBody.build(); - } - request = new Request.Builder().post(requestBody).url(url); - return this; - } - - /** - * 同步请求 - * - * @return - */ - public String sync() { - setHeader(request); - try { - Response response = okHttpClient.newCall(request.build()).execute(); - assert response.body() != null; - return response.body().string(); - } catch (IOException e) { - log.error("OkHttpUtils | sync | error => ", e); - return "请求失败：" + e.getMessage(); - } - } - - /** - * 异步请求，有返回值 - */ - public String async() { - StringBuilder buffer = new StringBuilder(""); - setHeader(request); - okHttpClient.newCall(request.build()).enqueue(new Callback() { - @Override - public void onFailure(Call call, IOException e) { - buffer.append("请求出错：").append(e.getMessage()); - } - - @Override - public void onResponse(Call call, Response response) throws IOException { - assert response.body() != null; - buffer.append(response.body().string()); - getSemaphoreInstance().release(); - } - }); - try { - getSemaphoreInstance().acquire(); - } catch (InterruptedException e) { - log.error("OkHttpUtils | async | error => ", e); - } - return buffer.toString(); - } - - /** - * 异步请求，带有接口回调 - * - * @param callBack - */ - public void async(ICallBack callBack) { - setHeader(request); - okHttpClient.newCall(request.build()).enqueue(new Callback() { - @Override - public void onFailure(Call call, IOException e) { - callBack.onFailure(call, e.getMessage()); - } - - @Override - public void onResponse(Call call, Response response) throws IOException { - assert response.body() != null; - callBack.onSuccessful(call, response.body().string()); - } - }); - } - - /** - * 为request添加请求头 - * - * @param request - */ - private void setHeader(Request.Builder request) { - if (headerMap != null) { - try { - for (Map.Entry entry : headerMap.entrySet()) { - request.addHeader(entry.getKey(), entry.getValue()); - } - } catch (Exception e) { - log.error("OkHttpUtils | setHeader | error => ", e); - } - } - } - - - /** - * 生成安全套接字工厂，用于https请求的证书跳过 - * - * @return - */ - private static SSLSocketFactory createSSLSocketFactory(TrustManager[] trustAllCerts) { - SSLSocketFactory ssfFactory = null; - try { - SSLContext sc = SSLContext.getInstance("SSL"); - sc.init(null, trustAllCerts, new SecureRandom()); - ssfFactory = sc.getSocketFactory(); - } catch (Exception e) { - log.error("OkHttpUtils | createSSLSocketFactory | error => ", e); - } - return ssfFactory; - } - - private static TrustManager[] buildTrustManagers() { - return new TrustManager[]{ - new X509TrustManager() { - @Override - public void checkClientTrusted(X509Certificate[] chain, String authType) { - } - - @Override - public void checkServerTrusted(X509Certificate[] chain, String authType) { - } - - @Override - public X509Certificate[] getAcceptedIssuers() { - return new X509Certificate[]{}; - } - } - }; - } - - /** - * 自定义一个接口回调 - */ - public interface ICallBack { - - void onSuccessful(Call call, String data); - - void onFailure(Call call, String errorMsg); - - } -} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/sink/constant/HttpConstant.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/sink/constant/HttpConstant.java deleted file mode 100644 index 09532b042..000000000 --- a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/sink/constant/HttpConstant.java +++ /dev/null @@ -1,8 +0,0 @@ -package org.apache.rocketmq.connect.http.sink.constant; - -public class HttpConstant { - - public static final String URL_CONSTANT = "url"; - - public static final String DATA_CONSTANT = "data"; -} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/util/CheckUtils.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/util/CheckUtils.java new file mode 100644 index 000000000..af268ec42 --- /dev/null +++ b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/util/CheckUtils.java @@ -0,0 +1,49 @@ +/* + * 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.rocketmq.connect.http.util; + +import org.apache.commons.lang3.StringUtils; + +public class CheckUtils { + + private static final String NULL_CONSTANT = "null"; + + public static Boolean checkNull(String check) { + if (StringUtils.isBlank(check)) { + return Boolean.TRUE; + } + if (StringUtils.isNotBlank(check) && NULL_CONSTANT.equalsIgnoreCase(check)) { + return Boolean.TRUE; + } + return Boolean.FALSE; + } + + public static Boolean checkNotNull(String check) { + if (StringUtils.isNotBlank(check) && !NULL_CONSTANT.equalsIgnoreCase(check)) { + return Boolean.TRUE; + } + return Boolean.FALSE; + } + + public static String checkNullReturnDefault(String check) { + if (NULL_CONSTANT.equalsIgnoreCase(check)) { + return null; + } + return check; + } + +} diff --git a/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/util/JsonUtils.java b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/util/JsonUtils.java new file mode 100644 index 000000000..850a3546e --- /dev/null +++ b/connectors/rocketmq-connect-http/src/main/java/org/apache/rocketmq/connect/http/util/JsonUtils.java @@ -0,0 +1,80 @@ +/* + * 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.rocketmq.connect.http.util; + +import com.alibaba.fastjson.JSONArray; +import com.alibaba.fastjson.JSONObject; +import org.apache.commons.lang3.StringUtils; +import java.io.UnsupportedEncodingException; +import java.net.URLEncoder; +import java.util.Map.Entry; + +public class JsonUtils { + public final static String QUESTION_MARK = "?"; + public static JSONObject mergeJson(JSONObject source, JSONObject target) { + if (target == null) { + return source; + } + if (source == null) { + return target; + } + for (String key : source.keySet()) { + Object value = source.get(key); + if (!target.containsKey(key)) { + target.put(key, value); + } else { + if (value instanceof JSONObject) { + JSONObject valueJson = (JSONObject) value; + JSONObject targetValue = mergeJson(valueJson, target.getJSONObject(key)); + target.put(key, targetValue); + } else if (value instanceof JSONArray) { + JSONArray valueArray = (JSONArray) value; + for (int i = 0; i < valueArray.size(); i++) { + JSONObject obj = (JSONObject) valueArray.get(i); + JSONObject targetValue = mergeJson(obj, (JSONObject) target.getJSONArray(key).get(i)); + target.getJSONArray(key).set(i, targetValue); + } + } else { + target.put(key, value); + } + } + } + return target; + } + + public static String addQueryStringAndPathValueToUrl(String url, String queryString) throws UnsupportedEncodingException { + StringBuilder queryStringBuilder = new StringBuilder(); + if (StringUtils.isNotBlank(queryString)) { + final JSONObject jsonObject = JSONObject.parseObject(queryString); + for (Entry next : jsonObject.entrySet()) { + if (next.getValue() instanceof JSONObject) { + queryStringBuilder.append(URLEncoder.encode(next.getKey(), "UTF-8")).append("=").append(URLEncoder.encode(((JSONObject) next.getValue()).toJSONString(), "UTF-8")).append("&"); + } else { + queryStringBuilder.append(URLEncoder.encode(next.getKey(), "UTF-8")).append("=").append(URLEncoder.encode((String) next.getValue(), "UTF-8")).append("&"); + } + } + } + String path = queryStringBuilder.toString(); + if (StringUtils.isNotBlank(path) && StringUtils.isNotBlank(url)) { + if (url.contains(QUESTION_MARK)) { + return url + "&" + path.substring(0, path.length() - 1); + } + return url + "?" + path.substring(0, path.length() - 1); + } + return url; + } +} diff --git a/connectors/rocketmq-connect-http/src/test/java/org/apache/rocketmq/connect/http/HttpSinkTaskTest.java b/connectors/rocketmq-connect-http/src/test/java/org/apache/rocketmq/connect/http/HttpSinkTaskTest.java new file mode 100644 index 000000000..9b6d2b7d3 --- /dev/null +++ b/connectors/rocketmq-connect-http/src/test/java/org/apache/rocketmq/connect/http/HttpSinkTaskTest.java @@ -0,0 +1,646 @@ +package org.apache.rocketmq.connect.http; + +import com.alibaba.fastjson.JSONObject; +import com.google.common.collect.Maps; +import com.google.gson.Gson; +import io.openmessaging.KeyValue; +import io.openmessaging.connector.api.component.task.sink.SinkTaskContext; +import io.openmessaging.connector.api.data.ConnectRecord; +import io.openmessaging.connector.api.data.RecordOffset; +import io.openmessaging.connector.api.data.RecordPartition; +import io.openmessaging.internal.DefaultKeyValue; +import java.lang.reflect.Field; +import java.nio.charset.StandardCharsets; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.Set; +import org.apache.rocketmq.connect.http.auth.AbstractHttpClient; +import org.apache.rocketmq.connect.http.auth.HttpCallback; +import org.apache.rocketmq.connect.http.constant.AuthTypeEnum; +import org.apache.rocketmq.connect.http.constant.HttpConstant; +import org.apache.rocketmq.connect.http.entity.HttpRequest; +import org.apache.rocketmq.connect.http.entity.TokenEntity; +import org.assertj.core.util.Lists; +import org.junit.Assert; +import org.junit.Before; +import org.junit.Ignore; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.mockito.Mock; +import org.mockito.junit.MockitoJUnitRunner; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; + +@RunWith(MockitoJUnitRunner.class) +public class HttpSinkTaskTest { + + @Mock + AbstractHttpClient httpClient; + + @Before + public void setUp() throws Exception { + } + + @Test + public void testHttpPutWithoutHeaderAndQueryAndAuth() throws Exception { + doAnswer(invocationOnMock -> { + HttpRequest httpRequest = invocationOnMock.getArgument(0); + Assert.assertEquals("http://localhost:8080/api", httpRequest.getUrl()); + Assert.assertEquals("POST", httpRequest.getMethod()); + Assert.assertEquals("{token=token}", httpRequest.getHeaderMap().toString()); + Assert.assertEquals("123", httpRequest.getBody()); + HttpCallback httpCallback = invocationOnMock.getArgument(1); + httpCallback.getCountDownLatch() + .countDown(); + return null; + }).when(httpClient).executeNotReturn(any(), any()); + HttpSinkTask httpSinkTask = new HttpSinkTask(); + KeyValue keyValue = new DefaultKeyValue(); + keyValue.put(HttpConstant.URL, "http://localhost:8080/api"); + keyValue.put(HttpConstant.METHOD, "POST"); + keyValue.put(HttpConstant.TIMEOUT, 60000); + Map bodyMap = Maps.newHashMap(); + bodyMap.put("form", "CONSTANT"); + bodyMap.put("value", "123"); + keyValue.put(HttpConstant.BODY, "123"); + keyValue.put(HttpConstant.TOKEN, "token"); + httpSinkTask.validate(keyValue); + + List connectRecordList = Lists.newArrayList(); + ConnectRecord connectRecord = new ConnectRecord(null, null, System.currentTimeMillis()); + connectRecord.setData(JSONObject.toJSONString(new HashMap<>())); + connectRecordList.add(connectRecord); + httpSinkTask.init(new SinkTaskContext() { + @Override + public String getConnectorName() { + return null; + } + + @Override + public String getTaskName() { + return null; + } + + @Override public KeyValue configs() { + return null; + } + + @Override + public void resetOffset(RecordPartition recordPartition, RecordOffset recordOffset) { + + } + + @Override + public void resetOffset(Map offsets) { + + } + + @Override + public void pause(List partitions) { + + } + + @Override + public void resume(List partitions) { + + } + + @Override + public Set assignment() { + return null; + } + }); + httpSinkTask.start(keyValue); + Field field = httpSinkTask.getClass().getDeclaredField("httpClient"); + field.setAccessible(true); + field.set(httpSinkTask, httpClient); + httpSinkTask.put(connectRecordList); + httpSinkTask.stop(); + } + + @Test + public void testHttpPutWithHeaderAndQueryWithoutAuth() throws Exception { + doAnswer(invocationOnMock -> { + HttpRequest httpRequest = invocationOnMock.getArgument(0); + System.out.println(new Gson().toJson(httpRequest)); + Assert.assertEquals("http://localhost:8080/api?action=action&accountId=accountId", + httpRequest.getUrl()); + Assert.assertEquals("POST", httpRequest.getMethod()); + Assert.assertEquals("{meetingName=swqd, id=45ef4dewdwe1-7c35-447a-bd93-fab****, userId=199525, groupId=456, token=token}", httpRequest.getHeaderMap().toString()); + Assert.assertEquals("{\"form\":\"CONSTANT\",\"value\":\"123\"}", httpRequest.getBody()); + HttpCallback httpCallback = invocationOnMock.getArgument(1); + httpCallback.getCountDownLatch() + .countDown(); + return null; + }).when(httpClient).executeNotReturn(any(), any()); + HttpSinkTask httpSinkTask = new HttpSinkTask(); + KeyValue keyValue = new DefaultKeyValue(); + keyValue.put(HttpConstant.URL, "http://localhost:8080/api"); + keyValue.put(HttpConstant.METHOD, "POST"); + keyValue.put(HttpConstant.TIMEOUT, 60000); + keyValue.put(HttpConstant.SOCKS5_ENDPOINT, "127.0.0.1"); + keyValue.put(HttpConstant.SOCKS5_USERNAME, "xxxx"); + keyValue.put(HttpConstant.SOCKS5_PASSWORD, "xxxx"); + keyValue.put(HttpConstant.HEADER_PARAMETERS, "{\n" + + " \"meetingName\": \"swqd\",\n" + + " \"groupId\": \"456\"\n" + + "}"); + keyValue.put(HttpConstant.FIXED_HEADER_PARAMETERS, "{\n" + + " \"id\": \"45ef4dewdwe1-7c35-447a-bd93-fab****\",\n" + + " \"userId\": \"199525\"\n" + + "}"); + keyValue.put(HttpConstant.QUERY_PARAMETERS, "{\n" + + " \"action\": \"action\"" + + "}"); + keyValue.put(HttpConstant.FIXED_QUERY_PARAMETERS, "{\n" + + " \"accountId\": \"accountId\"" + + "}"); + keyValue.put(HttpConstant.TOKEN, "token"); + Map bodyMap = Maps.newHashMap(); + bodyMap.put("form", "CONSTANT"); + bodyMap.put("value", "123"); + keyValue.put(HttpConstant.BODY, JSONObject.toJSONString(bodyMap)); + httpSinkTask.validate(keyValue); + + List connectRecordList = Lists.newArrayList(); + ConnectRecord connectRecord = new ConnectRecord(null, null, System.currentTimeMillis()); + String cloudEvent = "{\n" + + " \"data\":{\n" + + " \"meetingName\":\"swqd\",\n" + + " \"groupId\":\"456\",\n" + + " \"action\":\"camera_off\",\n" + + " \"time\":1590592527490,\n" + + " \"userId\":\"199525\",\n" + + " \"meetingUUID\":\"hz-20864c8f-b10d-45cd-9935-884bca1b****\"\n" + + " },\n" + + " \"id\":\"45ef4dewdwe1-7c35-447a-bd93-fab****\",\n" + + " \"source\":\"acs:aliyuncvc\",\n" + + " \"specversion\":\"1.0\",\n" + + " \"subject\":\"acs.aliyuncvc:cn-hangzhou:{AccountId}:215672\",\n" + + " \"time\":\"2020-11-19T21:04:41+08:00\",\n" + + " \"type\":\"aliyuncvc:MeetingEvent:MemberOperate\",\n" + + " \"aliyunaccountid\":\"123456789098****\",\n" + + " \"aliyunpublishtime\":\"2020-11-19T21:04:42.179PRC\",\n" + + " \"aliyuneventbusname\":\"default\",\n" + + " \"aliyunregionid\":\"cn-hangzhou\",\n" + + " \"aliyunpublishaddr\":\"172.25.XX.XX\"\n" + + "}"; + connectRecord.setData(cloudEvent); + connectRecordList.add(connectRecord); + httpSinkTask.init(new SinkTaskContext() { + @Override + public String getConnectorName() { + return null; + } + + @Override + public String getTaskName() { + return null; + } + + @Override public KeyValue configs() { + return null; + } + + @Override + public void resetOffset(RecordPartition recordPartition, RecordOffset recordOffset) { + + } + + @Override + public void resetOffset(Map offsets) { + + } + + @Override + public void pause(List partitions) { + + } + + @Override + public void resume(List partitions) { + + } + + @Override + public Set assignment() { + return null; + } + }); + httpSinkTask.start(keyValue); + Field field = httpSinkTask.getClass().getDeclaredField("httpClient"); + field.setAccessible(true); + field.set(httpSinkTask, httpClient); + httpSinkTask.put(connectRecordList); + httpSinkTask.stop(); + } + + @Test + public void testHttpPutWithHeaderAndQueryWithBasicAuth() throws Exception { + doAnswer(invocationOnMock -> { + HttpRequest httpRequest = invocationOnMock.getArgument(0); + System.out.println(new Gson().toJson(httpRequest)); + Assert.assertEquals("http://localhost:8080/api?action=action&accountId=accountId", + httpRequest.getUrl()); + Assert.assertEquals("POST", httpRequest.getMethod()); + Assert.assertEquals("{Authorization=Basic dXNlcm5hbWU6cGFzc3dvcmQ=, meetingName=swqd, id=45ef4dewdwe1-7c35-447a-bd93-fab****, userId=199525, groupId=456, token=token}", httpRequest.getHeaderMap().toString()); + Assert.assertEquals("{\"form\":\"CONSTANT\",\"value\":\"123\"}", httpRequest.getBody()); + HttpCallback httpCallback = invocationOnMock.getArgument(1); + httpCallback.getCountDownLatch() + .countDown(); + return null; + }).when(httpClient).executeNotReturn(any(), any()); + HttpSinkTask httpSinkTask = new HttpSinkTask(); + KeyValue keyValue = new DefaultKeyValue(); + keyValue.put(HttpConstant.URL, "http://localhost:8080/api"); + keyValue.put(HttpConstant.METHOD, "POST"); + keyValue.put(HttpConstant.TIMEOUT, 60000); + keyValue.put(HttpConstant.SOCKS5_ENDPOINT, "127.0.0.1"); + keyValue.put(HttpConstant.SOCKS5_USERNAME, "xxxx"); + keyValue.put(HttpConstant.SOCKS5_PASSWORD, "xxxx"); + keyValue.put(HttpConstant.HEADER_PARAMETERS, "{\n" + + " \"meetingName\": \"swqd\",\n" + + " \"groupId\": \"456\"\n" + + "}"); + keyValue.put(HttpConstant.FIXED_HEADER_PARAMETERS, "{\n" + + " \"id\": \"45ef4dewdwe1-7c35-447a-bd93-fab****\",\n" + + " \"userId\": \"199525\"\n" + + "}"); + keyValue.put(HttpConstant.QUERY_PARAMETERS, "{\n" + + " \"action\": \"action\"" + + "}"); + keyValue.put(HttpConstant.FIXED_QUERY_PARAMETERS, "{\n" + + " \"accountId\": \"accountId\"" + + "}"); + keyValue.put(HttpConstant.TOKEN, "token"); + keyValue.put(HttpConstant.AUTH_TYPE, "BASIC_AUTH"); + keyValue.put(HttpConstant.BASIC_USERNAME, "username"); + keyValue.put(HttpConstant.BASIC_PASSWORD, "password"); + Map bodyMap = Maps.newHashMap(); + bodyMap.put("form", "CONSTANT"); + bodyMap.put("value", "123"); + keyValue.put(HttpConstant.BODY, JSONObject.toJSONString(bodyMap)); + httpSinkTask.validate(keyValue); + + List connectRecordList = Lists.newArrayList(); + ConnectRecord connectRecord = new ConnectRecord(null, null, System.currentTimeMillis()); + String cloudEvent = "{\n" + + " \"data\":{\n" + + " \"meetingName\":\"swqd\",\n" + + " \"groupId\":\"456\",\n" + + " \"action\":\"camera_off\",\n" + + " \"time\":1590592527490,\n" + + " \"userId\":\"199525\",\n" + + " \"meetingUUID\":\"hz-20864c8f-b10d-45cd-9935-884bca1b****\"\n" + + " },\n" + + " \"id\":\"45ef4dewdwe1-7c35-447a-bd93-fab****\",\n" + + " \"source\":\"acs:aliyuncvc\",\n" + + " \"specversion\":\"1.0\",\n" + + " \"subject\":\"acs.aliyuncvc:cn-hangzhou:{AccountId}:215672\",\n" + + " \"time\":\"2020-11-19T21:04:41+08:00\",\n" + + " \"type\":\"aliyuncvc:MeetingEvent:MemberOperate\",\n" + + " \"aliyunaccountid\":\"123456789098****\",\n" + + " \"aliyunpublishtime\":\"2020-11-19T21:04:42.179PRC\",\n" + + " \"aliyuneventbusname\":\"default\",\n" + + " \"aliyunregionid\":\"cn-hangzhou\",\n" + + " \"aliyunpublishaddr\":\"172.25.XX.XX\"\n" + + "}"; + connectRecord.setData(cloudEvent); + connectRecordList.add(connectRecord); + httpSinkTask.init(new SinkTaskContext() { + @Override + public String getConnectorName() { + return null; + } + + @Override + public String getTaskName() { + return null; + } + + @Override public KeyValue configs() { + return null; + } + + @Override + public void resetOffset(RecordPartition recordPartition, RecordOffset recordOffset) { + + } + + @Override + public void resetOffset(Map offsets) { + + } + + @Override + public void pause(List partitions) { + + } + + @Override + public void resume(List partitions) { + + } + + @Override + public Set assignment() { + return null; + } + }); + httpSinkTask.start(keyValue); + Field field = httpSinkTask.getClass().getDeclaredField("httpClient"); + field.setAccessible(true); + field.set(httpSinkTask, httpClient); + httpSinkTask.put(connectRecordList); + httpSinkTask.stop(); + } + + + @Test + public void testHttpPutWithHeaderAndQueryWithApiKey() throws Exception { + doAnswer(invocationOnMock -> { + HttpRequest httpRequest = invocationOnMock.getArgument(0); + System.out.println(new Gson().toJson(httpRequest)); + Assert.assertEquals("http://localhost:8080/api?action=action&accountId=accountId", + httpRequest.getUrl()); + Assert.assertEquals("POST", httpRequest.getMethod()); + Assert.assertEquals("{meetingName=swqd, id=45ef4dewdwe1-7c35-447a-bd93-fab****, userId=199525, groupId=456, token=token, username=password}", httpRequest.getHeaderMap().toString()); + Assert.assertEquals("\"data\"", httpRequest.getBody()); + HttpCallback httpCallback = invocationOnMock.getArgument(1); + httpCallback.getCountDownLatch() + .countDown(); + return null; + }).when(httpClient).executeNotReturn(any(), any()); + HttpSinkTask httpSinkTask = new HttpSinkTask(); + KeyValue keyValue = new DefaultKeyValue(); + keyValue.put(HttpConstant.URL, "http://localhost:8080/api"); + keyValue.put(HttpConstant.METHOD, "POST"); + keyValue.put(HttpConstant.TIMEOUT, 60000); + keyValue.put(HttpConstant.SOCKS5_ENDPOINT, "127.0.0.1"); + keyValue.put(HttpConstant.SOCKS5_USERNAME, "xxxx"); + keyValue.put(HttpConstant.SOCKS5_PASSWORD, "xxxx"); + keyValue.put(HttpConstant.HEADER_PARAMETERS, "{\n" + + " \"meetingName\": \"swqd\",\n" + + " \"groupId\": \"456\"\n" + + "}"); + keyValue.put(HttpConstant.FIXED_HEADER_PARAMETERS, "{\n" + + " \"id\": \"45ef4dewdwe1-7c35-447a-bd93-fab****\",\n" + + " \"userId\": \"199525\"\n" + + "}"); + keyValue.put(HttpConstant.QUERY_PARAMETERS, "{\n" + + " \"action\": \"action\"" + + "}"); + keyValue.put(HttpConstant.FIXED_QUERY_PARAMETERS, "{\n" + + " \"accountId\": \"accountId\"" + + "}"); + keyValue.put(HttpConstant.TOKEN, "token"); + keyValue.put(HttpConstant.AUTH_TYPE, AuthTypeEnum.API_KEY.getAuthType()); + keyValue.put(HttpConstant.API_KEY_USERNAME, "username"); + keyValue.put(HttpConstant.API_KEY_PASSWORD, "password"); + httpSinkTask.validate(keyValue); + List connectRecordList = Lists.newArrayList(); + ConnectRecord connectRecord = new ConnectRecord(null, null, System.currentTimeMillis()); + connectRecord.setData("data"); + connectRecordList.add(connectRecord); + httpSinkTask.init(new SinkTaskContext() { + @Override + public String getConnectorName() { + return null; + } + + @Override + public String getTaskName() { + return null; + } + + @Override public KeyValue configs() { + return null; + } + + @Override + public void resetOffset(RecordPartition recordPartition, RecordOffset recordOffset) { + + } + + @Override + public void resetOffset(Map offsets) { + + } + + @Override + public void pause(List partitions) { + + } + + @Override + public void resume(List partitions) { + + } + + @Override + public Set assignment() { + return null; + } + }); + httpSinkTask.start(keyValue); + Field field = httpSinkTask.getClass().getDeclaredField("httpClient"); + field.setAccessible(true); + field.set(httpSinkTask, httpClient); + httpSinkTask.put(connectRecordList); + httpSinkTask.stop(); + } + + @Test + public void testHttpPutWithHeaderAndQueryWithOAuth() throws Exception { + doAnswer(invocationOnMock -> { + HttpRequest httpRequest = invocationOnMock.getArgument(0); + System.out.println(new Gson().toJson(httpRequest)); + Assert.assertEquals("http://localhost:8080/api?action=action&accountId=accountId", + httpRequest.getUrl()); + Assert.assertEquals("POST", httpRequest.getMethod()); + Assert.assertEquals("{Authorization=Bearer oAuthAccessToken, meetingName=swqd, id=45ef4dewdwe1-7c35-447a-bd93-fab****, userId=199525, groupId=456, token=token}", httpRequest.getHeaderMap().toString()); + Assert.assertEquals("\"data\"", httpRequest.getBody()); + HttpCallback httpCallback = invocationOnMock.getArgument(1); + httpCallback.getCountDownLatch() + .countDown(); + return null; + }).when(httpClient).executeNotReturn(any(), any()); + // for OAuth request + doAnswer(invocationOnMock -> { + HttpRequest httpRequest = invocationOnMock.getArgument(0); + Assert.assertEquals("{\"url\":\"oAuthEndpoint?headerKey1\\u003dheaderValue1\\u0026client_secret\\u003doAuthClientSecret\\u0026client_id\\u003doAuthClientId\",\"method\":\"POST\",\"headerMap\":{\"headerKey1\":\"headerValue1\"},\"body\":\"oAuthBody\",\"timeout\":\"3000\"}", new Gson().toJson(httpRequest)); + TokenEntity tokenEntity = new TokenEntity(); + tokenEntity.setAccessToken("oAuthAccessToken"); + tokenEntity.setTokenType("tokenType"); + tokenEntity.setExpiresIn(100); + tokenEntity.setError("error"); + tokenEntity.setMessage("message"); + tokenEntity.setPath("path"); + tokenEntity.setScope("scope"); + tokenEntity.setTokenTimestamp("tokenTimestamp"); + tokenEntity.setTimestamp("timestamp"); + tokenEntity.setExampleParameter("exampleParameter"); + tokenEntity.setStatus("status"); + return JSONObject.toJSONString(tokenEntity); + }).when(httpClient).execute(any(), any()); + HttpSinkTask httpSinkTask = new HttpSinkTask(); + KeyValue keyValue = new DefaultKeyValue(); + keyValue.put(HttpConstant.URL, "http://localhost:8080/api"); + keyValue.put(HttpConstant.METHOD, "POST"); + keyValue.put(HttpConstant.TIMEOUT, 60000); + keyValue.put(HttpConstant.SOCKS5_ENDPOINT, "127.0.0.1"); + keyValue.put(HttpConstant.SOCKS5_USERNAME, "xxxx"); + keyValue.put(HttpConstant.SOCKS5_PASSWORD, "xxxx"); + keyValue.put(HttpConstant.HEADER_PARAMETERS, "{\n" + + " \"meetingName\": \"swqd\",\n" + + " \"groupId\": \"456\"\n" + + "}"); + keyValue.put(HttpConstant.FIXED_HEADER_PARAMETERS, "{\n" + + " \"id\": \"45ef4dewdwe1-7c35-447a-bd93-fab****\",\n" + + " \"userId\": \"199525\"\n" + + "}"); + keyValue.put(HttpConstant.QUERY_PARAMETERS, "{\n" + + " \"action\": \"action\"" + + "}"); + keyValue.put(HttpConstant.FIXED_QUERY_PARAMETERS, "{\n" + + " \"accountId\": \"accountId\"" + + "}"); + keyValue.put(HttpConstant.TOKEN, "token"); + keyValue.put(HttpConstant.AUTH_TYPE, "OAUTH_AUTH"); + String oAuthEndpoint = "oAuthEndpoint"; + String oAuthHttpMethod = "POST"; + String oAuthClientId = "oAuthClientId"; + String oAuthClientSecret = "oAuthClientSecret"; + String oAuthHeaderParameters = "{\n" + + " \"headerKey1\": \"headerValue1\"" + + "}"; + String oAuthQueryParameters = "{\n" + + " \"queryKey1\": \"queryValue1\"" + + "}"; + String oAuthBody = "oAuthBody"; + keyValue.put(HttpConstant.OAUTH_ENDPOINT, oAuthEndpoint); + keyValue.put(HttpConstant.OAUTH_HTTP_METHOD, oAuthHttpMethod); + keyValue.put(HttpConstant.OAUTH_CLIENT_ID, oAuthClientId); + keyValue.put(HttpConstant.OAUTH_CLIENT_SECRET, oAuthClientSecret); + keyValue.put(HttpConstant.OAUTH_HEADER_PARAMETERS, oAuthHeaderParameters); + keyValue.put(HttpConstant.OAUTH_QUERY_PARAMETERS, oAuthQueryParameters); + keyValue.put(HttpConstant.OAUTH_BODY, oAuthBody); + httpSinkTask.validate(keyValue); + List connectRecordList = Lists.newArrayList(); + ConnectRecord connectRecord = new ConnectRecord(null, null, System.currentTimeMillis()); + connectRecord.setData("data"); + connectRecordList.add(connectRecord); + httpSinkTask.init(new SinkTaskContext() { + @Override + public String getConnectorName() { + return null; + } + + @Override + public String getTaskName() { + return null; + } + + @Override public KeyValue configs() { + return null; + } + + @Override + public void resetOffset(RecordPartition recordPartition, RecordOffset recordOffset) { + + } + + @Override + public void resetOffset(Map offsets) { + + } + + @Override + public void pause(List partitions) { + + } + + @Override + public void resume(List partitions) { + + } + + @Override + public Set assignment() { + return null; + } + }); + httpSinkTask.start(keyValue); + Field field = httpSinkTask.getClass().getDeclaredField("httpClient"); + field.setAccessible(true); + field.set(httpSinkTask, httpClient); + httpSinkTask.updateAuth(AuthTypeEnum.OAUTH_CLIENT_CREDENTIALS.getAuthType(), null, null, null, null, oAuthEndpoint, oAuthHttpMethod, oAuthClientId, oAuthClientSecret, oAuthHeaderParameters, oAuthHeaderParameters, oAuthBody); + httpSinkTask.put(connectRecordList); + httpSinkTask.stop(); + } + + @Test + @Ignore + // use for invoke local service + public void test() throws Exception { + HttpSinkTask httpSinkTask = new HttpSinkTask(); + KeyValue keyValue = new DefaultKeyValue(); + // test basic + keyValue.put(HttpConstant.URL, "http://localhost:8080"); + keyValue.put(HttpConstant.METHOD, "GET"); + keyValue.put(HttpConstant.TIMEOUT, 60000); + keyValue.put(HttpConstant.FIXED_HEADER_PARAMETERS, "{\n" + + " \"Content-Type\": \"application/json\",\n" + + " \"user\": \"user\"" + + "}"); + httpSinkTask.validate(keyValue); + List connectRecordList = Lists.newArrayList(); + ConnectRecord connectRecord = new ConnectRecord(null, null, System.currentTimeMillis()); + connectRecord.setData("data"); + connectRecordList.add(connectRecord); + httpSinkTask.init(new SinkTaskContext() { + @Override + public String getConnectorName() { + return null; + } + + @Override + public String getTaskName() { + return null; + } + + @Override public KeyValue configs() { + return null; + } + + @Override + public void resetOffset(RecordPartition recordPartition, RecordOffset recordOffset) { + + } + + @Override + public void resetOffset(Map offsets) { + + } + + @Override + public void pause(List partitions) { + + } + + @Override + public void resume(List partitions) { + + } + + @Override + public Set assignment() { + return null; + } + }); + httpSinkTask.start(keyValue); + httpSinkTask.put(connectRecordList); + httpSinkTask.stop(); + } +} diff --git a/connectors/rocketmq-connect-http/src/test/java/org/apache/rocketmq/connect/http/sink/HttpSinkConnectorTest.java b/connectors/rocketmq-connect-http/src/test/java/org/apache/rocketmq/connect/http/sink/HttpSinkConnectorTest.java deleted file mode 100644 index ac9efeadd..000000000 --- a/connectors/rocketmq-connect-http/src/test/java/org/apache/rocketmq/connect/http/sink/HttpSinkConnectorTest.java +++ /dev/null @@ -1,42 +0,0 @@ -package org.apache.rocketmq.connect.http.sink; - -import io.openmessaging.KeyValue; -import io.openmessaging.connector.api.data.ConnectRecord; -import io.openmessaging.internal.DefaultKeyValue; -import org.apache.rocketmq.connect.http.sink.constant.HttpConstant; -import org.junit.Assert; -import org.junit.Test; - -import java.util.ArrayList; -import java.util.List; - -public class HttpSinkConnectorTest { - - private final HttpSinkConnector httpSinkConnector = new HttpSinkConnector(); - - @Test - public void testTaskConfigs() { - Assert.assertEquals(httpSinkConnector.taskConfigs(1).size(), 1); - } - - @Test - public void testPut() { - HttpSinkTask httpSinkTask = new HttpSinkTask(); - KeyValue keyValue = new DefaultKeyValue(); - keyValue.put(HttpConstant.URL_CONSTANT, "http://127.0.0.1:8081/demo"); - httpSinkTask.start(keyValue); - List connectRecordList = new ArrayList<>(); - ConnectRecord connectRecord = new ConnectRecord(null ,null, System.currentTimeMillis()); - connectRecord.setData("test"); - connectRecordList.add(connectRecord); - httpSinkTask.put(connectRecordList); - } - - @Test(expected = RuntimeException.class) - public void testValidate() { - KeyValue keyValue = new DefaultKeyValue(); - // 需要添加测试的http地址 - keyValue.put(HttpConstant.URL_CONSTANT, "http://127.0.0.1"); - httpSinkConnector.validate(keyValue); - } -}