在 spring integration/xd http-outbound 中注入 2 个不同的 rest 模板
Injecting 2 different rest templates in spring integration/xd http-outbound
你好，我有像 below.Where 这样的流定义，我从 s3 中逐行提取文件并调用 http 客户端并命名为 channel.My transport is rabbit 和
stream aws-s3|custom processor| custom-http-client --url1=https://test1.com --url2=https://test2.com --filterAttribute=messageAttribute --httpMethod=POST --nonRetryErrorCodes=400,401,404,500 --charset=UTF-8 --replyTimeout=30000 --mapHeaders=Api-Key,Content-Type --requestTimeOut=30000 |processor> queue:testQueue
我的 http-config 如下所示，并使用 apache http 客户端进行连接池和多线程我将所有非常错误（如套接字超时）放回 DLQ 并重试。
现在我的 uri1 使用 Oauth，uri2 使用基本的 restTemplate。我如何向我的 http-outbound 注入两个 rest 模板？一个是oauth，另一个是basicTemplate？
<beans:beans xmlns="http://www.springframework.org/schema/integration"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:int-http="http://www.springframework.org/schema/integration/http"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/http
http://www.springframework.org/schema/integration/http/spring-integration-http.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context-4.0.xsd">
<!-- <context:property-placeholder location="${xd.module.config.location}\processor${xd.module.name}\batch-http.properties"
ignore-resource-not-found="true" local-override="true"/> -->
<context:property-placeholder />
<!-- logger changes start -->
<channel-interceptor pattern="*" order="3">
<beans:bean class="org.springframework.integration.channel.interceptor.WireTap">
<beans:constructor-arg ref="loggingChannel" />
</beans:bean>
</channel-interceptor>
<logging-channel-adapter id="loggingChannel" log-full-message="true" level="ERROR"/>
<!-- logger changes end -->
<header-filter input-channel="input"
output-channel="inputX" header-names="x-death"/>
<service-activator input-channel="inputX" ref="gw" />
<gateway id="gw" default-request-channel="toHttp" default-reply-timeout="0" error-channel="errors" />
<beans:bean id="inputfields" class="test.HTTPInputProperties">
<beans:property name="nonRetryErrorCodes" value="${nonRetryErrorCodes}"/>
</beans:bean>
<beans:bean id="responseInterceptor" class="test.ResponseInterceptor">
<beans:property name="inputProperties" ref="inputfields" />
</beans:bean>
<chain input-channel="errors" output-channel="output">
<!-- examine payload.cause (http status code etc) and decide whether
to throw an exception or return the status code for sending to output -->
<header-filter header-names="replyChannel, errorChannel" />
<transformer ref="responseInterceptor" />
</chain>
<int-http:outbound-gateway id='batch-http' header-mapper="headerMapper"
request-channel='toHttp'
rest-template="batchRestTemplate"
url-expression="payload.contains('${filterAttribute}') ? '${url1}' : '${url2}'" http-method="${httpMethod}"
expected-response-type='java.lang.String' charset='${charset}'
reply-timeout='${replyTimeout}' reply-channel='output'>
</int-http:outbound-gateway>
<beans:bean id="batchHTTPConverter" class="org.springframework.http.converter.StringHttpMessageConverter" >
<beans:constructor-arg index="0" value="${charset}"/>
<beans:property name="supportedMediaTypes" value = "application/json;UTF-8" />
</beans:bean>
<beans:bean id="batchRestTemplate" class="testBatchRestTemplate" >
<beans:constructor-arg name="requestTimeOut" value="${requestTimeOut}"/>
<beans:constructor-arg name="maxConnectionPerRoute" value="${maxConnectionPerRoute}"/>
<beans:constructor-arg name="totalMaxConnections" ref="${totalMaxConnections}"/>
</beans:bean>
<beans:bean id="headerMapper" class="org.springframework.integration.http.support.DefaultHttpHeaderMapper"
factory-method="outboundMapper">
<beans:property name="outboundHeaderNames" value="${mapHeaders}"/>
<beans:property name="userDefinedHeaderPrefix" value=""/>
</beans:bean>
<channel id="output" />
<channel id="input" />
<channel id="inputX" />
<channel id="toHttp" />
</beans:beans>
public class BatchRestTemplate extends RestTemplate{
private static final Logger LOGGER = LoggerFactory
.getLogger(BatchRestTemplate.class);
private static Integer requestTimeOut;
private static Integer totalMaxConnections;
private static Integer maxConnectionPerRoute;
public BatchRestTemplate(Integer requestTimeOut,Integer totalMaxConnections,Integer maxConnectionPerRoute) throws NoSuchAlgorithmException {
super(createBatchHttpRequestFactory());
List<HttpMessageConverter<?>> messageConverters= new ArrayList<HttpMessageConverter<?>>();
messageConverters.addAll(getMessageConverters());
messageConverters.add(0,new StringHttpMessageConverter(Charset.forName("UTF-8")));
setMessageConverters(messageConverters);
}
private static ClientHttpRequestFactory createBatchHttpRequestFactory() throws NoSuchAlgorithmException {
CloseableHttpClient httpClient;
HttpComponentsClientHttpRequestFactory httpRequestFactory;
SSLConnectionSocketFactory socketFactory;
socketFactory = new SSLConnectionSocketFactory(
SSLContext.getDefault(),
new String[] {"TLSv1"},
null,
SSLConnectionSocketFactory.ALLOW_ALL_HOSTNAME_VERIFIER);
Registry<ConnectionSocketFactory> socketFactoryRegistry = RegistryBuilder.<ConnectionSocketFactory>create()
.register("http", PlainConnectionSocketFactory.getSocketFactory())
.register("https", socketFactory)
.build();
PoolingHttpClientConnectionManager cm = new PoolingHttpClientConnectionManager(socketFactoryRegistry);
cm.setMaxTotal(250);
cm.setDefaultMaxPerRoute(100);
cm.closeExpiredConnections();
RequestConfig requestConfig = RequestConfig.custom().setConnectTimeout(30000)
.setConnectionRequestTimeout(30000).setSocketTimeout(30000).build();
httpClient = HttpClients.custom().setDefaultRequestConfig(requestConfig).setConnectionManager(cm).build();
httpRequestFactory = new HttpComponentsClientHttpRequestFactory(httpClient);
return httpRequestFactory;
}
}
响应拦截器
public class ResponseInterceptor {
private HTTPInputProperties inputProperties;
private static final Logger LOGGER = LoggerFactory.getLogger(ResponseInterceptor.class);
/**
* Intercepts the errorMessage from the API response and sends appropriate
* information to the Output channel.
*
* @param errorMessage
* @return Message
*/
public Message<String> transform(Message<MessagingException> errorMessage) {
LOGGER.error("Inside Response Interceptor !");
Message<String> responseMessage = null;
try {
if (null != errorMessage && null != errorMessage.getPayload()
&& null != errorMessage.getPayload().getCause()) {
LOGGER.error("Cause is - " + errorMessage.getPayload().getCause().getMessage());
if (errorMessage.getPayload().getCause() instanceof HttpClientErrorException) {
HttpClientErrorException clientError = (HttpClientErrorException) errorMessage.getPayload()
.getCause();
LOGGER.error("Error in ResponseInceptor", clientError);
List<String> errorCodeList = getErrorCodes(inputProperties.getNonRetryErrorCodes());
// intercept Only those errors that are defined as
// nonRetryErrorCodes options in stream definition
if (null != clientError.getStatusCode()
&& errorCodeList.contains(clientError.getStatusCode().toString())) {
LOGGER.error("Error in Response Body", clientError.getResponseBodyAsString());
LOGGER.debug("Non retry message found. Sending to output channel without retrying");
responseMessage = MessageBuilder.withPayload((null == clientError.getResponseBodyAsString() || clientError.getResponseBodyAsString().isEmpty())
? getDefaultPayload(clientError.getStatusCode().toString()) : clientError.getResponseBodyAsString())
.setHeader(BatchHttpClientConstants.HTTP_STATUS, clientError.getStatusCode().toString())
.setHeader(BatchHttpClientConstants.REQUEST_OBJECT,
getFailedMessagePayload(errorMessage))
.copyHeaders(errorMessage.getPayload().getFailedMessage().getHeaders())
.setReplyChannelName(BatchHttpClientConstants.OUTPUT).setErrorChannelName(null).build();
} else {
LOGGER.debug("Status code from API is not present in the nonRetryCodes");
}
} else if (errorMessage.getPayload().getCause() instanceof HttpServerErrorException) {
LOGGER.error("Error is Instance of HttpServerErrorException");
HttpServerErrorException serverError = (HttpServerErrorException) errorMessage.getPayload()
.getCause();
responseMessage = MessageBuilder
.withPayload((null == serverError.getResponseBodyAsString()
|| serverError.getResponseBodyAsString().isEmpty())
? getDefaultPayload(serverError.getStatusCode().toString())
: serverError.getResponseBodyAsString())
.setHeader(BatchHttpClientConstants.HTTP_STATUS, serverError.getStatusCode().toString())
.setHeader(BatchHttpClientConstants.REQUEST_OBJECT, getFailedMessagePayload(errorMessage))
.copyHeaders(errorMessage.getPayload().getFailedMessage().getHeaders())
.setReplyChannelName(BatchHttpClientConstants.OUTPUT).setErrorChannelName(null).build();
}
}
} catch (Exception exception) {
LOGGER.error("Exception occured while transforming errorResponse", exception);
}
// returning null will send the message back to previous module
return responseMessage;
}
private String getDefaultPayload(String httpStatusCode) {
JSONObject jsonResponse = new JSONObject();
if (BatchHttpClientConstants.INTERNAL_SERVER_ERROR.equalsIgnoreCase(httpStatusCode)) {
jsonResponse.put(BatchHttpClientConstants.ID, BatchHttpClientConstants.INTERNAL_SERVER_ERROR_SUBCODE);
jsonResponse.put(BatchHttpClientConstants.TEXT, "Internal Server Error");
} else if (BatchHttpClientConstants.RESOURCE_NOT_FOUND.equalsIgnoreCase(httpStatusCode)) {
jsonResponse.put(BatchHttpClientConstants.ID, BatchHttpClientConstants.RESOURCE_NOT_FOUND_SUBCODE);
jsonResponse.put(BatchHttpClientConstants.TEXT, "Empty Response From the API");
}else{
jsonResponse.put(BatchHttpClientConstants.ID, BatchHttpClientConstants.GENERIC_ERROR_SUBCODE);
jsonResponse.put(BatchHttpClientConstants.TEXT, "Generic Error Occured.");
}
return jsonResponse.toString();
}
/**
* Get Individual error codes using delimiter
*
* @param nonRetryErrorCodes
* @return List of Error Codes as string
*/
private List<String> getErrorCodes(String nonRetryErrorCodes) {
List<String> errorCodeList = new ArrayList<String>();
StringTokenizer st = new StringTokenizer(nonRetryErrorCodes, BatchHttpClientConstants.DELIMITER);
while (st.hasMoreElements()) {
errorCodeList.add(st.nextToken());
}
return errorCodeList;
}
/**
* returns failed Message Payload
*
* @param errorMessage
* @return String
* @throws UnsupportedEncodingException
*/
private byte[] getFailedMessagePayload(Message<MessagingException> errorMessage)
throws UnsupportedEncodingException {
if (null != errorMessage.getPayload().getFailedMessage()
&& null != errorMessage.getPayload().getFailedMessage().getPayload()) {
return errorMessage.getPayload().getFailedMessage().getPayload().toString()
.getBytes(BatchHttpClientConstants.UTF_8);
}
return "".getBytes(BatchHttpClientConstants.UTF_8);
}
public HTTPInputProperties getInputProperties() {
return inputProperties;
}
public void setInputProperties(HTTPInputProperties inputProperties) {
this.inputProperties = inputProperties;
}
}
你不能 - 你需要两个网关和一个路由器（使用 URL 路由到一个或另一个）。
你好，我有像 below.Where 这样的流定义，我从 s3 中逐行提取文件并调用 http 客户端并命名为 channel.My transport is rabbit 和
stream aws-s3|custom processor| custom-http-client --url1=https://test1.com --url2=https://test2.com --filterAttribute=messageAttribute --httpMethod=POST --nonRetryErrorCodes=400,401,404,500 --charset=UTF-8 --replyTimeout=30000 --mapHeaders=Api-Key,Content-Type --requestTimeOut=30000 |processor> queue:testQueue
我的 http-config 如下所示，并使用 apache http 客户端进行连接池和多线程我将所有非常错误（如套接字超时）放回 DLQ 并重试。
现在我的 uri1 使用 Oauth，uri2 使用基本的 restTemplate。我如何向我的 http-outbound 注入两个 rest 模板？一个是oauth，另一个是basicTemplate？
<beans:beans xmlns="http://www.springframework.org/schema/integration"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:beans="http://www.springframework.org/schema/beans"
xmlns:int-http="http://www.springframework.org/schema/integration/http"
xmlns:context="http://www.springframework.org/schema/context"
xsi:schemaLocation="http://www.springframework.org/schema/beans
http://www.springframework.org/schema/beans/spring-beans.xsd
http://www.springframework.org/schema/integration
http://www.springframework.org/schema/integration/spring-integration.xsd
http://www.springframework.org/schema/integration/http
http://www.springframework.org/schema/integration/http/spring-integration-http.xsd
http://www.springframework.org/schema/context
http://www.springframework.org/schema/context/spring-context-4.0.xsd">
<!-- <context:property-placeholder location="${xd.module.config.location}\processor${xd.module.name}\batch-http.properties"
ignore-resource-not-found="true" local-override="true"/> -->
<context:property-placeholder />
<!-- logger changes start -->
<channel-interceptor pattern="*" order="3">
<beans:bean class="org.springframework.integration.channel.interceptor.WireTap">
<beans:constructor-arg ref="loggingChannel" />
</beans:bean>
</channel-interceptor>
<logging-channel-adapter id="loggingChannel" log-full-message="true" level="ERROR"/>
<!-- logger changes end -->
<header-filter input-channel="input"
output-channel="inputX" header-names="x-death"/>
<service-activator input-channel="inputX" ref="gw" />
<gateway id="gw" default-request-channel="toHttp" default-reply-timeout="0" error-channel="errors" />
<beans:bean id="inputfields" class="test.HTTPInputProperties">
<beans:property name="nonRetryErrorCodes" value="${nonRetryErrorCodes}"/>
</beans:bean>
<beans:bean id="responseInterceptor" class="test.ResponseInterceptor">
<beans:property name="inputProperties" ref="inputfields" />
</beans:bean>
<chain input-channel="errors" output-channel="output">
<!-- examine payload.cause (http status code etc) and decide whether
to throw an exception or return the status code for sending to output -->
<header-filter header-names="replyChannel, errorChannel" />
<transformer ref="responseInterceptor" />
</chain>
<int-http:outbound-gateway id='batch-http' header-mapper="headerMapper"
request-channel='toHttp'
rest-template="batchRestTemplate"
url-expression="payload.contains('${filterAttribute}') ? '${url1}' : '${url2}'" http-method="${httpMethod}"
expected-response-type='java.lang.String' charset='${charset}'
reply-timeout='${replyTimeout}' reply-channel='output'>
</int-http:outbound-gateway>
<beans:bean id="batchHTTPConverter" class="org.springframework.http.converter.StringHttpMessageConverter" >
<beans:constructor-arg index="0" value="${charset}"/>
<beans:property name="supportedMediaTypes" value = "application/json;UTF-8" />
</beans:bean>
<beans:bean id="batchRestTemplate" class="testBatchRestTemplate" >
<beans:constructor-arg name="requestTimeOut" value="${requestTimeOut}"/>
<beans:constructor-arg name="maxConnectionPerRoute" value="${maxConnectionPerRoute}"/>
<beans:constructor-arg name="totalMaxConnections" ref="${totalMaxConnections}"/>
</beans:bean>
<beans:bean id="headerMapper" class="org.springframework.integration.http.support.DefaultHttpHeaderMapper"
factory-method="outboundMapper">
<beans:property name="outboundHeaderNames" value="${mapHeaders}"/>
<beans:property name="userDefinedHeaderPrefix" value=""/>
</beans:bean>
<channel id="output" />
<channel id="input" />
<channel id="inputX" />
<channel id="toHttp" />
</beans:beans>
public class BatchRestTemplate extends RestTemplate{
private static final Logger LOGGER = LoggerFactory
.getLogger(BatchRestTemplate.class);
private static Integer requestTimeOut;
private static Integer totalMaxConnections;
private static Integer maxConnectionPerRoute;
public BatchRestTemplate(Integer requestTimeOut,Integer totalMaxConnections,Integer maxConnectionPerRoute) throws NoSuchAlgorithmException {
super(createBatchHttpRequestFactory());
List<HttpMessageConverter<?>> messageConverters= new ArrayList<HttpMessageConverter<?>>();
messageConverters.addAll(getMessageConverters());
messageConverters.add(0,new StringHttpMessageConverter(Charset.forName("UTF-8")));
setMessageConverters(messageConverters);
}
private static ClientHttpRequestFactory createBatchHttpRequestFactory() throws NoSuchAlgorithmException {
CloseableHttpClient httpClient;
HttpComponentsClientHttpRequestFactory httpRequestFactory;
SSLConnectionSocketFactory socketFactory;
socketFactory = new SSLConnectionSocketFactory(
SSLContext.getDefault(),
new String[] {"TLSv1"},
null,
SSLConnectionSocketFactory.ALLOW_ALL_HOSTNAME_VERIFIER);
Registry<ConnectionSocketFactory> socketFactoryRegistry = RegistryBuilder.<ConnectionSocketFactory>create()
.register("http", PlainConnectionSocketFactory.getSocketFactory())
.register("https", socketFactory)
.build();
PoolingHttpClientConnectionManager cm = new PoolingHttpClientConnectionManager(socketFactoryRegistry);
cm.setMaxTotal(250);
cm.setDefaultMaxPerRoute(100);
cm.closeExpiredConnections();
RequestConfig requestConfig = RequestConfig.custom().setConnectTimeout(30000)
.setConnectionRequestTimeout(30000).setSocketTimeout(30000).build();
httpClient = HttpClients.custom().setDefaultRequestConfig(requestConfig).setConnectionManager(cm).build();
httpRequestFactory = new HttpComponentsClientHttpRequestFactory(httpClient);
return httpRequestFactory;
}
}
响应拦截器
public class ResponseInterceptor {
private HTTPInputProperties inputProperties;
private static final Logger LOGGER = LoggerFactory.getLogger(ResponseInterceptor.class);
/**
* Intercepts the errorMessage from the API response and sends appropriate
* information to the Output channel.
*
* @param errorMessage
* @return Message
*/
public Message<String> transform(Message<MessagingException> errorMessage) {
LOGGER.error("Inside Response Interceptor !");
Message<String> responseMessage = null;
try {
if (null != errorMessage && null != errorMessage.getPayload()
&& null != errorMessage.getPayload().getCause()) {
LOGGER.error("Cause is - " + errorMessage.getPayload().getCause().getMessage());
if (errorMessage.getPayload().getCause() instanceof HttpClientErrorException) {
HttpClientErrorException clientError = (HttpClientErrorException) errorMessage.getPayload()
.getCause();
LOGGER.error("Error in ResponseInceptor", clientError);
List<String> errorCodeList = getErrorCodes(inputProperties.getNonRetryErrorCodes());
// intercept Only those errors that are defined as
// nonRetryErrorCodes options in stream definition
if (null != clientError.getStatusCode()
&& errorCodeList.contains(clientError.getStatusCode().toString())) {
LOGGER.error("Error in Response Body", clientError.getResponseBodyAsString());
LOGGER.debug("Non retry message found. Sending to output channel without retrying");
responseMessage = MessageBuilder.withPayload((null == clientError.getResponseBodyAsString() || clientError.getResponseBodyAsString().isEmpty())
? getDefaultPayload(clientError.getStatusCode().toString()) : clientError.getResponseBodyAsString())
.setHeader(BatchHttpClientConstants.HTTP_STATUS, clientError.getStatusCode().toString())
.setHeader(BatchHttpClientConstants.REQUEST_OBJECT,
getFailedMessagePayload(errorMessage))
.copyHeaders(errorMessage.getPayload().getFailedMessage().getHeaders())
.setReplyChannelName(BatchHttpClientConstants.OUTPUT).setErrorChannelName(null).build();
} else {
LOGGER.debug("Status code from API is not present in the nonRetryCodes");
}
} else if (errorMessage.getPayload().getCause() instanceof HttpServerErrorException) {
LOGGER.error("Error is Instance of HttpServerErrorException");
HttpServerErrorException serverError = (HttpServerErrorException) errorMessage.getPayload()
.getCause();
responseMessage = MessageBuilder
.withPayload((null == serverError.getResponseBodyAsString()
|| serverError.getResponseBodyAsString().isEmpty())
? getDefaultPayload(serverError.getStatusCode().toString())
: serverError.getResponseBodyAsString())
.setHeader(BatchHttpClientConstants.HTTP_STATUS, serverError.getStatusCode().toString())
.setHeader(BatchHttpClientConstants.REQUEST_OBJECT, getFailedMessagePayload(errorMessage))
.copyHeaders(errorMessage.getPayload().getFailedMessage().getHeaders())
.setReplyChannelName(BatchHttpClientConstants.OUTPUT).setErrorChannelName(null).build();
}
}
} catch (Exception exception) {
LOGGER.error("Exception occured while transforming errorResponse", exception);
}
// returning null will send the message back to previous module
return responseMessage;
}
private String getDefaultPayload(String httpStatusCode) {
JSONObject jsonResponse = new JSONObject();
if (BatchHttpClientConstants.INTERNAL_SERVER_ERROR.equalsIgnoreCase(httpStatusCode)) {
jsonResponse.put(BatchHttpClientConstants.ID, BatchHttpClientConstants.INTERNAL_SERVER_ERROR_SUBCODE);
jsonResponse.put(BatchHttpClientConstants.TEXT, "Internal Server Error");
} else if (BatchHttpClientConstants.RESOURCE_NOT_FOUND.equalsIgnoreCase(httpStatusCode)) {
jsonResponse.put(BatchHttpClientConstants.ID, BatchHttpClientConstants.RESOURCE_NOT_FOUND_SUBCODE);
jsonResponse.put(BatchHttpClientConstants.TEXT, "Empty Response From the API");
}else{
jsonResponse.put(BatchHttpClientConstants.ID, BatchHttpClientConstants.GENERIC_ERROR_SUBCODE);
jsonResponse.put(BatchHttpClientConstants.TEXT, "Generic Error Occured.");
}
return jsonResponse.toString();
}
/**
* Get Individual error codes using delimiter
*
* @param nonRetryErrorCodes
* @return List of Error Codes as string
*/
private List<String> getErrorCodes(String nonRetryErrorCodes) {
List<String> errorCodeList = new ArrayList<String>();
StringTokenizer st = new StringTokenizer(nonRetryErrorCodes, BatchHttpClientConstants.DELIMITER);
while (st.hasMoreElements()) {
errorCodeList.add(st.nextToken());
}
return errorCodeList;
}
/**
* returns failed Message Payload
*
* @param errorMessage
* @return String
* @throws UnsupportedEncodingException
*/
private byte[] getFailedMessagePayload(Message<MessagingException> errorMessage)
throws UnsupportedEncodingException {
if (null != errorMessage.getPayload().getFailedMessage()
&& null != errorMessage.getPayload().getFailedMessage().getPayload()) {
return errorMessage.getPayload().getFailedMessage().getPayload().toString()
.getBytes(BatchHttpClientConstants.UTF_8);
}
return "".getBytes(BatchHttpClientConstants.UTF_8);
}
public HTTPInputProperties getInputProperties() {
return inputProperties;
}
public void setInputProperties(HTTPInputProperties inputProperties) {
this.inputProperties = inputProperties;
}
}
你不能 - 你需要两个网关和一个路由器（使用 URL 路由到一个或另一个）。