BAEL-3091: The Prototype Pattern in Java (changed code based on valid comments from a reader)

rxjava/src/main/java/com/baeldung/rxjava/ComputeFunction.java

@@ -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();
+        }
+    }
+}

rxjava/src/main/java/com/baeldung/rxjava/ConnectableObservableImpl.java

@@ -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");
+    }
+}

rxjava/src/main/java/com/baeldung/rxjava/MultipleSubscribersColdObs.java

@@ -0,0 +1,59 @@
+package com.baeldung.rxjava;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import rx.Observable;
+import rx.Subscription;
+import rx.observables.ConnectableObservable;
+import rx.subscriptions.Subscriptions;
+
+public class MultipleSubscribersColdObs {
+    private static final Logger LOGGER = LoggerFactory.getLogger(MultipleSubscribersColdObs.class);
+
+    public static void main(String[] args) throws InterruptedException {
+        defaultBehaviour();
+        // subscribeBeforeConnect();
+
+    }
+
+    private static void defaultBehaviour() {
+        Observable obs = getObservable();
+
+        LOGGER.info("Subscribing");
+        Subscription s1 = obs.subscribe(i -> LOGGER.info("subscriber#1 is printing " + i));
+        Subscription s2 = obs.subscribe(i -> LOGGER.info("subscriber#2 is printing " + i));
+
+        s1.unsubscribe();
+        s2.unsubscribe();
+    }
+
+    private static void subscribeBeforeConnect() throws InterruptedException {
+        ConnectableObservable obs = getObservable().publish();
+
+        LOGGER.info("Subscribing");
+        obs.subscribe(i -> LOGGER.info("subscriber #1 is printing " + i));
+        obs.subscribe(i -> LOGGER.info("subscriber #2 is printing " + i));
+        Thread.sleep(1000);
+        LOGGER.info("Connecting");
+        Subscription s = obs.connect();
+        s.unsubscribe();
+
+    }
+
+    private static Observable getObservable() {
+        return Observable.create(subscriber -> {
+            subscriber.onNext(gettingValue(1));
+            subscriber.onNext(gettingValue(2));
+
+            subscriber.add(Subscriptions.create(() -> {
+                LOGGER.info("Clear resources");
+            }));
+        });
+    }
+
+    private static Integer gettingValue(int i) {
+        LOGGER.info("Getting " + i);
+        return i;
+    }
+}

rxjava/src/main/java/com/baeldung/rxjava/MultipleSubscribersHotObs.java

