spring-20-reactor

2024-03/spring-20-reactor/.gitignore

@@ -0,0 +1,7 @@
+.idea/
+*.iml
+
+target/
+
+/node_modules
+/output

2024-03/spring-20-reactor/pom.xml

@@ -0,0 +1,70 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>ru.otus</groupId>
+    <artifactId>spring-20-reactor</artifactId>
+    <version>1.0-SNAPSHOT</version>
+
+    <parent>
+        <groupId>org.springframework.boot</groupId>
+        <artifactId>spring-boot-starter-parent</artifactId>
+        <version>3.1.0</version>
+    </parent>
+
+    <properties>
+        <maven.compiler.source>17</maven.compiler.source>
+        <maven.compiler.target>17</maven.compiler.target>
+    </properties>
+
+    <dependencyManagement>
+        <dependencies>
+            <dependency>
+                <groupId>io.projectreactor</groupId>
+                <artifactId>reactor-bom</artifactId>
+                <version>2022.0.7</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
+            <dependency>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-dependencies</artifactId>
+                <version>3.1.0</version>
+                <type>pom</type>
+                <scope>import</scope>
+            </dependency>
+        </dependencies>
+    </dependencyManagement>
+
+
+    <dependencies>
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-core</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>ch.qos.logback</groupId>
+            <artifactId>logback-classic</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>io.projectreactor</groupId>
+            <artifactId>reactor-test</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+
+    </dependencies>
+</project>

2024-03/spring-20-reactor/src/main/java/ru/otus/CreateExample.java

@@ -0,0 +1,74 @@
+package ru.otus;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import reactor.core.Disposable;
+import reactor.core.publisher.Flux;
+
+public class CreateExample {
+    private static final Logger logger = LoggerFactory.getLogger(CreateExample.class);
+
+    public static void main(String[] args) {
+//         onEachNext();
+//         lazyObservable();
+//         creatorExample();
+    }
+
+    private static void onEachNext() {
+        Flux<String> obs = Flux.just("one", "two", "three");
+        obs.doFirst(() -> logger.info("Starting:"))
+                .doOnComplete(() -> logger.info("The end!"))
+                .doOnEach(item -> logger.info("item_1:{}", item.get()))
+                .subscribe();
+
+        logger.info("-----");
+
+        obs.doOnNext(item -> logger.info("item_2:{}", item))
+                .map(String::length)
+                .doOnNext(item -> logger.info("length_2:{}", item))
+                .subscribe();
+    }
+
+    private static void lazyObservable() {
+        Flux<String> obs = Flux.defer(() -> {
+            logger.info("creating new Observable");
+            return Flux.just("one", "two", "three");
+        });
+
+        obs.doOnNext(item -> logger.info("item_1:{}", item))
+                .subscribe();
+
+        logger.info("---------------");
+
+        obs.doOnNext(item -> logger.info("item_2:{}", item))
+                .subscribe();
+    }
+
+    private static void creatorExample() {
+        Flux<String> obs = Flux.create(emitter -> {
+            emitter.next("one");
+            emitter.next("two");
+
+            emitter.error(new RuntimeException("Error!"));
+
+            emitter.next("three");
+            emitter.complete();
+        });
+
+        obs.onErrorResume(e -> {
+                    logger.error("error:{}", e.getMessage(), e);
+                    return Flux.just("r1", "r2", "r3");
+                })
+                .doOnNext(item -> logger.info("item__1:{}", item))
+                .subscribe();
+
+        logger.info("---------------");
+
+        Disposable disposable = obs.doOnNext(item -> logger.info("item__2:{}", item))
+                .subscribe(next -> logger.info("next:{}", next),
+                        error -> logger.info("error:{}", error.getMessage()),
+                        () -> logger.info("onComplete"));
+
+        logger.info("isDisposed:{}", disposable.isDisposed());
+    }
+}

2024-03/spring-20-reactor/src/main/java/ru/otus/OperatorsExample.java

