A bit of reactor updates

2026-05-30 10:50:42 +00:00 · 2021-11-14 00:43:59 +03:00
parent e1bfd6ae56
commit 76e3423102
10 changed files with 74 additions and 66 deletions
@@ -11,7 +11,7 @@
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
-        <version>2.2.6.RELEASE</version>
+        <version>2.5.6</version>
        <relativePath/>
    </parent>

@@ -24,6 +24,7 @@
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter</artifactId>
        </dependency>
+
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-data-mongodb-reactive</artifactId>
@@ -32,7 +33,6 @@
            <groupId>org.mongodb</groupId>
            <artifactId>mongodb-driver-reactivestreams</artifactId>
        </dependency>
-
        <dependency>
            <groupId>de.flapdoodle.embed</groupId>
            <artifactId>de.flapdoodle.embed.mongo</artifactId>
@@ -4,33 +4,33 @@ public class Account {
    private String id;
    private String personId;
    private Long amount;
-    
+
    public Account(String id, String personId, Long amount) {
        this.id = id;
        this.personId = personId;
        this.amount = amount;
    }
-    
+
    public String getId() {
        return id;
    }
-    
+
    public void setId(String id) {
        this.id = id;
    }
-    
+
    public String getPersonId() {
        return personId;
    }
-    
+
    public void setPersonId(String personId) {
        this.personId = personId;
    }
-    
+
    public Long getAmount() {
        return amount;
    }
-    
+
    public void setAmount(Long amount) {
        this.amount = amount;
    }
@@ -2,7 +2,6 @@ package ru.otus.spring.repostory;

 import org.springframework.data.mongodb.repository.ReactiveMongoRepository;
 import ru.otus.spring.domain.Account;
-import ru.otus.spring.domain.Person;

 public interface AccountRepository extends ReactiveMongoRepository<Account, String> {
 }
@@ -2,12 +2,12 @@ package ru.otus.spring.repostory;

 import org.springframework.data.mongodb.repository.Query;
 import org.springframework.data.mongodb.repository.ReactiveMongoRepository;
-import org.springframework.data.repository.reactive.ReactiveCrudRepository;
 import reactor.core.publisher.Flux;
 import reactor.core.publisher.Mono;
 import ru.otus.spring.domain.Person;

 public interface PersonRepository extends ReactiveMongoRepository<Person, String> {
+
    Flux<Person> findByName(String name);

    @Query("{ 'name': ?0 }")
@@ -11,7 +11,7 @@
    <parent>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-parent</artifactId>
-        <version>2.2.6.RELEASE</version>
+        <version>2.5.6</version>
        <relativePath/>
    </parent>

@@ -3,7 +3,7 @@ package ru.otus.spring;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.context.ConfigurableApplicationContext;
-import ru.otus.spring.reactor.FluxService;
+import ru.otus.spring.reactor.ReactiveProcessingService;

@SpringBootApplication
 public class Main {
@@ -11,7 +11,7 @@ public class Main {
    public static void main(String[] args) {
        ConfigurableApplicationContext context = SpringApplication.run(Main.class);

-        FluxService service = context.getBean(FluxService.class);
+        ReactiveProcessingService service = context.getBean(ReactiveProcessingService.class);

        service.printHello("Ivan");
    }
@@ -1,50 +0,0 @@
-package ru.otus.spring.reactor;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Service;
-import reactor.core.Disposable;
-import reactor.core.publisher.DirectProcessor;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
-
-@Service
-public class FluxService {
-
-    private final Logger logger = LoggerFactory.getLogger(FluxService.class);
-
-    private final NonFluxService nonFluxService;
-    private final DirectProcessor<Message> processor;
-    private final Disposable flow;
-
-    @Autowired
-    public FluxService(NonFluxService nonFluxService) {
-        this.nonFluxService = nonFluxService;
-        // Создаём процессор - это reactor-овская реализация reactive-stream интерфейса
-        // Direct processor, кстати - это простой последовательный вызов методов)
-        processor = DirectProcessor.create();
-        // Здесь мы настриваем flow
-        flow = Mono.from(processor)
-                   .map(nonFluxService::nonFluxSayHello)
-                   .subscribe(this::printMessage);
-    }
-
-    /**
-     * Этот метод будет инициировать асинзронную обрабтку сообщения
-     *
-     * @param name это имя будет приходить из не-reactor окружения
-     */
-    public void printHello(String name) {
-        processor.onNext(new Message(name));
-    }
-
-    /**
-     * А это терминальный шаг для сообщения
-     *
-     * @param message а это финальный шаг для сообщения, отсюда можно вернуть рзультат в не-реактив окружение
-     */
-    private void printMessage(Message message) {
-        logger.info("Message received: {}", message.getValue());
-    }
-}
@@ -0,0 +1,58 @@
+package ru.otus.spring.reactor;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.stereotype.Service;
+import reactor.core.Disposable;
+import reactor.core.publisher.Sinks;
+import ru.otus.spring.service.NonFluxService;
+
+import javax.annotation.PreDestroy;
+
+@Service
+public class ReactiveProcessingService {
+
+    private final Logger logger = LoggerFactory.getLogger(ReactiveProcessingService.class);
+
+    private final Sinks.Many<Message> sink;
+    private final Disposable flow;
+
+    public ReactiveProcessingService(NonFluxService nonFluxService) {
+        // Создаём sink (ранее - процессор)
+        // Это reactor-овская реализация reactive-stream интерфейса
+        // Обрабатывает данные как простой последовательный вызов методов :)
+        sink = Sinks.many().multicast().directBestEffort();
+        // Здесь мы настраиваем flow
+        flow = sink.asFlux()
+                .map(nonFluxService::nonFluxSayHello)
+                .subscribe(this::printMessage);
+        // в идеале в коде выше должен быть doOnNext
+        // в map не предполагаются задержки
+    }
+
+    /**
+     * Этот метод будет инициировать асинхронную обрабтку сообщения
+     *
+     * @param name это имя будет приходить из не-reactor окружения
+     */
+    public void printHello(String name) {
+        sink.tryEmitNext(new Message(name));
+    }
+
+    /**
+     * А это терминальный шаг для сообщения
+     *
+     * @param message а это финальный шаг для сообщения, отсюда можно вернуть рзультат в не-реактив окружение
+     */
+    private void printMessage(Message message) {
+        logger.info("Message received: {}", message.getValue());
+    }
+
+    /**
+     * Просто пример, как остановить процесс
+     */
+    @PreDestroy
+    public void dispose() {
+        this.flow.dispose();
+    }
+}
@@ -1,8 +1,9 @@
-package ru.otus.spring.reactor;
+package ru.otus.spring.service;

 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.stereotype.Service;
+import ru.otus.spring.reactor.Message;

@Service
 public class NonFluxService {
@@ -25,7 +25,7 @@
            <artifactId>spring-boot-starter-webflux</artifactId>
        </dependency>

-        <!-- работает и без неё, но нужна нам, для RxJava методов -->
+        <!-- работает и без неё, но нужна для RxJava контроллера -->
        <dependency>
            <groupId>io.reactivex.rxjava2</groupId>
            <artifactId>rxjava</artifactId>