TransactionSynchronizationFactory в сочетании с JmsTransactionManager не работает

Question

У меня есть случай, когда я читаю файл, конвертирую содержимое в String. Затем разделите строку на несколько полезных нагрузок и отправьте эти полезные нагрузки в очередь. Я хочу использовать JmsTransactionManager, чтобы все сообщения были отправлены или вообще не были отправлены.

Когда TX успешно, я хочу переместить файл в папку Archive, иначе переместите его в папку Failed. Я прочитал, что я могу использовать transactionSynchronizationFactory для этого. Но в сочетании с JmsTransactionManager файл не перемещается. Если я использую PseudoTransactionManager, тогда файл перемещается, но я теряю свое JmsTransaction.

Я сделал упрощенную версию, чтобы воспроизвести проблему. (Содержание файла в этом случае представляет собой простой список разделенных запятыми значений.)

@Bean 
public IntegrationFlow fileInboundAdaptor() { 
    return IntegrationFlows 
      .from(s -> s.file(new File(INBOUND_PATH)) 
          .patternFilter("*.txt"), 
        e -> e.poller(Pollers.fixedDelay(5000) 
          .transactionSynchronizationFactory(transactionSynchronizationFactory()) 
          .transactional(new JmsTransactionManager(connectionFactory)) 

        ) 

      ) 

      .transform(Transformers.fileToString()) 
      .split(s -> s.applySequence(false).get().getT2().setDelimiters(",")) 
      .handle((GenericHandler<String>) (payload, headers) -> { 
       jmsTemplate.send("SOME_QUEUE", (Session session) -> session.createTextMessage(payload)); 
       return payload; 
      }) 
      .channel(MessageChannels.queue("fileReadingResultChannel")) 
      .get(); 
} 

transactionSynchronizationFactory выглядит следующим образом:

@Bean 
public TransactionSynchronizationFactory transactionSynchronizationFactory() { 
    ExpressionParser parser = new SpelExpressionParser(); 

    ExpressionEvaluatingTransactionSynchronizationProcessor syncProcessor 
      = new ExpressionEvaluatingTransactionSynchronizationProcessor(); 
    syncProcessor.setBeanFactory(applicationContext.getAutowireCapableBeanFactory()); 
    syncProcessor.setAfterCommitExpression(parser.parseExpression(
      "payload.renameTo(new java.io.File('test/archive' " + 
        " + T(java.io.File).separator + 'ARCHIVE-' + payload.name))")); 
    syncProcessor.setAfterRollbackExpression(parser.parseExpression(
      "payload.renameTo(new java.io.File('test/fail' " + 
        " + T(java.io.File).separator + 'FAILED-' + payload.name))")); 
    return new DefaultTransactionSynchronizationFactory(syncProcessor); 
} 

Так что мой вопрос: действительно TransactionSynchronizationFactory работает только с PseudoTransactionManager или должен работать с JmsTransactionManager?

Решение

мне нужно установить transactionSynchronization на JmsTransaction. Что-то вроде этого:

public JmsTransactionManager transactionManager() { 
    JmsTransactionManager jmsTransactionManager = new JmsTransactionManager(connectionFactory); 
    jmsTransactionManager.setTransactionSynchronization(AbstractPlatformTransactionManager.SYNCHRONIZATION_ON_ACTUAL_TRANSACTION); 
    return jmsTransactionManager; 
} 

Answer 1

Ну, я думаю, что ваш вопрос здесь:

/** 
* Create a new JmsTransactionManager for bean-style usage. 
* <p>Note: The ConnectionFactory has to be set before using the instance. 
* This constructor can be used to prepare a JmsTemplate via a BeanFactory, 
* typically setting the ConnectionFactory via setConnectionFactory. 
* <p>Turns off transaction synchronization by default, as this manager might 
* be used alongside a datastore-based Spring transaction manager like 
* DataSourceTransactionManager, which has stronger needs for synchronization. 
* Only one manager is allowed to drive synchronization at any point of time. 
* @see #setConnectionFactory 
* @see #setTransactionSynchronization 
*/ 
public JmsTransactionManager() { 
    setTransactionSynchronization(SYNCHRONIZATION_NEVER); 
} 

Таким образом, вы должны вручную переключаться на него в setTransactionSynchronization(AbstractPlatformTransactionManager.SYNCHRONIZATION_ALWAYS);