@@ -0,0 +1,152 @@
+package com.baeldung.rxjava;
+
+import java.awt.Color;
+import java.awt.Dimension;
+import java.awt.event.MouseAdapter;
+import java.awt.event.MouseEvent;
+import java.awt.event.MouseListener;
+import java.lang.reflect.InvocationTargetException;
+
+import javax.swing.JFrame;
+import rx.Observable;
+import rx.Subscription;
+import rx.observables.ConnectableObservable;
+import rx.subscriptions.Subscriptions;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class MultipleSubscribersHotObs {
+    private static final Logger LOGGER = LoggerFactory.getLogger(MultipleSubscribersHotObs.class);
+    private static JFrame frame;
+
+    public static void main(String[] args) throws InterruptedException, InvocationTargetException {
+
+        javax.swing.SwingUtilities.invokeAndWait(new Runnable() {
+            public void run() {
+                createAndShowGUI();
+            }
+        });
+
+        defaultBehaviour();
+        // subscribeBeforeConnect();
+        // connectBeforeSubscribe();
+        // autoConnectAndSubscribe();
+        // refCountAndSubscribe();
+    }
+
+    private static void createAndShowGUI() {
+        frame = new JFrame("Hot Observable Demo");
+        frame.setDefaultCloseOperation(JFrame.EXIT_ON_CLOSE);
+        frame.getContentPane().setBackground(Color.GRAY);
+        frame.setPreferredSize(new Dimension(500, 500));
+        frame.pack();
+        frame.setVisible(true);
+    }
+
+    public static void defaultBehaviour() throws InterruptedException {
+        Observable obs = getObservable();
+
+        LOGGER.info("subscribing #1");
+        Subscription subscription1 = obs.subscribe((i) -> LOGGER.info("subscriber#1 is printing x-coordinate " + i));
+
+        Thread.sleep(1000);
+        LOGGER.info("subscribing #2");
+        Subscription subscription2 = obs.subscribe((i) -> LOGGER.info("subscriber#2 is printing x-coordinate " + i));
+        Thread.sleep(1000);
+        LOGGER.info("unsubscribe#1");
+        subscription1.unsubscribe();
+        Thread.sleep(1000);
+        LOGGER.info("unsubscribe#2");
+        subscription2.unsubscribe();
+    }
+
+    public static void subscribeBeforeConnect() throws InterruptedException {
+
+        ConnectableObservable obs = getObservable().publish();
+
+        LOGGER.info("subscribing #1");
+        Subscription subscription1 = obs.subscribe((i) -> LOGGER.info("subscriber#1 is printing x-coordinate " + i));
+        Thread.sleep(1000);
+        LOGGER.info("subscribing #2");
+        Subscription subscription2 = obs.subscribe((i) -> LOGGER.info("subscriber#2 is printing x-coordinate " + i));
+        Thread.sleep(1000);
+        LOGGER.info("connecting:");
+        Subscription s = obs.connect();
+        Thread.sleep(1000);
+        LOGGER.info("unsubscribe connected");
+        s.unsubscribe();
+
+    }
+
+    public static void connectBeforeSubscribe() throws InterruptedException {
+
+        ConnectableObservable obs = getObservable().doOnNext(x -> LOGGER.info("saving " + x)).publish();
+        LOGGER.info("connecting:");
+        Subscription s = obs.connect();
+        Thread.sleep(1000);
+        LOGGER.info("subscribing #1");
+        obs.subscribe((i) -> LOGGER.info("subscriber#1 is printing x-coordinate " + i));
+        Thread.sleep(1000);
+        LOGGER.info("subscribing #2");
+        obs.subscribe((i) -> LOGGER.info("subscriber#2 is printing x-coordinate " + i));
+        Thread.sleep(1000);
+        s.unsubscribe();
+
+    }
+
+    public static void autoConnectAndSubscribe() throws InterruptedException {
+        Observable obs = getObservable().doOnNext(x -> LOGGER.info("saving " + x)).publish().autoConnect();
+
+        LOGGER.info("autoconnect()");
+        Thread.sleep(1000);
+        LOGGER.info("subscribing #1");
+        Subscription s1 = obs.subscribe((i) -> LOGGER.info("subscriber#1 is printing x-coordinate " + i));
+        Thread.sleep(1000);
+        LOGGER.info("subscribing #2");
+        Subscription s2 = obs.subscribe((i) -> LOGGER.info("subscriber#2 is printing x-coordinate " + i));
+
+        Thread.sleep(1000);
+        LOGGER.info("unsubscribe 1");
+        s1.unsubscribe();
+        Thread.sleep(1000);
+        LOGGER.info("unsubscribe 2");
+        s2.unsubscribe();
+    }
+
+    public static void refCountAndSubscribe() throws InterruptedException {
+        Observable obs = getObservable().doOnNext(x -> LOGGER.info("saving " + x)).publish().refCount();
+
+        LOGGER.info("refcount()");
+        Thread.sleep(1000);
+        LOGGER.info("subscribing #1");
+        Subscription subscription1 = obs.subscribe((i) -> LOGGER.info("subscriber#1 is printing x-coordinate " + i));
+        Thread.sleep(1000);
+        LOGGER.info("subscribing #2");
+        Subscription subscription2 = obs.subscribe((i) -> LOGGER.info("subscriber#2 is printing x-coordinate " + i));
+
+        Thread.sleep(1000);
+        LOGGER.info("unsubscribe#1");
+        subscription1.unsubscribe();
+        Thread.sleep(1000);
+        LOGGER.info("unsubscribe#2");
+        subscription2.unsubscribe();
+
+    }
+
+    private static Observable getObservable() {
+        return Observable.create(subscriber -> {
+            frame.addMouseListener(new MouseAdapter() {
+                @Override
+                public void mouseClicked(MouseEvent e) {
+                    subscriber.onNext(e.getX());
+                }
+            });
+            subscriber.add(Subscriptions.create(() -> {
+                LOGGER.info("Clear resources");
+                for (MouseListener listener : frame.getListeners(MouseListener.class)) {
+                    frame.removeMouseListener(listener);
+                }
+            }));
+        });
+    }
+}

rxjava/src/main/java/com/baeldung/rxjava/ObservableImpl.java

