SpringCloud/spring-cloud-dataflow-samples
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Refactor the stream applications

Browse Source 
Thanks to Soby Chacko for the suggestions
This commit is contained in:
Ilayaperumal Gopinathan
2020-05-02 07:17:55 +05:30
parent 6324d6e78a
commit dc20f90f03
3 changed files with 30 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files

31
spring-cloud-stream-function-bindings/log-sink/src/main/java/com/example/logsink/LogConsumerConfiguration.java
View File

@@ -1,31 +0,0 @@
package com.example.logsink;
import java.util.function.Consumer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.handler.LoggingHandler;
import org.springframework.messaging.Message;
@Configuration
public class LogConsumerConfiguration {
public LogConsumerConfiguration() {
}
@Bean
IntegrationFlow logConsumerFlow() {
return IntegrationFlows.from(MessageConsumer.class, (gateway) -> gateway.beanName("logConsumer"))
.handle((payload, headers) -> {
if (payload instanceof byte[]) {
return new String((byte[]) payload);
}
return payload;
})
.log(LoggingHandler.Level.INFO, "log-consumer", "payload")
.get();
}
private interface MessageConsumer extends Consumer<Message<?>> {}
}

22
spring-cloud-stream-function-bindings/log-sink/src/main/java/com/example/logsink/LogSinkApplication.java
View File

@@ -1,11 +1,33 @@
package com.example.logsink;
import java.util.function.Consumer;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.integration.dsl.IntegrationFlow;
import org.springframework.integration.dsl.IntegrationFlows;
import org.springframework.integration.handler.LoggingHandler;
import org.springframework.messaging.Message;
@SpringBootApplication
public class LogSinkApplication {
@Bean
IntegrationFlow logConsumerFlow() {
return IntegrationFlows.from(MessageConsumer.class, (gateway) -> gateway.beanName("logConsumer"))
.handle((payload, headers) -> {
if (payload instanceof byte[]) {
return new String((byte[]) payload);
}
return payload;
})
.log(LoggingHandler.Level.INFO, "log-consumer", "payload")
.get();
}
private interface MessageConsumer extends Consumer<Message<?>> {}
public static void main(String[] args) {
SpringApplication.run(LogSinkApplication.class, args);
}

22
spring-cloud-stream-function-bindings/time-source/src/main/java/com/example/timesource/TimeSourceApplication.java
View File

@@ -7,26 +7,20 @@ import java.util.function.Supplier;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
@SpringBootApplication
public class TimeSourceApplication {
@Bean
public Supplier<String> timeSupplier() {
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
return () -> {
return sdf.format(new Date());
};
}
public static void main(String[] args) {
SpringApplication.run(TimeSourceApplication.class, args);
}
@Configuration
public class TimeSupplierConfiguration {
public TimeSupplierConfiguration() {
}
@Bean
public Supplier<String> timeSupplier() {
SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
return () -> {
return sdf.format(new Date());
};
}
}
}