diff --git a/pom.xml b/pom.xml index 52c6ec611d..a61bd6ccd4 100644 --- a/pom.xml +++ b/pom.xml @@ -163,6 +163,7 @@ spring-thymeleaf spring-userservice spring-zuul + spring-reactor testing diff --git a/spring-reactor/pom.xml b/spring-reactor/pom.xml new file mode 100644 index 0000000000..906110131d --- /dev/null +++ b/spring-reactor/pom.xml @@ -0,0 +1,67 @@ + + 4.0.0 + com.baeldung + spring-reactor + jar + 1.0-SNAPSHOT + spring-reactor + http://maven.apache.org + + + org.springframework.boot + spring-boot-starter-parent + 1.4.2.RELEASE + + + + 1.8 + + + + + org.springframework.boot + spring-boot-starter + + + org.springframework.boot + spring-boot-starter-web + + + org.springframework + spring-web + + + io.projectreactor + reactor-bus + + + + + + + org.springframework.boot + spring-boot-maven-plugin + + + + + + + spring-snapshots + Spring Snapshots + https://repo.spring.io/libs-snapshot + + true + + + + + + spring-releases + Spring Releases + https://repo.spring.io/libs-release + + + + diff --git a/spring-reactor/src/main/java/com/baeldung/Application.java b/spring-reactor/src/main/java/com/baeldung/Application.java new file mode 100644 index 0000000000..67f713cd83 --- /dev/null +++ b/spring-reactor/src/main/java/com/baeldung/Application.java @@ -0,0 +1,49 @@ +package com.baeldung; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.CommandLineRunner; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.EnableAutoConfiguration; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.ComponentScan; +import org.springframework.context.annotation.Configuration; + +import com.baeldung.consumer.NotificationConsumer; + +import reactor.Environment; +import reactor.bus.EventBus; + +import static reactor.bus.selector.Selectors.$; + + +@Configuration +@EnableAutoConfiguration +@ComponentScan +public class Application implements CommandLineRunner { + + @Autowired + private EventBus eventBus; + + @Autowired + private NotificationConsumer notificationConsumer; + + @Bean + Environment env() { + return Environment.initializeIfEmpty().assignErrorJournal(); + } + + @Bean + EventBus createEventBus(Environment env) { + return EventBus.create(env, Environment.THREAD_POOL); + } + + @Override + public void run(String... args) throws Exception { + eventBus.on($("notificationConsumer"), notificationConsumer); + } + + public static void main(String[] args){ + SpringApplication.run(Application.class, args); + } + +} diff --git a/spring-reactor/src/main/java/com/baeldung/consumer/NotificationConsumer.java b/spring-reactor/src/main/java/com/baeldung/consumer/NotificationConsumer.java new file mode 100644 index 0000000000..0d33f78477 --- /dev/null +++ b/spring-reactor/src/main/java/com/baeldung/consumer/NotificationConsumer.java @@ -0,0 +1,28 @@ +package com.baeldung.consumer; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Service; + +import com.baeldung.doman.NotificationData; +import com.baeldung.service.NotificationService; + +import reactor.bus.Event; +import reactor.fn.Consumer; + +@Service +public class NotificationConsumer implements Consumer> { + + @Autowired + private NotificationService notificationService; + + @Override + public void accept(Event notificationDataEvent) { + + NotificationData notificationData = notificationDataEvent.getData(); + try { + notificationService.initiateNotofication(notificationData); + } catch (InterruptedException e) {} + + } + +} diff --git a/spring-reactor/src/main/java/com/baeldung/controller/NotificationController.java b/spring-reactor/src/main/java/com/baeldung/controller/NotificationController.java new file mode 100644 index 0000000000..a6dba64bc2 --- /dev/null +++ b/spring-reactor/src/main/java/com/baeldung/controller/NotificationController.java @@ -0,0 +1,37 @@ +package com.baeldung.controller; + +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Controller; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RequestMethod; + +import com.baeldung.doman.NotificationData; + +import reactor.bus.Event; +import reactor.bus.EventBus; + +@Controller +public class NotificationController { + + @Autowired + private EventBus eventBus; + + @RequestMapping(value = "/startNotification/{param}", method = RequestMethod.GET) + public void startNotification(@PathVariable("param") String param) { + + int notificationSize = Integer.parseInt(param); + + for(int i = 0; i < notificationSize; i++) { + + NotificationData data = new NotificationData(); + data.setId(i); + + eventBus.notify("notificationConsumer",Event.wrap(data)); + + System.out.println("Notification " +i +": notification task submitted successfully"); + } + + } + +} diff --git a/spring-reactor/src/main/java/com/baeldung/doman/NotificationData.java b/spring-reactor/src/main/java/com/baeldung/doman/NotificationData.java new file mode 100644 index 0000000000..e623d80887 --- /dev/null +++ b/spring-reactor/src/main/java/com/baeldung/doman/NotificationData.java @@ -0,0 +1,35 @@ +package com.baeldung.doman; + +public class NotificationData { + + private long id; + private String name; + private String email; + private String mobile; + + public long getId() { + return id; + } + public void setId(long id) { + this.id = id; + } + public String getName() { + return name; + } + public void setName(String name) { + this.name = name; + } + public String getEmail() { + return email; + } + public void setEmail(String email) { + this.email = email; + } + public String getMobile() { + return mobile; + } + public void setMobile(String mobile) { + this.mobile = mobile; + } + +} diff --git a/spring-reactor/src/main/java/com/baeldung/service/NotificationService.java b/spring-reactor/src/main/java/com/baeldung/service/NotificationService.java new file mode 100644 index 0000000000..348024d8a1 --- /dev/null +++ b/spring-reactor/src/main/java/com/baeldung/service/NotificationService.java @@ -0,0 +1,9 @@ +package com.baeldung.service; + +import com.baeldung.doman.NotificationData; + +public interface NotificationService { + + public void initiateNotofication(NotificationData notificationData) throws InterruptedException; + +} diff --git a/spring-reactor/src/main/java/com/baeldung/service/impl/NotificationServiceimpl.java b/spring-reactor/src/main/java/com/baeldung/service/impl/NotificationServiceimpl.java new file mode 100644 index 0000000000..e5dc209384 --- /dev/null +++ b/spring-reactor/src/main/java/com/baeldung/service/impl/NotificationServiceimpl.java @@ -0,0 +1,21 @@ +package com.baeldung.service.impl; + +import org.springframework.stereotype.Service; + +import com.baeldung.doman.NotificationData; +import com.baeldung.service.NotificationService; + +@Service +public class NotificationServiceimpl implements NotificationService { + + @Override + public void initiateNotofication(NotificationData notificationData) throws InterruptedException { + + System.out.println("Notification service started for Notification ID: " +notificationData.getId()); + + Thread.sleep(5000); + + System.out.println("Notification service ended for Notification ID: " +notificationData.getId()); + } + +}