21-02 Spring 18 Reactive programming

2021-02/spring-18/.gitignore

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

2021-02/spring-18/pom.xml

@@ -0,0 +1,33 @@
+<?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-18</artifactId>
+    <version>1.0-SNAPSHOT</version>
+
+    <parent>
+        <groupId>org.springframework.boot</groupId>
+        <artifactId>spring-boot-starter-parent</artifactId>
+        <version>2.2.4.RELEASE</version>
+    </parent>
+
+    <properties>
+        <maven.compiler.source>11</maven.compiler.source>
+        <maven.compiler.target>11</maven.compiler.target>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>io.reactivex.rxjava2</groupId>
+            <artifactId>rxjava</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.google.guava</groupId>
+            <artifactId>guava</artifactId>
+            <version>27.1-jre</version>
+        </dependency>
+    </dependencies>
+</project>

2021-02/spring-18/src/main/java/ru/otus/CreateExamples.java

@@ -0,0 +1,35 @@
+package ru.otus;
+
+import io.reactivex.Observable;
+
+@SuppressWarnings("ResultOfMethodCallIgnored")
+public class CreateExamples {
+
+    public static void main(String[] args) {
+        Observable<String> obs = justExample();
+        obs.forEach(System.out::println);
+        obs.forEach(System.out::println);
+    }
+    
+    public static Observable<String> justExample() {
+        return Observable.just("one", "two", "three");
+    }
+    
+    public static Observable<String> createExample() {
+        return Observable.create(emitter -> {
+            if (emitter.isDisposed()) {
+                return;
+            }
+            emitter.onNext("one");
+            emitter.onNext("two");//!
+            emitter.onNext("three");
+            if (!emitter.isDisposed()) {
+                emitter.onComplete();
+            }
+        });
+    }
+    
+    public static Observable<String> deferExample() {
+        return Observable.defer(() -> Observable.just("one", "two", "three"));
+    }
+}

2021-02/spring-18/src/main/java/ru/otus/LiveLikeExample.java

@@ -0,0 +1,26 @@
+package ru.otus;
+
+import io.reactivex.Observable;
+
+import java.io.IOException;
+
+public class LiveLikeExample {
+    
+    public static void main(String[] args) throws IOException {
+        
+        System.in.read();
+    }
+    
+    static Observable<String> getName() {
+        return Observable.just("Jake");
+    }
+    
+    static Observable<String> getSurname() {
+        return Observable.just("Foo");
+    }
+    
+    static Observable<String> save(String fullName) {
+        System.out.println(fullName + " saved!");
+        return Observable.just("OK!");
+    }
+}

2021-02/spring-18/src/main/java/ru/otus/OperatorsExample.java

@@ -0,0 +1,79 @@
+package ru.otus;
+
+import com.google.common.collect.ImmutableList;
+import io.reactivex.Observable;
+import io.reactivex.ObservableTransformer;
+import io.reactivex.schedulers.Schedulers;
+import io.reactivex.subjects.PublishSubject;
+
+import java.time.LocalDate;
+import java.util.List;
+import java.util.Random;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+
+@SuppressWarnings("ResultOfMethodCallIgnored")
+public class OperatorsExample {
+    public static void main(String[] args) throws Exception {
+        simpleExample();
+        System.in.read();
+    }
+
+    public static void simpleExample() throws Exception {
+        List<Person> persons = ImmutableList.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)));
+
+        Observable.fromIterable(persons)
+            .filter(
+                person -> person.getBirth().isAfter(LocalDate.of(1990, 1, 1))
+            )
+            .map(p -> p.getFirstName() + " " + p.getLastName())
+            .toList()
+            .subscribe(System.out::println);
+    }
+
+    public static void publisherExample() throws Exception {
+        final Observable<String> ob = magicPublisher();
+        System.out.println("First subscribed");
+        ob.subscribe(System.out::println);
+        Thread.sleep(5000);
+        System.out.println("Second subscribed");
+        ob.subscribe(System.out::println);
+    }
+
+    public static Observable<String> magicPublisher() {
+        Random r = new Random(1);
+        AtomicInteger i = new AtomicInteger();
+        final Observable<String> obs = Observable.<String>generate(emitter ->
+            emitter.onNext("" + i.incrementAndGet()))
+            .concatMap(s -> Observable.just(s).delay(r.nextInt(1000), TimeUnit.MILLISECONDS))
+            .subscribeOn(Schedulers.newThread());
+        PublishSubject<String> subject = PublishSubject.create();
+
+//        BehaviorSubject<String> subject = BehaviorSubject.create();
+
+//        AsyncSubject<String> subject = AsyncSubject.create();
+//        CompletableFuture.runAsync(() -> {
+//            try {
+//                Thread.sleep(7000);
+//            } catch (InterruptedException e) {
+//                e.printStackTrace();
+//            }
+//            subject.onComplete();
+//        });
+
+//        ReplaySubject<String> subject = ReplaySubject.create();
+        obs.subscribe(subject);
+        return subject;
+    }
+
+    //composeExmaple
+    private static ObservableTransformer<String, String> filterAndUpperCase() {
+        return upstream -> upstream
+            .filter(s -> s.length() >= 4)
+            .map(String::toUpperCase);
+    }
+}

