如何将直接通道的消息负载复制到网关中的恢复通道？

Question

我有一个“数据通道”，它使用入站通道适配器从数据库获取结果集。 在这里，我从 DB 中获取了一个名为“process_id”的字段。通过 int:http-outbound gateway 调用外部系统后，我正在定义一个恢复通道。我只想对该 process_id 进行更新查询。但我无法在恢复通道中获取进程 ID。获取异常无效属性“payload [process_id]”..无论如何要将 process_id 传递给恢复通道，这样我就可以像这样执行我的更新查询

int-jdbc:outbound-channel-adapter query="更新 TBL_RECEIPT 设置receipt_status=1 其中 process_id in (:payload[process_id])" data-source="dataSource" channel="errors"/>

为清楚起见，下面是spring-integration xml配置

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
	xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:int="http://www.springframework.org/schema/integration"
	xmlns:jdbc="http://www.springframework.org/schema/jdbc" xmlns:int-jdbc="http://www.springframework.org/schema/integration/jdbc"
	xmlns:context="http://www.springframework.org/schema/context" xmlns:p="http://www.springframework.org/schema/p"
	xmlns:task="http://www.springframework.org/schema/task" xmlns:int-http="http://www.springframework.org/schema/integration/http"
	xmlns:stream="http://www.springframework.org/schema/integration/stream"
	xsi:schemaLocation="http://www.springframework.org/schema/beans
            http://www.springframework.org/schema/beans/spring-beans.xsd
              http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-4.1.xsd
            http://www.springframework.org/schema/integration
            http://www.springframework.org/schema/integration/spring-integration-4.1.xsd
            http://www.springframework.org/schema/integration/stream 
		http://www.springframework.org/schema/integration/stream/spring-integration-stream-4.1.xsd
               http://www.springframework.org/schema/integration/http
    http://www.springframework.org/schema/integration/http/spring-integration-http-4.1.xsd
            http://www.springframework.org/schema/integration/jdbc
            http://www.springframework.org/schema/integration/jdbc/spring-integration-jdbc-4.1.xsd
            http://www.springframework.org/schema/jdbc
            http://www.springframework.org/schema/jdbc/spring-jdbc-4.1.xsd">

	<int:channel id="requestchannel"></int:channel>
	<int:channel id="xtifyrequestchannel"></int:channel>
	<int:channel id="xtifyresponsechannel"></int:channel>
	<int:channel id="tpgrequestchannel"></int:channel>
	<int:channel id="tpgresponsechannel"></int:channel>
   <int:channel id="xtifyerrorchannel">
	</int:channel>

	<int:channel id="tpgerrorchannel">
	</int:channel>

	<int:channel id="executerchannel">
		<int:dispatcher task-executor="taskExecutor" />
	</int:channel>

	<task:executor id="taskExecutor" pool-size="2" />

	<bean id="pollerdatamapper" class="main.java.com.as.poller.PollerDataMapper" />

	<bean id="pollerservice" class="main.java.com.as.poller.PollerService" />

	<bean id="requestFactory"
		class="org.springframework.http.client.SimpleClientHttpRequestFactory">
		<property name="connectTimeout" value="10000" />
		<property name="readTimeout" value="10000" />
	</bean>

	<int:logging-channel-adapter id="logger"
		level="INFO" />

	<int-jdbc:inbound-channel-adapter id="datachannel"
		query="select loyalty_id,process_id,mobile_uid,mobile_os from TBL_RECEIPT where r_cre_time=(select min(r_cre_time) from TBL_RECEIPT where receipt_status=0)"
		data-source="dataSource" max-rows-per-poll="1" row-mapper="pollerdatamapper">

		<int:poller fixed-rate="5000">
		</int:poller>

	</int-jdbc:inbound-channel-adapter>


	<int:gateway id="requestGateway" service-interface="main.java.com.as.poller.RequestGateway"
		default-request-channel="requestchannel" default-reply-timeout="20000">
		<int:method name="pushNotification" />
		<int:method name="sendTPGRequest" request-channel="tpgrequestchannel">
			<int:header name="Content-Type" value="multipart/form-data" />
		</int:method>
	</int:gateway>

	<int:object-to-json-transformer
		input-channel="requestchannel" output-channel="xtifyrequestchannel"></int:object-to-json-transformer>

	<int-http:outbound-gateway id="xtifygateway"
		request-channel="xtifyrequestchannel" reply-channel="xtifyresponsechannel" request-factory="requestFactory"
		url="${xtifyUrl}" http-method="POST">
		<int-http:request-handler-advice-chain>
			<int:retry-advice max-attempts="3" recovery-channel="xtifyerrorchannel">
			</int:retry-advice>
		</int-http:request-handler-advice-chain>
	</int-http:outbound-gateway>

	<int-http:outbound-gateway id="tpggateway"
		request-channel="tpgrequestchannel" reply-channel="tpgresponsechannel"
		request-factory="requestFactory" expected-response-type="java.lang.String"
		url="${tpg_url}" http-method="POST">
		<int-http:request-handler-advice-chain>
			<int:retry-advice max-attempts="3" recovery-channel="tpgerrorchannel">
			</int:retry-advice>
		</int-http:request-handler-advice-chain>
	</int-http:outbound-gateway>

	<int:json-to-object-transformer
		input-channel="tpgresponsechannel" type="main.java.com.as.rest.response.TPGResponse" />


	<int:service-activator input-channel="datachannel"
		output-channel="executerchannel" ref="pollerservice" method="getRecordFromPoller">
	</int:service-activator>

	<int:service-activator input-channel="executerchannel"
		ref="pollerservice" method="getDataFromExecuterChannel">
	</int:service-activator>

	<int-jdbc:outbound-channel-adapter
		id="tpgsystemfailure"
		query="update TBL_RECEIPT set receipt_status=1 
		where process_id in (:payload.failedMessage.payload[process_id])"
		data-source="dataSource" channel="tpgerrorchannel" />

	<int-jdbc:outbound-channel-adapter
		id="xtifysystemfailure"
		query="update TBL_RECEIPT set receipt_status=4 where process_id in (:payload.failedMessage.payload[process_id])"
		data-source="dataSource" channel="xtifyerrorchannel" />
		
		<int-jdbc:outbound-channel-adapter
		id="xtifysystemsuccess"
		query="update TBL_RECEIPT set receipt_status=5 where process_id in (:payload.process_id)"
		data-source="dataSource" channel="xtifyresponsechannel" />
		
		
</beans>
	
 

Answer 1

recovery-channel 得到一个 ErrorMessage。有效负载是 MessagingException，具有两个属性 failedMessage 和 cause。

使用payload.failedMessage.payload[process_id]。