MockIntegrationMessage 在缓存上下文时不起作用
MockIntegrationMessage not working when context cached
我正在使用 Spring Integration 5.1.7.RELEASE、Spring Boot 2.1.7.RELEASE 和 Spring Integration Test 5.0.11.RELEASE 进行一个项目。
作为我测试的一部分，我正在尝试 unit test 执行对不同 MessageHandlers 的调用的 IntegrationFlow，所以我 mocking 他们使用MockIntegrationContext 来自 Spring Integration Test.
我的代码在我们的 Jenkins 服务器上出现间歇性问题，这似乎与缓存的上下文有关。我有一个没有模拟的集成测试 (@SpringBootTest) 和一个专门针对 IntegrationFlow 的测试 class，其中模拟了 MessageHandlers。
如果之前运行了集成测试并且缓存了上下文，则 MockIntegrationContext 执行的设置似乎没有任何效果，底层的 service-activator 被调用，即使日志显示它已取消订阅，模拟订阅了该频道。
轮询器和句柄端点的定义如下：
@Configuration
public class FlowConfig {
@Bean
public IntegrationFlow fileFlow() {
return IntegrationFlows.from(
Files.inboundAdapter(new File(fileDir)).autoCreateDirectory(true).preventDuplicates(false),
e -> e.poller(Pollers.fixedDelay(pollingFrequency)).id("fileInboundSourceEndpoint").get())
...
.handle(fileParseHandler, e -> e.id("fileParseHandlerEndpoint"))
...
.get()
}
}
这是测试设置：
@RunWith(SpringRunner.class)
@ContextConfiguration(classes = FlowConfig.class)
@SpringIntegrationTest
@EnableIntegration
public class FlowConfigTest {
@Autowired
MockIntegrationContext mockIntegrationCtx;
@Test
public void test_fileFlow() throws IOException {
mockIntegrationCtx.substituteMessageHandlerFor("fileParseHandlerEndpoint",
MockIntegration.mockMessageHandler().handleNextAndReply(m -> m));
new File(fileDir + "test").createNewFile();
}
}
这是两组不同的日志，分别来自一次不成功和一次成功的测试执行：
2019-09-09 15:17:02.059 INFO 14462 --- [ main] o.s.i.endpoint.EventDrivenConsumer : Removing {service-activator:fileParseHandlerEndpoint} as a subscriber to the 'fileFlow.channel#2' channel
2019-09-09 15:17:02.059 INFO 14462 --- [ main] o.s.integration.channel.DirectChannel : Channel 'org.springframework.context.support.GenericApplicationContext@51f95f0d.fileFlow.channel#2' has 0 subscriber(s).
2019-09-09 15:17:02.059 INFO 14462 --- [ main] o.s.i.endpoint.EventDrivenConsumer : stopped fileParseHandlerEndpoint
2019-09-09 15:17:02.062 INFO 14462 --- [ main] o.s.i.endpoint.EventDrivenConsumer : Adding {message-handler:fileParseHandlerEndpoint} as a subscriber to the 'fileFlow.channel#2' channel
2019-09-09 15:17:02.062 INFO 14462 --- [ main] o.s.integration.channel.DirectChannel : Channel 'org.springframework.context.support.GenericApplicationContext@51f95f0d.fileFlow.channel#2' has 1 subscriber(s).
2019-09-09 15:17:02.062 INFO 14462 --- [ main] o.s.i.endpoint.EventDrivenConsumer : started fileParseHandlerEndpoint
...
2019-09-09 15:17:03.731 DEBUG 14462 --- [ scheduling-1] o.s.i.file.FileReadingMessageSource : Added to queue: [/tmp/file]
2019-09-09 15:17:03.732 DEBUG 14462 --- [ scheduling-1] o.s.i.e.SourcePollingChannelAdapter : Poll resulted in Message: GenericMessage [payload=/tmp/file, headers={file_originalFile=/tmp/file, id=4323b036-4edd-019e-2c61-6917c7970767, file_name=file, file_relativePath=file, timestamp=1568056623732}]
...
2019-09-09 15:17:03.732 DEBUG 14462 --- [ scheduling-1] o.s.integration.channel.DirectChannel : preSend on channel 'fileFlow.channel#2', message: GenericMessage [payload=/tmp/file, headers={file_originalFile=/tmp/file, errorChannel=error_channel, id=af997a89-40d6-9024-65fc-7ab3883c329b, file_name=file, file_relativePath=file, timestamp=1568056623732}]
2019-09-09 15:17:03.732 DEBUG 14462 --- [ scheduling-1] o.s.i.handler.ServiceActivatingHandler : ServiceActivator for [org.springframework.integration.handler.MethodInvokingMessageProcessor@7b4acdc2] (fileParseHandlerEndpoint) received message: GenericMessage [payload=/tmp/file, headers={file_originalFile=/tmp/file, errorChannel=error_channel, id=af997a89-40d6-9024-65fc-7ab3883c329b, file_name=file, file_relativePath=file, timestamp=1568056623732}]
[Fatal Error] file:1:1: Premature end of file.
13:41:05.126 [main] INFO o.s.i.endpoint.EventDrivenConsumer - Removing {service-activator:fileParseHandlerEndpoint} as a subscriber to the 'fileFlow.channel#2' channel
13:41:05.126 [main] INFO o.s.i.channel.DirectChannel - Channel 'org.springframework.context.support.GenericApplicationContext@6fb365ed.fileFlow.channel#2' has 0 subscriber(s).
13:41:05.126 [main] INFO o.s.i.endpoint.EventDrivenConsumer - stopped fileParseHandlerEndpoint
13:41:05.128 [main] INFO o.s.i.endpoint.EventDrivenConsumer - Adding {message-handler:fileParseHandlerEndpoint} as a subscriber to the 'fileFlow.channel#2' channel
13:41:05.128 [main] INFO o.s.i.channel.DirectChannel - Channel 'org.springframework.context.support.GenericApplicationContext@6fb365ed.fileFlow.channel#2' has 1 subscriber(s).
13:41:05.128 [main] INFO o.s.i.endpoint.EventDrivenConsumer - started fileParseHandlerEndpoint
...
13:41:06.775 [task-scheduler-1] DEBUG o.s.i.file.FileReadingMessageSource - Added to queue: [/tmp/file]
13:41:06.775 [task-scheduler-1] DEBUG o.s.i.e.SourcePollingChannelAdapter - Poll resulted in Message: GenericMessage [payload=/tmp/file, headers={file_originalFile=/tmp/file, id=3bb52d11-49ad-3ea8-eb67-7e6da721022a, file_name=file, file_relativePath=file, timestamp=1568050866775}]
...
13:41:06.788 [task-scheduler-1] DEBUG o.s.i.channel.DirectChannel - preSend on channel 'fileFlow.channel#2', message: GenericMessage [payload=/tmp/file, headers={file_originalFile=/tmp/file, errorChannel=error_channel, id=ec6b0213-8109-ae43-ac27-d17b1bbad8aa, file_name=file, file_relativePath=file, timestamp=1568050866777}]
13:41:06.789 [task-scheduler-1] DEBUG o.s.i.test.mock.MockMessageHandler - org.springframework.integration.test.mock.MockMessageHandler$MockitoMock34676946@76fb98a2 received message: GenericMessage [payload=/tmp/file, headers={file_originalFile=/tmp/file, errorChannel=error_channel, id=ec6b0213-8109-ae43-ac27-d17b1bbad8aa, file_name=file, file_relativePath=file, timestamp=1568050866777}]
我知道可能有 @DirtiesContext 的组合应该可以解决这个问题，但我仍在努力理解为什么 MockIntegrationContext 无法从缓存上下文。
有什么方法可以在不使用 @DirtiesContext 的情况下解决这个问题？
首先当你使用SpringBoot时，你需要依赖它的版本管理。
此外，当您按原样使用 Spring 集成时，请考虑使用 spring-integration-bom 进行导入。这样你就不需要为单个模块指定版本了。
我的观点是 spring-integration-core 和 spring-integration-test 有不同的版本。他们必须相同。否则你可能会在运行时搞得一团糟。
现在说正题。
我认为这不是缓存问题。
此测试套件的应用程序上下文启动与目标测试方法中的 mockIntegrationCtx.substituteMessageHandlerFor() 之间存在竞争条件。
问题出在您从 e.poller(Pollers.fixedDelay(pollingFrequency) 开始流程这一事实。这个由 TaskScheduler 在后台处理，并独立于测试套件的主线程生成消息。这就是您观察到意外行为的原因。
您可以通过 autoStartup 操作来解决这个问题。 @SpringIntegrationTest 为您准备了这个：
/**
* Specify a simple matching patterns ("xxx*", "*xxx", "*xxx*" or "xxx*yyy") for
* {@link org.springframework.integration.endpoint.AbstractEndpoint}
* bean names to mark them as {@code autoStartup = false}
* during context initialization.
* @return the endpoints name patterns to stop during context initialization
* @see IntegrationEndpointsInitializer
* @see org.springframework.util.PatternMatchUtils
*/
String[] noAutoStartup() default {};
所以，你这样做：
@SpringIntegrationTest(noAutoStartup = "fileInboundSourceEndpoint")
然后在测试中自动连接通道适配器：
@Autowired
私有 SourcePollingChannelAdapter fileInboundSourceEndpoint；
在 substituteMessageHandlerFor() 之后启动它：
this.fileInboundSourceEndpoint.start();
我正在使用 Spring Integration 5.1.7.RELEASE、Spring Boot 2.1.7.RELEASE 和 Spring Integration Test 5.0.11.RELEASE 进行一个项目。
作为我测试的一部分，我正在尝试 unit test 执行对不同 MessageHandlers 的调用的 IntegrationFlow，所以我 mocking 他们使用MockIntegrationContext 来自 Spring Integration Test.
我的代码在我们的 Jenkins 服务器上出现间歇性问题，这似乎与缓存的上下文有关。我有一个没有模拟的集成测试 (@SpringBootTest) 和一个专门针对 IntegrationFlow 的测试 class，其中模拟了 MessageHandlers。
如果之前运行了集成测试并且缓存了上下文，则 MockIntegrationContext 执行的设置似乎没有任何效果，底层的 service-activator 被调用，即使日志显示它已取消订阅，模拟订阅了该频道。
轮询器和句柄端点的定义如下：
@Configuration
public class FlowConfig {
@Bean
public IntegrationFlow fileFlow() {
return IntegrationFlows.from(
Files.inboundAdapter(new File(fileDir)).autoCreateDirectory(true).preventDuplicates(false),
e -> e.poller(Pollers.fixedDelay(pollingFrequency)).id("fileInboundSourceEndpoint").get())
...
.handle(fileParseHandler, e -> e.id("fileParseHandlerEndpoint"))
...
.get()
}
}
这是测试设置：
@RunWith(SpringRunner.class)
@ContextConfiguration(classes = FlowConfig.class)
@SpringIntegrationTest
@EnableIntegration
public class FlowConfigTest {
@Autowired
MockIntegrationContext mockIntegrationCtx;
@Test
public void test_fileFlow() throws IOException {
mockIntegrationCtx.substituteMessageHandlerFor("fileParseHandlerEndpoint",
MockIntegration.mockMessageHandler().handleNextAndReply(m -> m));
new File(fileDir + "test").createNewFile();
}
}
这是两组不同的日志，分别来自一次不成功和一次成功的测试执行：
2019-09-09 15:17:02.059 INFO 14462 --- [ main] o.s.i.endpoint.EventDrivenConsumer : Removing {service-activator:fileParseHandlerEndpoint} as a subscriber to the 'fileFlow.channel#2' channel
2019-09-09 15:17:02.059 INFO 14462 --- [ main] o.s.integration.channel.DirectChannel : Channel 'org.springframework.context.support.GenericApplicationContext@51f95f0d.fileFlow.channel#2' has 0 subscriber(s).
2019-09-09 15:17:02.059 INFO 14462 --- [ main] o.s.i.endpoint.EventDrivenConsumer : stopped fileParseHandlerEndpoint
2019-09-09 15:17:02.062 INFO 14462 --- [ main] o.s.i.endpoint.EventDrivenConsumer : Adding {message-handler:fileParseHandlerEndpoint} as a subscriber to the 'fileFlow.channel#2' channel
2019-09-09 15:17:02.062 INFO 14462 --- [ main] o.s.integration.channel.DirectChannel : Channel 'org.springframework.context.support.GenericApplicationContext@51f95f0d.fileFlow.channel#2' has 1 subscriber(s).
2019-09-09 15:17:02.062 INFO 14462 --- [ main] o.s.i.endpoint.EventDrivenConsumer : started fileParseHandlerEndpoint
...
2019-09-09 15:17:03.731 DEBUG 14462 --- [ scheduling-1] o.s.i.file.FileReadingMessageSource : Added to queue: [/tmp/file]
2019-09-09 15:17:03.732 DEBUG 14462 --- [ scheduling-1] o.s.i.e.SourcePollingChannelAdapter : Poll resulted in Message: GenericMessage [payload=/tmp/file, headers={file_originalFile=/tmp/file, id=4323b036-4edd-019e-2c61-6917c7970767, file_name=file, file_relativePath=file, timestamp=1568056623732}]
...
2019-09-09 15:17:03.732 DEBUG 14462 --- [ scheduling-1] o.s.integration.channel.DirectChannel : preSend on channel 'fileFlow.channel#2', message: GenericMessage [payload=/tmp/file, headers={file_originalFile=/tmp/file, errorChannel=error_channel, id=af997a89-40d6-9024-65fc-7ab3883c329b, file_name=file, file_relativePath=file, timestamp=1568056623732}]
2019-09-09 15:17:03.732 DEBUG 14462 --- [ scheduling-1] o.s.i.handler.ServiceActivatingHandler : ServiceActivator for [org.springframework.integration.handler.MethodInvokingMessageProcessor@7b4acdc2] (fileParseHandlerEndpoint) received message: GenericMessage [payload=/tmp/file, headers={file_originalFile=/tmp/file, errorChannel=error_channel, id=af997a89-40d6-9024-65fc-7ab3883c329b, file_name=file, file_relativePath=file, timestamp=1568056623732}]
[Fatal Error] file:1:1: Premature end of file.
13:41:05.126 [main] INFO o.s.i.endpoint.EventDrivenConsumer - Removing {service-activator:fileParseHandlerEndpoint} as a subscriber to the 'fileFlow.channel#2' channel
13:41:05.126 [main] INFO o.s.i.channel.DirectChannel - Channel 'org.springframework.context.support.GenericApplicationContext@6fb365ed.fileFlow.channel#2' has 0 subscriber(s).
13:41:05.126 [main] INFO o.s.i.endpoint.EventDrivenConsumer - stopped fileParseHandlerEndpoint
13:41:05.128 [main] INFO o.s.i.endpoint.EventDrivenConsumer - Adding {message-handler:fileParseHandlerEndpoint} as a subscriber to the 'fileFlow.channel#2' channel
13:41:05.128 [main] INFO o.s.i.channel.DirectChannel - Channel 'org.springframework.context.support.GenericApplicationContext@6fb365ed.fileFlow.channel#2' has 1 subscriber(s).
13:41:05.128 [main] INFO o.s.i.endpoint.EventDrivenConsumer - started fileParseHandlerEndpoint
...
13:41:06.775 [task-scheduler-1] DEBUG o.s.i.file.FileReadingMessageSource - Added to queue: [/tmp/file]
13:41:06.775 [task-scheduler-1] DEBUG o.s.i.e.SourcePollingChannelAdapter - Poll resulted in Message: GenericMessage [payload=/tmp/file, headers={file_originalFile=/tmp/file, id=3bb52d11-49ad-3ea8-eb67-7e6da721022a, file_name=file, file_relativePath=file, timestamp=1568050866775}]
...
13:41:06.788 [task-scheduler-1] DEBUG o.s.i.channel.DirectChannel - preSend on channel 'fileFlow.channel#2', message: GenericMessage [payload=/tmp/file, headers={file_originalFile=/tmp/file, errorChannel=error_channel, id=ec6b0213-8109-ae43-ac27-d17b1bbad8aa, file_name=file, file_relativePath=file, timestamp=1568050866777}]
13:41:06.789 [task-scheduler-1] DEBUG o.s.i.test.mock.MockMessageHandler - org.springframework.integration.test.mock.MockMessageHandler$MockitoMock34676946@76fb98a2 received message: GenericMessage [payload=/tmp/file, headers={file_originalFile=/tmp/file, errorChannel=error_channel, id=ec6b0213-8109-ae43-ac27-d17b1bbad8aa, file_name=file, file_relativePath=file, timestamp=1568050866777}]
我知道可能有 @DirtiesContext 的组合应该可以解决这个问题，但我仍在努力理解为什么 MockIntegrationContext 无法从缓存上下文。
有什么方法可以在不使用 @DirtiesContext 的情况下解决这个问题？
首先当你使用SpringBoot时，你需要依赖它的版本管理。
此外，当您按原样使用 Spring 集成时，请考虑使用 spring-integration-bom 进行导入。这样你就不需要为单个模块指定版本了。
我的观点是 spring-integration-core 和 spring-integration-test 有不同的版本。他们必须相同。否则你可能会在运行时搞得一团糟。
现在说正题。
我认为这不是缓存问题。
此测试套件的应用程序上下文启动与目标测试方法中的 mockIntegrationCtx.substituteMessageHandlerFor() 之间存在竞争条件。
问题出在您从 e.poller(Pollers.fixedDelay(pollingFrequency) 开始流程这一事实。这个由 TaskScheduler 在后台处理，并独立于测试套件的主线程生成消息。这就是您观察到意外行为的原因。
您可以通过 autoStartup 操作来解决这个问题。 @SpringIntegrationTest 为您准备了这个：
/**
* Specify a simple matching patterns ("xxx*", "*xxx", "*xxx*" or "xxx*yyy") for
* {@link org.springframework.integration.endpoint.AbstractEndpoint}
* bean names to mark them as {@code autoStartup = false}
* during context initialization.
* @return the endpoints name patterns to stop during context initialization
* @see IntegrationEndpointsInitializer
* @see org.springframework.util.PatternMatchUtils
*/
String[] noAutoStartup() default {};
所以，你这样做：
@SpringIntegrationTest(noAutoStartup = "fileInboundSourceEndpoint")
然后在测试中自动连接通道适配器：
@Autowired 私有 SourcePollingChannelAdapter fileInboundSourceEndpoint；
在 substituteMessageHandlerFor() 之后启动它：
this.fileInboundSourceEndpoint.start();