@@ -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);
+    }
+}

rxjava/src/main/java/com/baeldung/rxjava/ResourceManagement.java

@@ -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
+        );
+    }
+}
+

rxjava/src/main/java/com/baeldung/rxjava/SingleImpl.java

@@ -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();
+    }
+}

rxjava/src/main/java/com/baeldung/rxjava/SubjectImpl.java

@@ -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());
+    }
+}

rxjava/src/main/java/com/baeldung/rxjava/jdbc/Connector.java

@@ -0,0 +1,13 @@
+package com.baeldung.rxjava.jdbc;
+
+import com.github.davidmoten.rx.jdbc.ConnectionProvider;
+import com.github.davidmoten.rx.jdbc.ConnectionProviderFromUrl;
+
+class Connector {
+
+    private static final String DB_CONNECTION = "jdbc:h2:mem:test;DB_CLOSE_DELAY=-1";
+    private static final String DB_USER = "";
+    private static final String DB_PASSWORD = "";
+
+    static final ConnectionProvider connectionProvider = new ConnectionProviderFromUrl(DB_CONNECTION, DB_USER, DB_PASSWORD);
+}

rxjava/src/main/java/com/baeldung/rxjava/jdbc/Employee.java

@@ -0,0 +1,13 @@
+package com.baeldung.rxjava.jdbc;
+
+import com.github.davidmoten.rx.jdbc.annotations.Column;
+
+public interface Employee {
+
+    @Column("id")
+    int id();
+
+    @Column("name")
+    String name();
+    
+}

rxjava/src/main/java/com/baeldung/rxjava/jdbc/Manager.java

@@ -0,0 +1,28 @@
+package com.baeldung.rxjava.jdbc;
+
+public class Manager {
+
+    private int id;
+    private String name;
+
+    public Manager(int id, String name) {
+        this.id = id;
+        this.name = name;
+    }
+
+    public int getId() {
+        return id;
+    }
+
+    public void setId(int id) {
+        this.id = id;
+    }
+
+    public String getName() {
+        return name;
+    }
+
+    public void setName(String name) {
+        this.name = name;
+    }
+}

rxjava/src/main/java/com/baeldung/rxjava/jdbc/Utils.java

@@ -0,0 +1,16 @@
+package com.baeldung.rxjava.jdbc;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.StringWriter;
+
+import org.apache.commons.io.IOUtils;
+
+class Utils {
+
+    static String getStringFromInputStream(InputStream input) throws IOException {
+        StringWriter writer = new StringWriter();
+        IOUtils.copy(input, writer, "UTF-8");
+        return writer.toString();
+    }
+}

rxjava/src/main/java/com/baeldung/rxjava/operator/ToCleanString.java

@@ -0,0 +1,42 @@
+package com.baeldung.rxjava.operator;
+
+import rx.Observable.Operator;
+import rx.Subscriber;
+
+public class ToCleanString implements Operator<String, String> {
+
+    public static ToCleanString toCleanString() {
+        return new ToCleanString();
+    }
+
+    private ToCleanString() {
+        super();
+    }
+
+    @Override
+    public Subscriber<? super String> call(final Subscriber<? super String> subscriber) {
+        return new Subscriber<String>(subscriber) {
+            @Override
+            public void onCompleted() {
+                if (!subscriber.isUnsubscribed()) {
+                    subscriber.onCompleted();
+                }
+            }
+
+            @Override
+            public void onError(Throwable t) {
+                if (!subscriber.isUnsubscribed()) {
+                    subscriber.onError(t);
+                }
+            }
+
+            @Override
+            public void onNext(String item) {
+                if (!subscriber.isUnsubscribed()) {
+                    final String result = item.replaceAll("[^A-Za-z0-9]", "");
+                    subscriber.onNext(result);
+                }
+            }
+        };
+    }
+}

rxjava/src/main/java/com/baeldung/rxjava/operator/ToLength.java

@@ -0,0 +1,20 @@
+package com.baeldung.rxjava.operator;
+
+import rx.Observable;
+import rx.Observable.Transformer;
+
+public class ToLength implements Transformer<String, Integer> {
+
+    public static ToLength toLength() {
+        return new ToLength();
+    }
+
+    private ToLength() {
+        super();
+    }
+
+    @Override
+    public Observable<Integer> call(Observable<String> source) {
+        return source.map(String::length);
+    }
+}