JAVA-11499 move rxjava* modules to new rxjava-modules (#12342)

* JAVA-11499 move rxjava* modules to new rxjava-modules * JAVA-11499 addressed PR comments
2022-06-19 01:51:25 +10:00
parent 997307e55f
commit 17ba25e795
64 changed files with 79 additions and 111 deletions
@@ -0,0 +1,43 @@
+package com.baeldung.rxjava;
+
+import rx.Observable;
+
+import java.util.List;
+
+public class ComputeFunction {
+    public static void compute(Integer v) {
+        try {
+            System.out.println("compute integer v: " + v);
+            Thread.sleep(1000);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+
+    public static void compute(List<Integer> v) {
+        try {
+            System.out.println("compute integer v: " + v);
+            Thread.sleep(1000);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+
+    public static void compute(Observable<Integer> v) {
+        try {
+            v.forEach(System.out::println);
+            Thread.sleep(1000);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+
+    public static void compute(Long v) {
+        try {
+            System.out.println("compute integer v: " + v);
+            Thread.sleep(1000);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+}
@@ -0,0 +1,22 @@
+package com.baeldung.rxjava;
+
+import rx.Observable;
+import rx.observables.ConnectableObservable;
+
+import java.util.concurrent.TimeUnit;
+
+public class ConnectableObservableImpl {
+
+    public static void main(String[] args) throws InterruptedException {
+
+        ConnectableObservable<Long> connectable
+          = Observable.interval(200, TimeUnit.MILLISECONDS).publish();
+        connectable.subscribe(System.out::println);
+
+        System.out.println("Connect");
+        connectable.connect();
+
+        Thread.sleep(500);
+        System.out.println("Sleep");
+    }
+}
@@ -0,0 +1,81 @@
+package com.baeldung.rxjava;
+
+import rx.Observable;
+import rx.observables.BlockingObservable;
+
+import java.util.Arrays;
+import java.util.List;
+
+public class ObservableImpl {
+
+    private static Integer[] numbers = {0, 1, 2, 3, 4, 5, 6, 7, 8, 9};
+
+    private static String[] letters = {"a", "b", "c", "d", "e", "f", "g", "h", "i"};
+    private static String[] titles = {"title"};
+    private static List<String> titleList = Arrays.asList(titles);
+
+    static Observable<String> getTitle() {
+        return Observable.from(titleList);
+    }
+
+    public static void main(String[] args) {
+
+        System.out.println("-------Just-----------");
+        Observable<String> observable = Observable.just("Hello");
+        observable.subscribe(
+          System.out::println, //onNext
+          Throwable::printStackTrace, //onError
+          () -> System.out.println("onCompleted")  //onCompleted
+        );
+
+        BlockingObservable<String> blockingObservable = observable.toBlocking();
+
+        System.out.println();
+        System.out.println("-------Map-----------");
+        Observable.from(letters)
+          .map(String::toUpperCase)
+          .subscribe(System.out::print);
+
+        System.out.println();
+        System.out.println("-------FlatMap-----------");
+        Observable.just("book1", "book2")
+          .flatMap(s -> getTitle())
+          .subscribe(System.out::print);
+
+        System.out.println();
+        System.out.println("--------Scan----------");
+        Observable.from(letters)
+          .scan(new StringBuilder(), StringBuilder::append)
+          .subscribe(System.out::println);
+
+        System.out.println();
+        System.out.println("------GroubBy------------");
+        Observable.from(numbers)
+          .groupBy(i -> 0 == (i % 2) ? "EVEN" : "ODD")
+          .subscribe((group) -> group.subscribe((number) -> {
+              System.out.println(group.getKey() + " : " + number);
+          }));
+
+        System.out.println();
+        System.out.println("-------Filter-----------");
+        Observable.from(numbers)
+          .filter(i -> (i % 2 == 1))
+          .subscribe(System.out::println);
+
+        System.out.println("------DefaultIfEmpty------------");
+        Observable.empty()
+          .defaultIfEmpty("Observable is empty")
+          .subscribe(System.out::println);
+
+        System.out.println("------DefaultIfEmpty-2-----------");
+        Observable.from(letters)
+          .defaultIfEmpty("Observable is empty")
+          .first()
+          .subscribe(System.out::println);
+
+        System.out.println("-------TakeWhile-----------");
+        Observable.from(numbers)
+          .takeWhile(i -> i < 5)
+          .subscribe(System.out::println);
+    }
+}
@@ -0,0 +1,30 @@
+package com.baeldung.rxjava;
+
+import rx.Observable;
+
+public class ResourceManagement {
+
+    public static void main(String[] args) {
+
+        Observable<Character> values = Observable.using(
+          () -> {
+              String resource = "MyResource";
+              System.out.println("Leased: " + resource);
+              return resource;
+          },
+          r -> Observable.create(o -> {
+              for (Character c : r.toCharArray()) {
+                  o.onNext(c);
+              }
+              o.onCompleted();
+          }),
+          r -> System.out.println("Disposed: " + r)
+        );
+
+        values.subscribe(
+          System.out::println,
+          System.out::println
+        );
+    }
+}
+
@@ -0,0 +1,18 @@
+package com.baeldung.rxjava;
+
+import rx.Observable;
+import rx.Single;
+
+public class SingleImpl {
+
+    public static void main(String[] args) {
+
+        Single<String> single = Observable.just("Hello")
+          .toSingle()
+          .doOnSuccess(System.out::print)
+          .doOnError(e -> {
+              throw new RuntimeException(e.getMessage());
+          });
+        single.subscribe();
+    }
+}
@@ -0,0 +1,73 @@
+package com.baeldung.rxjava;
+
+import rx.Observer;
+import rx.subjects.PublishSubject;
+
+public class SubjectImpl {
+
+    static Integer subscriber1 = 0;
+    static Integer subscriber2 = 0;
+
+    private static Integer subjectMethod() {
+        PublishSubject<Integer> subject = PublishSubject.create();
+
+        subject.subscribe(getFirstObserver());
+
+        subject.onNext(1);
+        subject.onNext(2);
+        subject.onNext(3);
+
+        subject.subscribe(getSecondObserver());
+
+        subject.onNext(4);
+        subject.onCompleted();
+        return subscriber1 + subscriber2;
+    }
+
+
+    static Observer<Integer> getFirstObserver() {
+        return new Observer<Integer>() {
+
+            @Override
+            public void onNext(Integer value) {
+                subscriber1 += value;
+                System.out.println("Subscriber1: " + value);
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                System.out.println("error");
+            }
+
+            @Override
+            public void onCompleted() {
+                System.out.println("Subscriber1 completed");
+            }
+        };
+    }
+
+    static Observer<Integer> getSecondObserver() {
+        return new Observer<Integer>() {
+
+            @Override
+            public void onNext(Integer value) {
+                subscriber2 += value;
+                System.out.println("Subscriber2: " + value);
+            }
+
+            @Override
+            public void onError(Throwable e) {
+                System.out.println("error");
+            }
+
+            @Override
+            public void onCompleted() {
+                System.out.println("Subscriber2 completed");
+            }
+        };
+    }
+
+    public static void main(String[] args) throws InterruptedException {
+        System.out.println(subjectMethod());
+    }
+}
@@ -0,0 +1,13 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<configuration>
+    <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+        <encoder>
+            <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n
+            </pattern>
+        </encoder>
+    </appender>
+
+    <root level="INFO">
+        <appender-ref ref="STDOUT" />
+    </root>
+</configuration>