Merge branch '2.0.x' into 2.1.x
Closes gh-17078
This commit is contained in:
@@ -54,25 +54,22 @@ public class SampleIntegrationApplication {
|
||||
|
||||
@Bean
|
||||
public FileWritingMessageHandler fileWriter() {
|
||||
FileWritingMessageHandler writer = new FileWritingMessageHandler(
|
||||
new File("target/output"));
|
||||
FileWritingMessageHandler writer = new FileWritingMessageHandler(new File("target/output"));
|
||||
writer.setExpectReply(false);
|
||||
return writer;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public IntegrationFlow integrationFlow(SampleEndpoint endpoint) {
|
||||
return IntegrationFlows.from(fileReader(), new FixedRatePoller())
|
||||
.channel(inputChannel()).handle(endpoint).channel(outputChannel())
|
||||
.handle(fileWriter()).get();
|
||||
return IntegrationFlows.from(fileReader(), new FixedRatePoller()).channel(inputChannel()).handle(endpoint)
|
||||
.channel(outputChannel()).handle(fileWriter()).get();
|
||||
}
|
||||
|
||||
public static void main(String[] args) {
|
||||
SpringApplication.run(SampleIntegrationApplication.class, args);
|
||||
}
|
||||
|
||||
private static class FixedRatePoller
|
||||
implements Consumer<SourcePollingChannelAdapterSpec> {
|
||||
private static class FixedRatePoller implements Consumer<SourcePollingChannelAdapterSpec> {
|
||||
|
||||
@Override
|
||||
public void accept(SourcePollingChannelAdapterSpec spec) {
|
||||
|
||||
@@ -79,42 +79,37 @@ public class SampleIntegrationApplicationTests {
|
||||
|
||||
@Test
|
||||
public void testMessageGateway() throws Exception {
|
||||
this.context = SpringApplication.run(SampleIntegrationApplication.class,
|
||||
"testviamg");
|
||||
this.context = SpringApplication.run(SampleIntegrationApplication.class, "testviamg");
|
||||
String output = getOutput();
|
||||
assertThat(output).contains("testviamg");
|
||||
}
|
||||
|
||||
private String getOutput() throws Exception {
|
||||
Future<String> future = Executors.newSingleThreadExecutor()
|
||||
.submit(new Callable<String>() {
|
||||
@Override
|
||||
public String call() throws Exception {
|
||||
Resource[] resources = getResourcesWithContent();
|
||||
while (resources.length == 0) {
|
||||
Thread.sleep(200);
|
||||
resources = getResourcesWithContent();
|
||||
}
|
||||
StringBuilder builder = new StringBuilder();
|
||||
for (Resource resource : resources) {
|
||||
try (InputStream inputStream = resource.getInputStream()) {
|
||||
builder.append(new String(
|
||||
StreamUtils.copyToByteArray(inputStream)));
|
||||
}
|
||||
}
|
||||
return builder.toString();
|
||||
Future<String> future = Executors.newSingleThreadExecutor().submit(new Callable<String>() {
|
||||
@Override
|
||||
public String call() throws Exception {
|
||||
Resource[] resources = getResourcesWithContent();
|
||||
while (resources.length == 0) {
|
||||
Thread.sleep(200);
|
||||
resources = getResourcesWithContent();
|
||||
}
|
||||
StringBuilder builder = new StringBuilder();
|
||||
for (Resource resource : resources) {
|
||||
try (InputStream inputStream = resource.getInputStream()) {
|
||||
builder.append(new String(StreamUtils.copyToByteArray(inputStream)));
|
||||
}
|
||||
});
|
||||
}
|
||||
return builder.toString();
|
||||
}
|
||||
});
|
||||
return future.get(30, TimeUnit.SECONDS);
|
||||
}
|
||||
|
||||
private Resource[] getResourcesWithContent() throws IOException {
|
||||
Resource[] candidates = ResourcePatternUtils
|
||||
.getResourcePatternResolver(new DefaultResourceLoader())
|
||||
Resource[] candidates = ResourcePatternUtils.getResourcePatternResolver(new DefaultResourceLoader())
|
||||
.getResources("file:target/output/**");
|
||||
for (Resource candidate : candidates) {
|
||||
if ((candidate.getFilename() != null
|
||||
&& candidate.getFilename().endsWith(".writing"))
|
||||
if ((candidate.getFilename() != null && candidate.getFilename().endsWith(".writing"))
|
||||
|| candidate.contentLength() == 0) {
|
||||
return new Resource[0];
|
||||
}
|
||||
|
||||
@@ -30,8 +30,7 @@ public class ProducerApplication implements CommandLineRunner {
|
||||
public void run(String... args) throws Exception {
|
||||
new File("target/input").mkdirs();
|
||||
if (args.length > 0) {
|
||||
FileOutputStream stream = new FileOutputStream(
|
||||
"target/input/data" + System.currentTimeMillis() + ".txt");
|
||||
FileOutputStream stream = new FileOutputStream("target/input/data" + System.currentTimeMillis() + ".txt");
|
||||
for (String arg : args) {
|
||||
stream.write(arg.getBytes());
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user