Распространение транзакций в весеннем интеграционном тесте с JPA и JMS

Question

Пожалуйста, помогите мне понять следующее.

У меня есть пружинный интеграционный тест, который я пытаюсь проверить методом класса ProcessCommentsDao:

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = {"classpath:testContext.xml"})
@Transactional()
public class ParseCommentsTest {

  @Resource
  private ProcessCommentsDao processCommentsDao;

  @Test
  public void testJMS() throws Exception {

     // Test data creation
     .........................

     processCommentsDao.parseComments();
   }
 }

В методе parseComments(), Я получаю список сущностей, затем каждая сущность обрабатывается с помощью реализации Spring JMS MessageListener:

@Service
public class ProcessCommentsDaoImpl extends BaseDao implements IProcessCommentsDao {

    private static final int PARSE_COMMENTS_COUNT_LIMIT = 100;
    @Autowired
    private JmsTemplate jmsTemplate;
    @Autowired
    private Queue parseCommentsDestination;

    @Override
    public void parseComments() {

      List<Comment> comments = attributeDao.findTopUnparsedComments(PARSE_COMMENTS_COUNT_LIMIT);

      for (Comment comment : comments) {
        jmsTemplate.convertAndSend(parseCommentsDestination, comment);
      }
    }
}

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

@Component
public class QueueListener implements MessageListener {

  @PersistenceContext
  private  EntityManager em;

  @Transactional()
  public void onMessage(final Message message) {
      try {
         if (message instanceof ObjectMessage) {
            final ObjectMessage objMessage = (ObjectMessage) message;
            Comment comment = (Comment) objMessage.getObject();

            //...Some logic ...

             comment = em.find(Comment.class, comment.getId());
             comment.setPosStatus(ParsingType.PROCESSED);
             em.merge(comment);

              //...Some logic ...

    } catch (final JMSException e) {
        e.printStackTrace();
    }
}

}

В результате метод em.find (Comment.class, comment.getId ()) возвращает значение null, поскольку данные были созданы в другом потоке, а текущий поток ничего не знает об этой транзакции.Есть ли способ настроить распространение транзакции так, чтобы метод MessageListener видел объекты, созданные в основном потоке, в котором выполняется метод теста?

Answer 1

Я нашел следующее решение моей проблемы. Тестовые данные генерируются в отдельной транзакции, которая создается вручную:

@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(locations = {"classpath:testContext.xml"})
@Transactional()
public class ParseCommentsTest {

  @Resource
  private ProcessCommentsDao processCommentsDao;
  @Autowired
  private PlatformTransactionManager transactionManager;

  @Before
  public void tearUp() {
    createTestData();
  }

  @Test
  @Rollback(false)
  public void testJMS() throws Exception {
   processCommentsDao.parseComments();
  }

  @After
  public void tearDown() {
   removeTestData();
 }

 private void createTestData() {
    TransactionTemplate txTemplate = new TransactionTemplate(transactionManager);
    txTemplate.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
    txTemplate.execute(new TransactionCallback<Object>() {

        @Override
        public Object doInTransaction(TransactionStatus status) {
            try {
            // Test data creation
            ...........................
        }
    });
  }
}

В методе ProcessCommentsDaoImpl.parseComments () реализовано ожидание завершения всех асинхронных JMS-запросов. Основной поток завершил свою работу, пока не будут обработаны все объекты:

@Service
public class ProcessCommentsDaoImpl extends BaseDao implements IProcessCommentsDao {

  private static final int PARSE_COMMENTS_COUNT_LIMIT = 100;
  @Autowired
  private JmsTemplate jmsTemplate;
  @Autowired
  private Queue parseCommentsDestination;

  @Override
  public void parseComments() {

    List<Comment> comments =    attributeDao.findTopUnparsedComments(PARSE_COMMENTS_COUNT_LIMIT);

    for (Comment comment : comments) {
     jmsTemplate.convertAndSend(parseCommentsDestination, comment);
    }
    // Wait all request procesed
    waitParseCommentsProcessed(comments);
   }

  @Override
  @Transactional(rollbackFor = Exception.class, propagation = Propagation.REQUIRES_NEW)
  public void parseComment(Long commentId) {
     ......................
     //Some logic
    ....................
    }
}

И рефакторинг MessageListener следующим образом:

public class ParseCommentQueueListener {

  private static Logger log = Logger.getLogger(ParseCommentQueueListener.class);

  @Resource(name = SpringContext.DAO_PROCESS_COMMENTS)
  private IProcessCommentsDao processCommentsDao;

  public Object receive(ObjectMessage message) {
    try {
        Long id = (Long) message.getObject();
        processCommentsDao.parseComment(id);
    } catch (JMSException ex) {
        log.error(ex.toString());
    }
    return message;
  }
} 

XML-конфигурация ParseCommentQueueListener выглядит следующим образом:

<bean id="messageListener" class="org.springframework.jms.listener.adapter.MessageListenerAdapter">
    <constructor-arg>
        <bean class="com.provoxlabs.wordminer.parsing.ParseCommentQueueListener"/>
    </constructor-arg>
    <property name="defaultListenerMethod" value="receive"/>
    <property name="defaultResponseDestination" ref="parseCommentsStatusDestination"/>
    <!-- we don't want automatic message context extraction -->
    <property name="messageConverter">
        <null/>
    </property>
</bean>

Answer 2

В результате метод em.find (Comment.class, comment.getId ()) возвращает значение null, поскольку данные были созданы в другом потоке, а текущий поток ничего не знает об этой транзакции.

Точнее, прослушиватель сообщений запускается в отдельной транзакции и не видит данных, созданных ParseCommentsTest.testJMS, потому что этот метод не зафиксировал.

Более тоговажно, ваш тест написан неправильно.Он имеет условие состязания: вызовы jmsTemplate.convertAndSend() являются асинхронными, поэтому логика в QueueListener.messageListener() может быть вызвана после завершения метода теста (и откатывает сделанные изменения).Этот тест может давать разные результаты при каждом запуске.

Ваш код также не легко тестируется.Рассмотрите возможность извлечения логики обработки из метода onMessage() в POJO и протестируйте ее отдельно.