title: Spring-Integration transaction commit before rest of pipeline
body: <p>So, this is my current setup:</p>
  
  <pre><code>&lt;int-amqp:inbound-channel-adapter channel="input-channel" queue-names="probni" message-converter="jsonMessageConverter"
                                    channel-transacted="true"
                                    transaction-manager="dataSourceTransactionManager"/&gt;
  &lt;int:chain input-channel="input-channel" output-channel="inputc1"&gt;
      &lt;int:service-activator ref="h1" method="handle" /&gt;
      &lt;int:service-activator ref="h2" method="handle" /&gt;
      &lt;int:service-activator ref="h3" method="handle" /&gt;
      &lt;int:splitter  /&gt;
  &lt;/int:chain&gt;
  
  &lt;int:publish-subscribe-channel id="inputc1"/&gt;
  
  &lt;int:claim-check-in input-channel="inputc1" output-channel="nullChannel" message-store="messageStore" order="1" /&gt;
  
  &lt;int:bridge input-channel="inputc1" output-channel="inputc2" order="2" /&gt;
  
  &lt;int:publish-subscribe-channel id="inputc2" task-executor="taskExecutor" /&gt;
  
  &lt;int-amqp:outbound-channel-adapter channel="inputc2" exchange-name="exch" amqp-template="rabbitTemplate" order="1" /&gt;
  &lt;int:service-activator input-channel="inputc2" output-channel="nullChannel"
                         expression="@messageStore.removeMessage(headers['id'])" order="2" /&gt;
  </code></pre>
  
  <p>and the image of this is:</p>
  
  <p><a href="https://i.stack.imgur.com/cRQ1O.png" rel="nofollow noreferrer"><img src="https://i.stack.imgur.com/cRQ1O.png" alt="enter image description here"></a></p>
  
  <p>What I would need is that transaction (transaction-manager="dataSourceTransactionManager" which should be part in pink) is committed completely before any message is forwarded to int:brdige or from int:bridge (the blue arrow - so bridge is border component that should actually forward messages only if jdbc has committed).</p>
  
  <p>Thanks!</p>
  
  <p><strong>UPDATE</strong></p>
  
  <p>Here is description why I need this kind of setup:</p>
  
  <p>USE CASE: receive amqp message, process it and save to db and forward resulting amqp message down the pipeline. message should not be lost (neither arriving nor outgoing message e.g. on power loss etc.).
  Multiple messages that form a sequence can arrive in multiple different processes that has same organization like below.</p>
  
  <p>How I wanted to solve it:</p>
  
  <p>THREAD1:</p>
  
  <ol>
  <li>Receive Message from AMQP and start transaction</li>
  <li>Process AMQP message in series of service-activators (possibly each activator will write to db but changes will not be saved until commit)</li>
  <li>Write outgoing AMQP message to database (serialize it fully) - will not be saved until commit</li>
  <li>COMMIT</li>
  <li>ACK incoming message to AMQP</li>
  <li>Forward outgoing message to THREAD2 in-process (not DB pointer, but real message)</li>
  </ol>
  
  <p>THREAD2:</p>
  
  <ol>
  <li>Receive message in-process from THREAD1</li>
  <li>Try to send message to AMQP</li>
  <li>If send successful, delete outgoing AMQP message from DB that is saved in THREAD1 step 3.</li>
  </ol>
  
  <p>THREAD3:</p>
  
  <ol>
  <li>Poll DB for messages to be sent (from THREAD1 step 3) (each 10 sec)</li>
  <li>If any new message found, flag it for sending in next poll (in the meantime THREAD2 can delete this messages)</li>
  <li>If on second poll (after 20sec) message is still there - means THREAD2 failed to send it, so here we spawn new thread that will do the same task as THREAD2.</li>
  </ol>
  
  <p><strong>UPDATE 2</strong></p>
  
  <p>Tried this setup, but have some problems:</p>
  
  <pre><code>&lt;int:transaction-synchronization-factory id="transactionSynchronizationFactory"&gt;
      &lt;int:after-commit expression="payload" channel="committed-channel" /&gt;
  &lt;/int:transaction-synchronization-factory&gt;
  
  &lt;int-amqp:inbound-channel-adapter channel="input-channel" queue-names="probni" message-converter="jsonMessageConverter"
                                    channel-transacted="true"
                                    transaction-manager="dataSourceTransactionManager" advice-chain="amqpMethodInterceptor"/&gt;
  </code></pre>
  
  <p>and:</p>
  
  <pre><code>@Component
  public class AmqpMethodInterceptor implements MethodInterceptor {
  
      private TransactionSynchronizationFactory factory;
  
      public AmqpMethodInterceptor(TransactionSynchronizationFactory factory){
          this.factory = factory;
      }
  
      @Override
      public Object invoke(MethodInvocation invocation) throws Throwable {
  
  
          if (TransactionSynchronizationManager.isActualTransactionActive()) {
              TransactionSynchronization synchronization = factory.create("123");
              TransactionSynchronizationManager.registerSynchronization(synchronization);
          }
  
          Object result = invocation.proceed();
  
          return result;
      }
  }
  </code></pre>
  
  <p>after-commit is called, but the message is null at this point in time, so seems like I have nothing to forward to committed-channel. Any idea how to do this part?</p>