2021-02/spring-18/src/main/java/ru/otus/Person.java

@@ -0,0 +1,66 @@
+package ru.otus;
+
+import java.time.LocalDate;
+import java.util.Objects;
+
+public class Person {
+    private String firstName;
+    private String lastName;
+    private String gender;
+    private LocalDate birth;
+    
+    public Person(String firstName, String lastName, String gender, LocalDate birth) {
+        this.firstName = firstName;
+        this.lastName = lastName;
+        this.gender = gender;
+        this.birth = birth;
+    }
+    
+    public String getFirstName() {
+        return firstName;
+    }
+    
+    public void setFirstName(String firstName) {
+        this.firstName = firstName;
+    }
+    
+    public String getLastName() {
+        return lastName;
+    }
+    
+    public void setLastName(String lastName) {
+        this.lastName = lastName;
+    }
+    
+    public String getGender() {
+        return gender;
+    }
+    
+    public void setGender(String gender) {
+        this.gender = gender;
+    }
+    
+    public LocalDate getBirth() {
+        return birth;
+    }
+    
+    public void setBirth(LocalDate birth) {
+        this.birth = birth;
+    }
+    
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+        Person person = (Person) o;
+        return Objects.equals(firstName, person.firstName) &&
+               Objects.equals(lastName, person.lastName) &&
+               Objects.equals(gender, person.gender) &&
+               Objects.equals(birth, person.birth);
+    }
+    
+    @Override
+    public int hashCode() {
+        return Objects.hash(firstName, lastName, gender, birth);
+    }
+}

2021-02/spring-18/src/main/java/ru/otus/comparison/AsyncComparison.java

@@ -0,0 +1,40 @@
+package ru.otus.comparison;
+
+import io.reactivex.Observable;
+import io.reactivex.schedulers.Schedulers;
+
+import java.io.IOException;
+
+public class AsyncComparison {
+    
+    public static void main(String[] args) throws IOException {
+        final long timeStarted = System.currentTimeMillis();
+        final Observable<String> obs = controller();
+        obs.subscribe(System.out::println);
+        System.out.println("Wait time " + (System.currentTimeMillis() - timeStarted));
+        System.in.read();
+    }
+    
+    static Observable<String> controller() {
+        return service();
+    }
+    
+    static Observable<String> service() {
+        return repository();
+    }
+    
+    static Observable<String> repository() {
+        return database();
+    }
+    
+    static Observable<String> database() {
+        return Observable.defer(() -> {
+            try {
+                Thread.sleep(4000);
+            } catch (Exception e) {
+                System.out.println("Don't do this");
+            }
+            return Observable.just("Hello world");
+        }).subscribeOn(Schedulers.newThread());
+    }
+}

2021-02/spring-18/src/main/java/ru/otus/comparison/SyncComparison.java

@@ -0,0 +1,31 @@
+package ru.otus.comparison;
+
+public class SyncComparison {
+    
+    public static void main(String[] args) {
+        final long timeStarted = System.currentTimeMillis();
+        System.out.println(controller());
+        System.out.println(System.currentTimeMillis() - timeStarted);
+    }
+    
+    static String controller() {
+        return service();
+    }
+    
+    static String service() {
+        return repository();
+    }
+    
+    static String repository() {
+        return database();
+    }
+    
+    static String database() {
+        try {
+            Thread.sleep(4000);
+        } catch (Exception e) {
+            System.out.println("Don't do this");
+        }
+        return "Hello world";
+    }
+}