From 7b84810388f91dc7c0524100e80f4059ee0a88da Mon Sep 17 00:00:00 2001 From: Gary Russell Date: Thu, 13 Oct 2016 17:18:17 -0400 Subject: [PATCH] file-split-ftp Move errorChannel to Poller --- .../samples/filesplit/Application.java | 16 ++++++++++++---- .../integration/spring-integration-context.xml | 2 +- basic/amqp/src/main/resources/log4j.xml | 6 +++++- .../integration/samples/barrier/Application.java | 8 +++++--- .../src/main/resources/application.properties | 1 - .../src/main/webapp/WEB-INF/servlet-config.xml | 5 +++++ .../integration/samples/kafka/Application.java | 5 ++++- build.gradle | 5 +++-- 8 files changed, 35 insertions(+), 13 deletions(-) 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 index d7e93f79..203de59f 100644 --- 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 @@ -28,6 +28,7 @@ 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.channel.DirectChannel; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.dsl.core.Pollers; @@ -41,6 +42,7 @@ 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.MessageChannel; import org.springframework.messaging.MessageHeaders; import org.springframework.messaging.MessagingException; import org.springframework.web.servlet.config.annotation.CorsRegistry; @@ -69,8 +71,7 @@ public class Application { 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")) + .patternFilter("*.txt"), e -> e.poller(Pollers.fixedDelay(5000).errorChannel(tfrErrorChannel()))) .handle(Files.splitter(true, true)) .>route(Object::getClass, m -> m .channelMapping(FileSplitter.FileMarker.class, "markers.input") @@ -163,13 +164,19 @@ public class Application { return ftp; } + @Bean + public MessageChannel tfrErrorChannel() { + return new DirectChannel(); + } + /** * Error flow - email failure * @return the flow. */ @Bean public IntegrationFlow tfrErrors() { - return f -> f.enrichHeaders(Mail.headers() + return IntegrationFlows.from(tfrErrorChannel()) + .enrichHeaders(Mail.headers() .subject("File split and transfer failed") .from("foo@bar") .toFunction(m -> new String[] {"bar@baz"})) @@ -178,7 +185,8 @@ public class Application { + FileHeaders.ORIGINAL_FILE + "']")) .transform(p -> p.getFailedMessage().getPayload().toString() + "\n" + getStackTraceAsString(p)) - .channel("toMail.input"); + .channel("toMail.input") + .get(); } @Bean diff --git a/basic/amqp/src/main/resources/META-INF/spring/integration/spring-integration-context.xml b/basic/amqp/src/main/resources/META-INF/spring/integration/spring-integration-context.xml index 8dac90e7..72d0bb92 100644 --- a/basic/amqp/src/main/resources/META-INF/spring/integration/spring-integration-context.xml +++ b/basic/amqp/src/main/resources/META-INF/spring/integration/spring-integration-context.xml @@ -25,7 +25,7 @@ - diff --git a/basic/amqp/src/main/resources/log4j.xml b/basic/amqp/src/main/resources/log4j.xml index 094beea2..ac4d2b98 100644 --- a/basic/amqp/src/main/resources/log4j.xml +++ b/basic/amqp/src/main/resources/log4j.xml @@ -11,6 +11,10 @@ + + + + @@ -29,4 +33,4 @@ - \ No newline at end of file + diff --git a/basic/barrier/src/main/java/org/springframework/integration/samples/barrier/Application.java b/basic/barrier/src/main/java/org/springframework/integration/samples/barrier/Application.java index 0706ffc4..90bf216e 100644 --- a/basic/barrier/src/main/java/org/springframework/integration/samples/barrier/Application.java +++ b/basic/barrier/src/main/java/org/springframework/integration/samples/barrier/Application.java @@ -23,12 +23,14 @@ import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.boot.builder.SpringApplicationBuilder; import org.springframework.context.ConfigurableApplicationContext; import org.springframework.context.annotation.ImportResource; +import org.springframework.integration.http.config.EnableIntegrationGraphController; /** * @author Gary Russell * @since 4.2 */ @SpringBootApplication +@EnableIntegrationGraphController @ImportResource("/META-INF/spring/integration/server-context.xml") public class Application { @@ -51,9 +53,9 @@ public class Application { System.out.println("\n\n++++++++++++ Sending: " + request + " ++++++++++++\n"); String reply = requestGateway.echo(request); System.out.println("\n\n++++++++++++ Replied with: " + reply + " ++++++++++++\n"); - client.close(); - server.close(); - System.exit(0); // AMQP-519 +// client.close(); +// server.close(); +// System.exit(0); // AMQP-519 } diff --git a/basic/barrier/src/main/resources/application.properties b/basic/barrier/src/main/resources/application.properties index 7a694787..e69de29b 100644 --- a/basic/barrier/src/main/resources/application.properties +++ b/basic/barrier/src/main/resources/application.properties @@ -1 +0,0 @@ -spring.output.ansi.enabled=always diff --git a/basic/http/src/main/webapp/WEB-INF/servlet-config.xml b/basic/http/src/main/webapp/WEB-INF/servlet-config.xml index c61a810f..8795a827 100644 --- a/basic/http/src/main/webapp/WEB-INF/servlet-config.xml +++ b/basic/http/src/main/webapp/WEB-INF/servlet-config.xml @@ -12,11 +12,16 @@ + + + + diff --git a/basic/kafka/src/main/java/org/springframework/integration/samples/kafka/Application.java b/basic/kafka/src/main/java/org/springframework/integration/samples/kafka/Application.java index a41ea0ce..5cce17be 100644 --- a/basic/kafka/src/main/java/org/springframework/integration/samples/kafka/Application.java +++ b/basic/kafka/src/main/java/org/springframework/integration/samples/kafka/Application.java @@ -46,6 +46,7 @@ import org.springframework.kafka.listener.KafkaMessageListenerContainer; import org.springframework.kafka.listener.config.ContainerProperties; import org.springframework.kafka.support.KafkaNull; import org.springframework.kafka.support.TopicPartitionInitialOffset; +import org.springframework.kafka.support.serializer.JsonDeserializer; import org.springframework.messaging.Message; import org.springframework.messaging.MessageChannel; import org.springframework.messaging.MessageHandler; @@ -140,7 +141,9 @@ public class Application { props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, 15000); props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class); - return new DefaultKafkaConsumerFactory<>(props); + final DefaultKafkaConsumerFactory cf = new DefaultKafkaConsumerFactory<>(props); + cf.setValueDeserializer(new JsonDeserializer<>(String.class)); + return cf; } @Bean diff --git a/build.gradle b/build.gradle index 1af71e22..bac3b6dd 100644 --- a/build.gradle +++ b/build.gradle @@ -174,7 +174,7 @@ subprojects { subproject -> groovyVersion = '2.3.0' hsqldbVersion = '2.3.2' h2Version = '1.3.175' - jacksonVersion = '2.3.2' + jacksonVersion = '2.6.7' jasyptVersion = '1.7' javaxInjectVersion = '1' javaxMailVersion = '1.5.5' @@ -201,7 +201,7 @@ subprojects { subproject -> springIntegrationKafkaVersion = '2.1.0.RELEASE' springIntegrationSplunkVersion = '1.1.0.RELEASE' springKafkaVersion = '1.1.0.RELEASE' - springVersion = '4.3.2.RELEASE' + springVersion = '4.3.3.RELEASE' springSecurityVersion = '4.0.2.RELEASE' springWebFlowVersion = '2.3.3.RELEASE' tilesJspVersion = '2.2.1' @@ -612,6 +612,7 @@ project('kafka') { compile ("org.springframework.kafka:spring-kafka-test:$springKafkaVersion") { exclude group: 'org.slf4j' } + compile "com.fasterxml.jackson.core:jackson-databind:$jacksonVersion" compile "log4j:log4j:$log4jVersion"