Клянусь, у меня это сработало, но когда я смогу вернуться к нему через несколько месяцев (и после обновления до Boot 1.5.9), у меня возникнут проблемы.
Я установил JdbcPollingChannelAdapter, на котором я могу выполнять функцию receive (), но когда я помещаю адаптер в поток, который не делает ничего, кроме очереди, результат адаптера, запуск .receive в очереди всегда возвращает нуль (я Однако в журнале консоли можно увидеть, что выполняется SQL адаптера).
Тесты ниже. Почему я могу получить результаты от адаптера, но не поставить их в очередь? Заранее благодарим вас за любую помощь.
@RunWith(SpringRunner.class)
@SpringBootTest
@AutoConfigureTestDatabase
@JdbcTest
public class JdbcpollingchanneladapterdemoTests {
@Autowired
@Qualifier("dataSource")
DataSource dataSource;
private static PollableChannel outputQueue;
@BeforeClass
public static void setupClass() {
outputQueue = MessageChannels.queue().get();
return;
}
@Test
public void Should_HaveQueue() {
assertThat(outputQueue, instanceOf(QueueChannel.class));
}
@Test
@Sql(executionPhase = ExecutionPhase.BEFORE_TEST_METHOD,
statements = "Create Table DEMO (CODE VARCHAR(5));")
@Sql(executionPhase = ExecutionPhase.AFTER_TEST_METHOD,
statements = "Drop Table DEMO ;")
public void Should_Not_HaveMessageOnTheQueue_When_No_DemosAreInTheDatabase() {
Message<?> message = outputQueue.receive(5000);
assertThat(message, nullValue()) ;
}
@Test
@Sql(executionPhase = ExecutionPhase.BEFORE_TEST_METHOD,
statements = "Create Table DEMO (CODE VARCHAR(5));")
@Sql(executionPhase = ExecutionPhase.BEFORE_TEST_METHOD,
statements = "Insert into DEMO (CODE) VALUES ('12345');")
@Sql(executionPhase = ExecutionPhase.AFTER_TEST_METHOD,
statements = "Drop Table DEMO ;")
public void Should_HaveMessageOnTheQueue_When_DemosIsInTheDatabase() {
assertThat(outputQueue, instanceOf(QueueChannel.class));
Message<?> message = outputQueue.receive(5000);
assertThat(message, notNullValue());
assertThat(message.getPayload().toString(), equalTo("15317")) ;
}
@Test
@Sql(executionPhase = ExecutionPhase.BEFORE_TEST_METHOD,
statements = "Create Table DEMO (CODE VARCHAR(5));")
@Sql(executionPhase = ExecutionPhase.BEFORE_TEST_METHOD,
statements = "Insert into DEMO (CODE) VALUES ('12345');")
@Sql(executionPhase = ExecutionPhase.AFTER_TEST_METHOD,
statements = "Drop Table DEMO ;")
public void get_message_directly_from_adapter() {
JdbcPollingChannelAdapter adapter =
new JdbcPollingChannelAdapter(dataSource, "SELECT CODE FROM DEMO");
adapter.setRowMapper(new DemoRowMapper());
adapter.setMaxRowsPerPoll(1);
Message<?> message = adapter.receive();
assertThat(message, notNullValue());
}
private static class Demo {
private String demo;
String getDemo() {
return demo;
}
void setDemo(String value) {
this.demo = value;
}
@Override
public String toString() {
return "Demo [value=" + this.demo + "]";
}
}
public static class DemoRowMapper implements RowMapper<Demo> {
@Override
public Demo mapRow(ResultSet rs, int rowNum) throws SQLException {
Demo demo = new Demo();
demo.setDemo(rs.getString("CODE"));
return demo;
}
}
@Component
public static class MyFlowAdapter extends IntegrationFlowAdapter {
@Autowired
@Qualifier("dataSource")
DataSource dataSource;
@Override
protected IntegrationFlowDefinition<?> buildFlow() {
JdbcPollingChannelAdapter adapter =
new JdbcPollingChannelAdapter(dataSource, "SELECT CODE FROM DEMO");
adapter.setRowMapper(new DemoRowMapper());
adapter.setMaxRowsPerPoll(1);
return from(adapter,
c -> c.poller(Pollers.fixedRate(1000L, 2000L)
.maxMessagesPerPoll(1)
.get()))
.channel(outputQueue);
}
}
}
ИЗМЕНИТЬ. Я упростил его, насколько смог, рефакторинг кода ниже. Тест проходит поток с общим источником сообщения и терпит неудачу в потоке с источником сообщения JdbcPollingChannelAdapter. Мне просто не очевидно, как мне настроить второй источник сообщений, чтобы он работал так же, как первый источник сообщений.
@Test
@Sql(executionPhase = ExecutionPhase.BEFORE_TEST_METHOD,
statements = "Create Table DEMO (CODE VARCHAR(5));")
@Sql(executionPhase = ExecutionPhase.BEFORE_TEST_METHOD,
statements = "Insert into DEMO (CODE) VALUES ('12345');")
public void Should_HaveMessageOnTheQueue_When_UnsentDemosIsInTheDatabase() {
this.genericFlowContext.registration(new GenericFlowAdapter()).register();
PollableChannel genericChannel = this.beanFactory.getBean("GenericFlowAdapterOutput",
PollableChannel.class);
this.jdbcPollingFlowContext.registration(new JdbcPollingFlowAdapter()).register();
PollableChannel jdbcPollingChannel = this.beanFactory.getBean("JdbcPollingFlowAdapterOutput",
PollableChannel.class);
assertThat(genericChannel.receive(5000).getPayload(), equalTo("15317"));
assertThat(jdbcPollingChannel.receive(5000).getPayload(), equalTo("15317"));
}
private static class GenericFlowAdapter extends IntegrationFlowAdapter {
@Override
protected IntegrationFlowDefinition<?> buildFlow() {
return from(getObjectMessageSource(),
e -> e.poller(Pollers.fixedRate(100)))
.channel(c -> c.queue("GenericFlowAdapterOutput"));
}
private MessageSource<Object> getObjectMessageSource() {
return () -> new GenericMessage<>("15317");
}
}
private static class JdbcPollingFlowAdapter extends IntegrationFlowAdapter {
@Autowired
@Qualifier("dataSource")
DataSource dataSource;
@Override
protected IntegrationFlowDefinition<?> buildFlow() {
return from(getObjectMessageSource(),
e -> e.poller(Pollers.fixedRate(100)))
.channel(c -> c.queue("JdbcPollingFlowAdapterOutput"));
}
private MessageSource<Object> getObjectMessageSource() {
JdbcPollingChannelAdapter adapter =
new JdbcPollingChannelAdapter(dataSource, "SELECT CODE FROM DEMO");
adapter.setRowMapper(new DemoRowMapper());
adapter.setMaxRowsPerPoll(1);
return adapter;
}
}