如何使消息流（僞）事務的任意部分？

Question

我想通過syslog入站通道適配器消息事務性啓動一個流程。不幸的是，適配器沒有采用可以作爲事務處理的輪詢器（許多示例中使用其他入站適配器的典型方法）。

有什麼解決方法嗎？

編輯

經過一番思考，我意識到，我的意圖是比最初描述的（因此標題的變化）有一點不同。基本上我只想做一些簡單而直接的方法，讓我的消息流僞事務性的任意部分（從流中的某個任意通道開始）。含義 - 如果流程完成時沒有任何異常，我想執行一些自定義代碼（但請注意，我不希望我的自定義僞提交代碼成爲流本身的一部分（步驟））。如果發生任何異常，我想執行一些自定義代碼。

使用TransactionSynchronizationFactory的語義將很適合我。

<int:transaction-synchronization-factory id="syncFactory"> 
    <int:after-commit expression="payload.renameTo('/success/' + payload.name)" channel="committedChannel" /> 
    <int:after-rollback expression="payload.renameTo('/failed/' + payload.name)" channel="rolledBackChannel" /> 
</int:transaction-synchronization-factory> 

唯一的問題是如何將它與其餘的流連接在一起。我所嘗試的是定義中間虛擬服務激活端點，該中間虛擬服務激活端點從我希望事務開始的通道接收消息。然後將事務性輪詢器添加到該服務激活器。但是這種方法有其自身的問題，因爲爲了使用輪詢器，你必須將傳入通道定義爲一個隊列通道，這似乎在單獨的線程中執行流（或者至少我觀察到一些異步行爲）。

Answer 1

從消息驅動的適配器的任何流可以在事務的範圍內通過使適配器的信道開始一個事務運行：

<tx:advice id="txAdvice" transaction-manager="transactionManager"> 
    <tx:attributes> 
     <tx:method name="send"/> 
    </tx:attributes> 
</tx:advice> 

<aop:config> 
    <aop:advisor advice-ref="txAdvice" pointcut="bean(fromSyslog)"/> 
</aop:config> 

<int:channel id="fromSyslog" /> 

信道（和所有下游信道）必須是DirectChannel（沒有隊列通道或任務執行器）。異步切換到另一個線程後，事務將提交。

我相信你明白，但是爲了其他讀者的利益，這並不會使系統日誌適配器本身事務性，只是下游流。

編輯

另一種方法是使用一箇中等流動交易網關...

@Transactional 
public interface TxGate { 

    void oneWay(Message<?> message); 

} 

<int:service-activator input-channel="fromSyslog" ref="txGate" /> 

<int:gateway id="txGate" service-interface="foo.TxGate" 
    default-request-channel="txSyslog" error-channel="foo" /> 

這樣的話，你可以處理的事務範圍內的異常，並決定是否不提交（從錯誤流拋出異常將回滾）。

無效回報很重要;因爲下游流量沒有回覆。

EDIT2

在回答您的問題編輯。

因此，您提出的解決方案似乎與您提出的問題（特別是在中流事務處理網關中）只允許您在出現錯誤時採取某些措施，而您也希望在成功後採取一些（不同的）措施。

有兩種方法可以做到這一點。

1. 使子流程中的最後一個通道a publish-subscribe-channel;添加第二個消費者（使用order明確定義它們的調用順序）並對第二個訂閱者執行「成功」操作 - 在發生異常（默認情況下）後不會調用它，以便繼續處理網關錯誤通道上的例外情況。
2. 使用ChannelInterceptor。在preSend()開始交易。子流完成後將調用afterSendCompletion()方法。如果子流引發異常，則會填充Exception參數的存在（或不存在）。

如果你想考慮contributing it，我們可以考慮在框架中加入這樣一個攔截器。