diff --git a/README.md b/README.md index 21bd7500..d74a0f4b 100644 --- a/README.md +++ b/README.md @@ -117,6 +117,7 @@ This category targets developers and architects who have a good understanding of * **cafe-scripted** - Scripted implementation of the classic **cafe** sample application. Supports **JavaScript**, **Groovy**, **Ruby**, and **Python**. * **loan-broker** - Simulates a simple banking application (Uses **Gateway**, **Chain**, **Header Enricher**, **Recipient List Router**, **Aggregator**) see [Appendix A of the reference documentation](http://docs.spring.io/spring-integration/reference/html/samples.html) for more details * **loanshark** This extension to the loan broker sample shows how to exchange messages between Spring Integration applications (and other technologies) using **UDP**. + **file-split-ftp** - Reads a file; splits into 3 based on contents; sends files over ftp; sends email with results. #Contributing diff --git a/applications/file-split-ftp/.gitignore b/applications/file-split-ftp/.gitignore new file mode 100644 index 00000000..ce3767e5 --- /dev/null +++ b/applications/file-split-ftp/.gitignore @@ -0,0 +1,7 @@ +/target/ +.classpath +.project +.settings +.factorypath +.mvn +mvnw* diff --git a/applications/file-split-ftp/README.adoc b/applications/file-split-ftp/README.adoc new file mode 100644 index 00000000..e70160fc --- /dev/null +++ b/applications/file-split-ftp/README.adoc @@ -0,0 +1,21 @@ +:imagesdir: ./images + += File Split and FTP Sample + +- Looks for files `*.txt` in `/tmp/in` +- Reads file and writes to 3 files based on account (first 4 bytes in file) +- FileSplitter -> router [lines -> `lines`], [file marks -> `marks`] +- `marks` filters on EOF marker (drops SOF) +- pubsub [flush files], [ftp (x3)], [email result] +- Exceptions go to `tfrErrors` - sends failure email +- input file is renamed, based on final dispostion + +Test cases are provided for happy path and failure scenarios + +Run `ApplicationTests` as a JUnit test from your IDE. + += Flow Visualization + +image::flo1.png[] + +image::flo2.png[] diff --git a/applications/file-split-ftp/images/flo1.png b/applications/file-split-ftp/images/flo1.png new file mode 100644 index 00000000..5643320f Binary files /dev/null and b/applications/file-split-ftp/images/flo1.png differ diff --git a/applications/file-split-ftp/images/flo2.png b/applications/file-split-ftp/images/flo2.png new file mode 100644 index 00000000..383e1d54 Binary files /dev/null and b/applications/file-split-ftp/images/flo2.png differ diff --git a/applications/file-split-ftp/pom.xml b/applications/file-split-ftp/pom.xml new file mode 100644 index 00000000..002e98f8 --- /dev/null +++ b/applications/file-split-ftp/pom.xml @@ -0,0 +1,151 @@ + + + 4.0.0 + + org.springframework.boot + spring-boot-starter-parent + 1.4.0.RELEASE + + org.springframework.integration.samples + file-split-ftp + 4.3.0.BUILD-SNAPSHOT + File Split FTP + File Split FTP + http://projects.spring.io/spring-integration + + SpringIO + https://spring.io + + + + The Apache Software License, Version 2.0 + http://www.apache.org/licenses/LICENSE-2.0.txt + repo + + + + + garyrussell + Gary Russell + grussell@pivotal.io + + project lead + + + + markfisher + Mark Fisher + mfisher@pivotal.io + + project founder and lead emeritus + + + + ghillert + Gunnar Hillert + ghillert@pivotal.io + + + abilan + Artem Bilan + abilan@pivotal.io + + + + scm:git:scm:git:git://github.com/spring-projects/spring-integration-samples.git + scm:git:scm:git:ssh://git@github.com:spring-projects/spring-integration-samples.git + https://github.com/spring-projects/spring-integration-samples + + + + org.springframework.boot + spring-boot-starter-web + compile + + + org.springframework.boot + spring-boot-starter-integration + compile + + + org.springframework.integration + spring-integration-ftp + 4.3.2.RELEASE + compile + + + org.springframework.integration + spring-integration-http + 4.3.2.RELEASE + compile + + + org.springframework.integration + spring-integration-mail + 4.3.2.RELEASE + compile + + + org.springframework.integration + spring-integration-test + 5.0.0.BUILD-SNAPSHOT + compile + + + javax.mail + javax.mail-api + 1.5.5 + compile + + + com.sun.mail + javax.mail + 1.5.5 + compile + + + junit + junit + 4.12 + test + + + org.hamcrest + hamcrest-all + 1.3 + test + + + org.mockito + mockito-core + 1.10.19 + test + + + org.springframework + spring-test + 4.3.2.RELEASE + test + + + org.springframework.boot + spring-boot-starter-test + test + + + + + repo.spring.io.milestone + Spring Framework Maven Milestone Repository + https://repo.spring.io/libs-milestone + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + + diff --git a/applications/file-split-ftp/src/main/java/org/springframework/integration/samples/filesplit/Application.java b/applications/file-split-ftp/src/main/java/org/springframework/integration/samples/filesplit/Application.java new file mode 100644 index 00000000..d7e93f79 --- /dev/null +++ b/applications/file-split-ftp/src/main/java/org/springframework/integration/samples/filesplit/Application.java @@ -0,0 +1,234 @@ +/* + * Copyright 2002-2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.samples.filesplit; + +import java.io.File; +import java.io.PrintWriter; +import java.io.StringWriter; + +import org.aopalliance.intercept.MethodInterceptor; +import org.apache.commons.net.ftp.FTPFile; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.boot.autoconfigure.mail.MailProperties; +import org.springframework.context.annotation.Bean; +import org.springframework.integration.dsl.IntegrationFlow; +import org.springframework.integration.dsl.IntegrationFlows; +import org.springframework.integration.dsl.core.Pollers; +import org.springframework.integration.dsl.file.Files; +import org.springframework.integration.dsl.mail.Mail; +import org.springframework.integration.file.FileHeaders; +import org.springframework.integration.file.FileWritingMessageHandler; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.file.splitter.FileSplitter; +import org.springframework.integration.file.support.FileExistsMode; +import org.springframework.integration.ftp.session.DefaultFtpSessionFactory; +import org.springframework.integration.http.config.EnableIntegrationGraphController; +import org.springframework.messaging.Message; +import org.springframework.messaging.MessageHeaders; +import org.springframework.messaging.MessagingException; +import org.springframework.web.servlet.config.annotation.CorsRegistry; +import org.springframework.web.servlet.config.annotation.WebMvcConfigurer; +import org.springframework.web.servlet.config.annotation.WebMvcConfigurerAdapter; + +@SpringBootApplication +@EnableIntegrationGraphController +public class Application { + + private static final String EMAIL_SUCCESS_SUFFIX = "emailSuccessSuffix"; + + public static void main(String[] args) { + SpringApplication.run(Application.class, args); + } + + @Autowired + private MailProperties mailProperties; + + /** + * Poll for files, add an error channel, split into lines route the start/end markers + * to {@link #markers()} and the lines to {@link #lines()}. + * @return the flow. + */ + @Bean + public IntegrationFlow fromFile() { + return IntegrationFlows.from(Files.inboundAdapter(new File("/tmp/in")) + .preventDuplicates(false) + .patternFilter("*.txt"), e -> e.poller(Pollers.fixedDelay(5000))) + .enrichHeaders(h -> h.header(MessageHeaders.ERROR_CHANNEL, "tfrErrors.input")) + .handle(Files.splitter(true, true)) + .>route(Object::getClass, m -> m + .channelMapping(FileSplitter.FileMarker.class, "markers.input") + .channelMapping(String.class, "lines.input")) + .get(); + } + + /** + * Process lines; append (no flush) to the appropriate file. + * @return the flow. + */ + @Bean + public IntegrationFlow lines() { + return f -> f.handle(fileOut()); + } + + @Bean + public FileWritingMessageHandler fileOut() { + return Files.outboundAdapter("'/tmp/out'") + .appendNewLine(true) + .fileNameExpression("payload.substring(1, 4) + '.txt'") + .fileExistsMode(FileExistsMode.APPEND_NO_FLUSH) // files remain open for efficiency + .get(); + } + + /** + * Process file markers; ignore START, when END, flush the files, ftp them and + * send an email. + * @return the flow. + */ + @Bean + public IntegrationFlow markers() { + return f -> f.filter(m -> m.getMark().equals(FileSplitter.FileMarker.Mark.END), + e -> e.id("markerFilter")) + .publishSubscribeChannel(s -> s + + // first trigger file flushes + .subscribe(sf -> sf.transform("'/tmp/out/.*\\.txt'", e -> e.id("toTriggerPattern")) + .handle("fileOut", "trigger", e -> e.id("flusher"))) + + // send the first file + .subscribe(sf -> sf.transform(p -> new File("/tmp/out/002.txt")) + .enrichHeaders(h -> h.header(FileHeaders.FILENAME, "002.txt", true)) + .handleWithAdapter(a -> a.ftp(ftp1()).remoteDirectory("foo"), e -> e.id("ftp002"))) + + // send the second file + .subscribe(sf -> sf.transform(p -> new File("/tmp/out/006.txt")) + .enrichHeaders(h -> h.header(FileHeaders.FILENAME, "006.txt", true)) + .handleWithAdapter(a -> a.ftp(ftp2()).remoteDirectory("foo"), e -> e.id("ftp006"))) + + // send the third file + .subscribe(sf -> sf.transform(p -> new File("/tmp/out/009.txt")) + .enrichHeaders(h -> h.header(FileHeaders.FILENAME, "009.txt", true)) + .handleWithAdapter(a -> a.ftp(ftp3()).remoteDirectory("foo"), e -> e.id("ftp009"))) + + // send an email + .subscribe(sf -> sf.transform(FileSplitter.FileMarker::getFilePath) + .enrichHeaders(Mail.headers() + .subject("File successfully split and transferred") + .from("foo@bar") + .toFunction(m -> new String[] {"bar@baz"})) + .enrichHeaders(h -> h.header(EMAIL_SUCCESS_SUFFIX, ".success")) + .channel("toMail.input"))); + } + + @Bean + public SessionFactory ftp1() { + DefaultFtpSessionFactory ftp = new DefaultFtpSessionFactory(); + ftp.setHost("host3"); + ftp.setUsername("user"); + ftp.setPassword("ftp"); + return ftp; + } + + @Bean + public SessionFactory ftp2() { + DefaultFtpSessionFactory ftp = new DefaultFtpSessionFactory(); + ftp.setHost("host3"); + ftp.setUsername("user"); + ftp.setPassword("ftp"); + return ftp; + } + + @Bean + public SessionFactory ftp3() { + DefaultFtpSessionFactory ftp = new DefaultFtpSessionFactory(); + ftp.setHost("host3"); + ftp.setUsername("user"); + ftp.setPassword("ftp"); + return ftp; + } + + /** + * Error flow - email failure + * @return the flow. + */ + @Bean + public IntegrationFlow tfrErrors() { + return f -> f.enrichHeaders(Mail.headers() + .subject("File split and transfer failed") + .from("foo@bar") + .toFunction(m -> new String[] {"bar@baz"})) + .enrichHeaders(h -> h.header(EMAIL_SUCCESS_SUFFIX, ".failed") + .headerExpression(FileHeaders.ORIGINAL_FILE, "payload.failedMessage.headers['" + + FileHeaders.ORIGINAL_FILE + "']")) + .transform(p -> + p.getFailedMessage().getPayload().toString() + "\n" + getStackTraceAsString(p)) + .channel("toMail.input"); + } + + @Bean + public IntegrationFlow toMail() { + return f -> f.handleWithAdapter(a -> a.mail(this.mailProperties.getHost()) +// .javaMailProperties(b -> b.put("mail.debug", "true")) + .port(this.mailProperties.getPort()) + .credentials(this.mailProperties.getUsername(), this.mailProperties.getPassword()), + e -> e.id("mailOut").advice(afterMailAdvice())); + } + + /** + * Rename the input file after success/failure. + * @return the flow. + */ + @Bean + public MethodInterceptor afterMailAdvice() { + return invocation -> { + Message message = (Message) invocation.getArguments()[0]; + MessageHeaders headers = message.getHeaders(); + File originalFile = headers.get(FileHeaders.ORIGINAL_FILE, File.class); + try { + invocation.proceed(); + originalFile.renameTo( + new File(originalFile.getAbsolutePath() + headers.get(EMAIL_SUCCESS_SUFFIX))); + } + catch (Exception e) { + originalFile.renameTo( + new File(originalFile.getAbsolutePath() + headers.get(EMAIL_SUCCESS_SUFFIX) + ".email.failed")); + } + return null; + }; + } + + // Integration Graph CORS + @Bean + public WebMvcConfigurer corsConfigurer() { + return new WebMvcConfigurerAdapter() { + @Override + public void addCorsMappings(CorsRegistry registry) { + registry.addMapping("/integration").allowedOrigins("http://localhost:8082"); + } + }; + } + + private String getStackTraceAsString(Throwable cause) { + StringWriter stringWriter = new StringWriter(); + PrintWriter printWriter = new PrintWriter(stringWriter, true); + cause.printStackTrace(printWriter); + return stringWriter.getBuffer().toString(); + } + +} diff --git a/applications/file-split-ftp/src/main/resources/application.properties b/applications/file-split-ftp/src/main/resources/application.properties new file mode 100644 index 00000000..330425fd --- /dev/null +++ b/applications/file-split-ftp/src/main/resources/application.properties @@ -0,0 +1,4 @@ +spring.mail.host=localhost +spring.mail.port=25 +spring.mail.username=user +spring.mail.password=pw diff --git a/applications/file-split-ftp/src/main/resources/logback.xml b/applications/file-split-ftp/src/main/resources/logback.xml new file mode 100644 index 00000000..bb023166 --- /dev/null +++ b/applications/file-split-ftp/src/main/resources/logback.xml @@ -0,0 +1,17 @@ + + + + + + %d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n + + + + + + + + + + diff --git a/applications/file-split-ftp/src/test/java/org/springframework/integration/samples/filesplit/ApplicationTests.java b/applications/file-split-ftp/src/test/java/org/springframework/integration/samples/filesplit/ApplicationTests.java new file mode 100644 index 00000000..3e781ca0 --- /dev/null +++ b/applications/file-split-ftp/src/test/java/org/springframework/integration/samples/filesplit/ApplicationTests.java @@ -0,0 +1,208 @@ +/* + * Copyright 2002-2016 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.springframework.integration.samples.filesplit; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.BDDMockito.given; +import static org.mockito.BDDMockito.willThrow; +import static org.mockito.Matchers.any; +import static org.mockito.Matchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.reset; +import static org.mockito.Mockito.verify; + +import java.io.BufferedReader; +import java.io.File; +import java.io.FileOutputStream; +import java.io.FileReader; +import java.io.IOException; +import java.io.InputStream; +import java.util.List; + +import org.apache.commons.net.ftp.FTPFile; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; +import org.junit.runner.RunWith; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.annotation.Import; +import org.springframework.integration.file.remote.session.Session; +import org.springframework.integration.file.remote.session.SessionFactory; +import org.springframework.integration.test.mail.TestMailServer; +import org.springframework.integration.test.mail.TestMailServer.SmtpServer; +import org.springframework.integration.test.util.TestUtils; +import org.springframework.test.annotation.DirtiesContext; +import org.springframework.test.context.junit4.SpringRunner; + +@RunWith(SpringRunner.class) +@SpringBootTest +@DirtiesContext(classMode = DirtiesContext.ClassMode.AFTER_EACH_TEST_METHOD) +public class ApplicationTests { + + private static final SmtpServer smtpServer = TestMailServer.smtp(0); + + @Autowired + private Session session; + + @BeforeClass + public static void setup() { + // Configure the boot property to send email to the test email server. + System.setProperty("spring.mail.port", Integer.toString(smtpServer.getPort())); + new File("/tmp/in/foo.txt").delete(); + new File("/tmp/in/foo.txt.success").delete(); + new File("/tmp/in/foo.txt.failed").delete(); + } + + @Before + public void beforeTest() { + smtpServer.getMessages().clear(); + } + + @SuppressWarnings("unchecked") + @Test + public void testSuccess() throws Exception { + reset(this.session); + String message = runTest(false); + assertThat(message).contains("File successfully split and transferred"); + assertThat(message).contains(TestUtils.applySystemFileSeparator("/tmp/in/foo.txt")); + } + + @Test + public void testFailure() throws Exception { + willThrow(new RuntimeException("fail test exception")) + .given(this.session).write(any(InputStream.class), eq("foo/002.txt.writing")); + String message = runTest(true); + assertThat(message).contains("File split and transfer failed"); + assertThat(message).contains("fail test exception"); + assertThat(message).contains(TestUtils.applySystemFileSeparator("/tmp/out/002.txt")); + } + + /* + * Create a test file containing one row per account. + * Verify the three files appear, with the correct contents. + * Verify the input file was renamed based on success/failure. + * Verify the email was sent. + */ + private String runTest(boolean fail) throws Exception { + File in = new File("/tmp/in/", "foo"); + FileOutputStream fos = new FileOutputStream(in); + fos.write("*002,foo,bar\n*006,baz,qux\n*009,fiz,buz\n".getBytes()); + fos.close(); + in.renameTo(new File("/tmp/in/", "foo.txt")); + File out = new File("/tmp/out/002.txt"); + int n = 0; + while(n++ < 100 && !out.exists() && out.length() < 12) { + Thread.sleep(100); + } + assertThat(out.exists()).isTrue(); + BufferedReader br = new BufferedReader(new FileReader(out)); + assertThat(br.readLine()).isEqualTo("*002,foo,bar"); + br.close(); + out.delete(); + out = new File("/tmp/out/006.txt"); + assertThat(out.exists()).isTrue(); + br = new BufferedReader(new FileReader(out)); + assertThat(br.readLine()).isEqualTo("*006,baz,qux"); + br.close(); + out.delete(); + out = new File("/tmp/out/009.txt"); + assertThat(out.exists()).isTrue(); + br = new BufferedReader(new FileReader(out)); + assertThat(br.readLine()).isEqualTo("*009,fiz,buz"); + br.close(); + out.delete(); + if (!fail) { + in = new File("/tmp/in/", "foo.txt.success"); + } + else { + in = new File("/tmp/in/", "foo.txt.failed"); + } + n = 0; + while(n++ < 100 && !in.exists()) { + Thread.sleep(100); + } + assertThat(in.exists()).isTrue(); + in.delete(); + // verify FTP + verify(this.session).write(any(InputStream.class), eq("foo/002.txt.writing")); + if (!fail) { + verify(this.session).write(any(InputStream.class), eq("foo/006.txt.writing")); + verify(this.session).write(any(InputStream.class), eq("foo/009.txt.writing")); + verify(this.session).rename("foo/002.txt.writing", "foo/002.txt"); + verify(this.session).rename("foo/006.txt.writing", "foo/006.txt"); + verify(this.session).rename("foo/009.txt.writing", "foo/009.txt"); + } + + String message = verifyMail(); + assertThat(message).contains("From: foo@bar"); + assertThat(message).contains("To: bar@baz"); + return message; + } + + public String verifyMail() throws Exception { + List messages = smtpServer.getMessages(); + int n = 0; + while (n++ < 100 && messages.size() < 1) { + Thread.sleep(100); + } + assertThat(messages).hasSize(1); + return messages.get(0); + } + + /** + * Overrides the ftp session factories with mocks. + * + */ + @Configuration + @Import(Application.class) + public static class Config { + + @Bean + public SessionFactory ftp1() throws IOException { + return mockSf(); + } + + @Bean + public SessionFactory ftp2() throws IOException { + return mockSf(); + } + + @Bean + public SessionFactory ftp3() throws IOException { + return mockSf(); + } + + private SessionFactory mockSf() throws IOException { + @SuppressWarnings("unchecked") + SessionFactory mocksf = mock(SessionFactory.class); + given(mocksf.getSession()).willReturn(mockSession()); + return mocksf; + } + + @SuppressWarnings("unchecked") + @Bean + public Session mockSession() throws IOException { + return mock(Session.class); + } + + } + +} diff --git a/build.gradle b/build.gradle index b6a660fb..a7ec078d 100644 --- a/build.gradle +++ b/build.gradle @@ -189,7 +189,7 @@ subprojects { subproject -> junitVersion = '4.12' jythonVersion = '2.5.3' log4jVersion = '1.2.17' - mockitoVersion = '1.9.5' + mockitoVersion = '1.10.19' openJpaVersion = '2.4.0' oracleDriverVersion = '11.2.0.3' reactorVersion = '2.0.8.RELEASE' @@ -1371,6 +1371,37 @@ project('kafka-dsl') { } } +project('file-split-ftp') { + description = 'File Split FTP' + + apply plugin: 'spring-boot' + + dependencies { + compile 'org.springframework.boot:spring-boot-starter-web' + compile 'org.springframework.boot:spring-boot-starter-integration' + compile "org.springframework.integration:spring-integration-core:$springIntegrationVersion" + compile "org.springframework.integration:spring-integration-ftp:$springIntegrationVersion" + compile "org.springframework.integration:spring-integration-http:$springIntegrationVersion" + compile "org.springframework.integration:spring-integration-mail:$springIntegrationVersion" + compile "org.springframework.integration:spring-integration-java-dsl:$springIntegrationDslVersion" + compile "org.springframework.integration:spring-integration-test:5.0.0.BUILD-SNAPSHOT" // test mail server + compile "javax.mail:javax.mail-api:$javaxMailVersion" + compile "com.sun.mail:javax.mail:$javaxMailVersion" + + testCompile 'org.springframework.boot:spring-boot-starter-test' + } + + springBoot { + mainClass = 'org.springframework.integration.samples.filesplit.Application' + } + + task run(type: JavaExec) { + main 'org.springframework.integration.samples.filesplit.Application' + classpath = sourceSets.main.runtimeClasspath + } +} + + sonarqube { properties { property "sonar.jacoco.reportPath", "${buildDir.name}/jacoco.exec"