概要
記事一覧はこちらです。
- org.springframework.integration.core.MessageSource からデータ取得して処理する処理を任意に stop したり start したり制御できないか考えてみました。そのメモ書きです。
参照したサイト・書籍
目次
- ksbysample-eipapp-messagesource-controll プロジェクトを作成する
- まずは 1秒毎に hello を出力する処理を書いてみる
- MessageSource から null を返せば何も処理されないのか？
- MessageSource から Message を返すか否かを 5秒毎に切り替えてみる
手順
ksbysample-eipapp-messagesource-controll プロジェクトを作成する
今回は Spring Initializr で作成します。
生成された build.gradle は以下のものですが、
buildscript {
ext {
springBootVersion = '2.0.4.RELEASE'
}
repositories {
mavenCentral()
}
dependencies {
classpath("org.springframework.boot:spring-boot-gradle-plugin:${springBootVersion}")
}
}
apply plugin: 'java'
apply plugin: 'eclipse'
apply plugin: 'org.springframework.boot'
apply plugin: 'io.spring.dependency-management'
group = 'ksbysample'
version = '0.0.1-SNAPSHOT'
sourceCompatibility = 1.8
repositories {
mavenCentral()
}
dependencies {
compile('org.springframework.boot:spring-boot-starter-integration')
testCompile('org.springframework.boot:spring-boot-starter-test')
}
lombok の @Slf4j を使いたいので dependencies block を以下のように変更します。Spring Initializr で生成された Gradle プロジェクトの Gradle は 4.8.1 だったので、記述も少し変更しました。
dependencies {
def lombokVersion = "1.18.2"
implementation('org.springframework.boot:spring-boot-starter-integration')
testImplementation('org.springframework.boot:spring-boot-starter-test')
// for lombok
annotationProcessor("org.projectlombok:lombok:${lombokVersion}")
compileOnly("org.projectlombok:lombok:${lombokVersion}")
}
build.gradle を以下の内容に変更した後、Gradle Tool Window の左上にある「Refresh all Gradle projects」ボタンをクリックして更新します。
@SpringBootApplication が付与されているクラス名が長かったので、短く Application クラスに変更します。
まずは 1秒毎に hello を出力する処理を書いてみる
以下の仕様で実装します。
- MessageSource は "hello" の文字列が payload にセットされたメッセージを返す。
- IntegrationFlow を返す Bean の処理は、1秒毎に MessageSource からメッセージを取得してログに出力する。
src/main/java/ksbysample/eipapp/mscontroll の下に FlowConfig.java を新規作成し、以下の内容を記述します。
package ksbysample.eipapp.mscontroll; import lombok.extern.slf4j.Slf4j; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.integration.core.MessageSource; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.dsl.Pollers; import org.springframework.integration.support.MessageBuilder; @Slf4j @Configuration public class FlowConfig { @Bean public MessageSource<String> helloMessageSource() { return () -> MessageBuilder.withPayload("hello").build(); } @Bean public IntegrationFlow helloFlow() { return IntegrationFlows.from(helloMessageSource() , e -> e.poller(Pollers.fixedDelay(1000))) .handle((p, h) -> { log.info("★★★ " + p); return null; }) .get(); } }
実行すると以下のようにログが出力されます。
MessageSource から null を返せば何も処理されないのか？
MessageSource から Message を返さずに null を返すと処理は何も起きない気がするので、試してみます。helloMessageSource メソッドを以下のように変更します。
@Bean public MessageSource<String> helloMessageSource() { // return () -> MessageBuilder.withPayload("hello").build(); return () -> null; }
実行してみると Started Application in 1.981 seconds が出力された後は何も出力されませんでした。
MessageSource から Message を返すか否かを 5秒毎に切り替えてみる
以下の仕様で実装し直してみます。
- MessageSource は boolean 型のフィールド変数を見て Message を返すか否かを切り替えられるようにする。
- フィールド変数の値は @Scheduled アノテーションを付与したメソッドで 5秒毎に変更する。
まずは @Scheduled アノテーションが使用できるよう src/main/java/ksbysample/eipapp/mscontroll/Application.java を以下のように変更します。
package ksbysample.eipapp.mscontroll; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; import org.springframework.scheduling.annotation.EnableScheduling; @SpringBootApplication @EnableScheduling public class Application { public static void main(String[] args) { SpringApplication.run(Application.class, args); } }
@EnableSchedulingを追加します。
src/main/java/ksbysample/eipapp/mscontroll/FlowConfig.java を以下のように変更します。
package ksbysample.eipapp.mscontroll; import lombok.extern.slf4j.Slf4j; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.integration.core.MessageSource; import org.springframework.integration.dsl.IntegrationFlow; import org.springframework.integration.dsl.IntegrationFlows; import org.springframework.integration.dsl.Pollers; import org.springframework.integration.support.MessageBuilder; import org.springframework.scheduling.annotation.Scheduled; @Slf4j @Configuration public class FlowConfig { private boolean messageSourceControlFlg = true; @Bean public MessageSource<String> helloMessageSource() { return () -> messageSourceControlFlg ? MessageBuilder.withPayload("hello").build() : null; } @Bean public IntegrationFlow helloFlow() { return IntegrationFlows.from(helloMessageSource() , e -> e.poller(Pollers.fixedDelay(1000))) .handle((p, h) -> { log.info("★★★ " + p); return null; }) .get(); } @Scheduled(initialDelay = 5000, fixedDelay = 5000) public void messageSourceControlFlgChanger() { messageSourceControlFlg = !messageSourceControlFlg; log.warn(String.valueOf(messageSourceControlFlg)); } }
private boolean messageSourceControlFlg = true;を追加します。- helloMessageSource Bean の処理を
return () -> MessageBuilder.withPayload("hello").build();→return () -> messageSourceControlFlg ? MessageBuilder.withPayload("hello").build() : null;に変更します。 - messageSourceControlFlgChanger メソッドを追加します。@Scheduled アノテーションを付与して 5秒毎に実行されるようにします。
実行すると 5秒間 hello のログが出力されて、true → false に変わると5秒間何も起きず、false → true に変わると再び hello が出力されるようになりました。やりたかったことは実現できました。
履歴
2018/08/21
初版発行。
