gfwleak/zx-dataserker-yb-flume-cus-sink-file
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feature(cookieId): 因总线connect 接口改为请求携带cookie 将不返回session问题，修改请求参数中不设置请求cookie。

Browse Source
This commit is contained in:
doufenghu
2020-11-24 21:08:47 +08:00
parent be38915890
commit 6c536289e0
3 changed files with 12 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
yb_http_avro_sink_file/src/main/java/cn/ac/iie/cusflume/sink/HttpAsyncUtils/HttpClientUtil.java
View File

@@ -143,10 +143,8 @@ public class HttpClientUtil {
HttpClientContext localContext = HttpClientContext.create(); HttpClientContext localContext = HttpClientContext.create();
BasicCookieStore cookieStore = new BasicCookieStore(); BasicCookieStore cookieStore = new BasicCookieStore();
httpPut = new HttpPut(baseUrl); httpPut = new HttpPut(baseUrl);
// httpPost.setHeader("Transfer-Encoding","close"); // httpPost.setHeader("Transfer-Encoding","close");
// httpPost.setHeader("Content-Type", "binary/octet-stream"); // httpPost.setHeader("Content-Type", "binary/octet-stream");
httpPut.setHeader("Content-Type", "binary/octet-stream"); httpPut.setHeader("Content-Type", "binary/octet-stream");

5
yb_http_avro_sink_file/src/main/java/cn/ac/iie/cusflume/sink/YbHttpAvroSinkFile.java
View File

@@ -90,6 +90,7 @@ public class YbHttpAvroSinkFile extends AbstractSink implements Configurable {
logger.error("Starting YbHttpAvroSinkFile is error==>checkMsgUrl and postMsgUrl can not be null!!!!"); logger.error("Starting YbHttpAvroSinkFile is error==>checkMsgUrl and postMsgUrl can not be null!!!!");
} }
new Thread(new RetrySendMessage()).start(); new Thread(new RetrySendMessage()).start();
logger.warn("Starting YbHttpAvroSinkFile ... ..."); logger.warn("Starting YbHttpAvroSinkFile ... ...");
} }
@@ -348,6 +349,10 @@ public class YbHttpAvroSinkFile extends AbstractSink implements Configurable {
logger.info("AC msg sessionId already exists,msg sessionId is ===>" + msgSessionCookie); logger.info("AC msg sessionId already exists,msg sessionId is ===>" + msgSessionCookie);
} }
if (StringUtils.isBlank(configInfo.getMsgSessionCookie())) {
throw new RuntimeException("Get Cookie Error : " + configInfo.getCheckMsgUrl());
}
AvroMonitorTimerTask.addTotalNum(contents.size()); AvroMonitorTimerTask.addTotalNum(contents.size());
AvroMonitorTimerTask.addReadyPostNum(contents.size()); AvroMonitorTimerTask.addReadyPostNum(contents.size());
if (isSingle(topicName)) { if (isSingle(topicName)) {

4
yb_http_avro_sink_file/src/main/java/cn/ac/iie/cusflume/sink/avroUtils/HttpManager.java
View File

@@ -104,6 +104,7 @@ public class HttpManager {
} }
}; };
ctx.init(null, new TrustManager[]{tm}, null); ctx.init(null, new TrustManager[]{tm}, null);
SSLConnectionSocketFactory socketFactory = new SSLConnectionSocketFactory(ctx, NoopHostnameVerifier.INSTANCE); SSLConnectionSocketFactory socketFactory = new SSLConnectionSocketFactory(ctx, NoopHostnameVerifier.INSTANCE);
Registry<ConnectionSocketFactory> socketFactoryRegistry = RegistryBuilder.<ConnectionSocketFactory>create() Registry<ConnectionSocketFactory> socketFactoryRegistry = RegistryBuilder.<ConnectionSocketFactory>create()
.register("http", PlainConnectionSocketFactory.INSTANCE) .register("http", PlainConnectionSocketFactory.INSTANCE)
@@ -123,6 +124,7 @@ public class HttpManager {
} }
httpClient = getHttpClient(); httpClient = getHttpClient();
} }
//请求重试机制 //请求重试机制
@@ -169,6 +171,7 @@ public class HttpManager {
RequestConfig requestConfig = RequestConfig.custom() RequestConfig requestConfig = RequestConfig.custom()
.setConnectTimeout(3000) .setConnectTimeout(3000)
.setSocketTimeout(3000) .setSocketTimeout(3000)
.setCookieSpec(CookieSpecs.IGNORE_COOKIES)
//.setCookieSpec(CookieSpecs.BEST_MATCH) //.setCookieSpec(CookieSpecs.BEST_MATCH)
.build(); .build();
// 声明重定向策略对象 // 声明重定向策略对象
@@ -648,6 +651,7 @@ public class HttpManager {
httpPost.addHeader("User-Agent", userAgent); httpPost.addHeader("User-Agent", userAgent);
httpPost.addHeader("X-Tag", xTag);//根据最新文档,目前已经不需要此头-20191217 httpPost.addHeader("X-Tag", xTag);//根据最新文档,目前已经不需要此头-20191217
httpPost.addHeader("Content-Type", "application/json"); httpPost.addHeader("Content-Type", "application/json");
httpPost.setHeader("Cookie", "");
// httpPost.setHeader(HTTP.CONN_DIRECTIVE, HTTP.CONN_KEEP_ALIVE); // httpPost.setHeader(HTTP.CONN_DIRECTIVE, HTTP.CONN_KEEP_ALIVE);
StringEntity payload = new StringEntity(data, Charset.forName("utf-8")); StringEntity payload = new StringEntity(data, Charset.forName("utf-8"));
//payload.setContentType("text/xml; charset=UTF-8"); //payload.setContentType("text/xml; charset=UTF-8");