@@ -0,0 +1,51 @@
+package ru.otus;
+
+import java.time.LocalDate;
+import java.util.List;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import reactor.core.publisher.Flux;
+
+
+public class OperatorsExample {
+    private static final Logger logger = LoggerFactory.getLogger(OperatorsExample.class);
+
+    public static void main(String[] args) {
+//         merge();
+//         fromList();
+    }
+
+    private static void merge() {
+        var listFlux1 = Flux.fromIterable(List.of(
+                new Person("John", "Dow", "male", LocalDate.of(1992, 3, 12)),
+                new Person("Jane", "Dow", "female", LocalDate.of(2001, 6, 23))));
+
+        var listFlux2 = Flux.fromIterable(List.of(
+                new Person("Howard", "Lovecraft", "male", LocalDate.of(1890, 8, 20)),
+                new Person("Joanne", "Rowling", "female", LocalDate.of(1965, 6, 30))));
+
+        var listFlux3 = Flux.fromIterable(List.of(
+                new Person("Ivan", "Petrov", "male", LocalDate.of(1890, 2, 10)),
+                new Person("Joanne", "Stuard", "female", LocalDate.of(1965, 1, 3))));
+
+        Flux.merge(listFlux1, listFlux2, listFlux3)
+                .subscribe(person -> logger.info("person:{}", person));
+
+    }
+
+    private static void fromList() {
+        var persons = List.of(
+                new Person("John", "Dow", "male", LocalDate.of(1992, 3, 12)),
+                new Person("Jane", "Dow", "female", LocalDate.of(2001, 6, 23)),
+                new Person("Howard", "Lovecraft", "male", LocalDate.of(1890, 8, 20)),
+                new Person("Joanne", "Rowling", "female", LocalDate.of(1965, 6, 30)));
+
+        var disposable = Flux.fromIterable(persons)
+                .filter(person -> person.birth().isAfter(LocalDate.of(1990, 1, 1)))
+                .map(p -> p.firstName() + " " + p.lastName())
+                .collectList()
+                .subscribe(item -> logger.info("item: {}", item));
+
+        logger.info("disposable.isDisposed:{}", disposable.isDisposed());
+    }
+}

2024-03/spring-20-reactor/src/main/java/ru/otus/Person.java

@@ -0,0 +1,6 @@
+package ru.otus;
+
+import java.time.LocalDate;
+
+public record Person(String firstName, String lastName, String gender, LocalDate birth) {
+}

2024-03/spring-20-reactor/src/main/java/ru/otus/PublisherExample.java

@@ -0,0 +1,56 @@
+package ru.otus;
+
+import java.time.Duration;
+import java.util.function.BiFunction;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import reactor.core.publisher.ConnectableFlux;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.SynchronousSink;
+import reactor.core.scheduler.Schedulers;
+
+public class PublisherExample {
+    private static final Logger logger = LoggerFactory.getLogger(PublisherExample.class);
+
+    public static void main(String[] args) throws Exception {
+        publisherExample();
+    }
+
+    public static void publisherExample() throws InterruptedException {
+        Flux<String> ob = magicPublisher();
+        Thread.sleep(5000);
+        logger.info("First subscribed");
+        var disposable1 = ob.subscribe(item -> logger.info("item: {}", item));
+
+        logger.info("disposable1.isDisposed:{}", disposable1.isDisposed());
+        Thread.sleep(5000);
+
+        logger.info("Second subscribed");
+        var disposable2 = ob.subscribe(item -> logger.info("item second: {}", item));
+
+
+        logger.info("disposable2.isDisposed():{}", disposable2.isDisposed());
+
+        Thread.sleep(60_000);
+    }
+
+    public static Flux<String> magicPublisher() {
+        var schedulerGenerator = Schedulers.newParallel("generator", 1);
+        var generator = Flux.generate(
+                        () -> 0L,
+                        (BiFunction<Long, SynchronousSink<Long>, Long>)
+                                (prev, sink) -> {
+                                    var newValue = prev + 1;
+                                    sink.next(newValue);
+                                    logger.info("newValue:{}", newValue);
+                                    return newValue;
+                                })
+                .delayElements(Duration.ofSeconds(5), schedulerGenerator)
+                .map(id -> "new id:" + id)
+                .doOnNext(val -> logger.info("val:{}", val));
+
+        ConnectableFlux<String> generatorConnectable = generator.publish();
+
+        return generatorConnectable.autoConnect(0);
+    }
+}