我发誓我有这个工作，但几个月后我可以恢复它（并升级到 Boot 1.5.9），我遇到了问题。
我设置了一个 JdbcPollingChannelAdapter，我可以很好地执行接收（），但是当我将适配器放入一个只对适配器的结果进行排队的流中时，在队列上运行 .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;
}
}