add Person Observable

Author: meowpunch

pom.xml

@@ -32,6 +32,21 @@
             <artifactId>javatuples</artifactId>
             <version>1.2</version>
         </dependency>
+
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+            <version>1.18.24</version>
+            <scope>provided</scope>
+        </dependency>
+
+        <dependency>
+            <groupId>org.reactivestreams</groupId>
+            <artifactId>reactive-streams</artifactId>
+            <version>1.0.4</version>
+        </dependency>
+
+
     </dependencies>
 
 </project>

src/main/java/concurrency/reactive/Ob.java

@@ -2,28 +2,33 @@
 
 import java.util.Observable;
 import java.util.Observer;
-import java.util.concurrent.Executor;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
-import java.util.stream.IntStream;
 import java.util.stream.Stream;
+import lombok.AllArgsConstructor;
+import lombok.ToString;
 
 @SuppressWarnings("deprecation")
 public class Ob {
 
-
   /*
-  *   Observable : source
-  *   Observer :
-  *
-  *
-  */
+   *   Observable : Publisher - data source
+   *   Observer : Subscriber - consume data
+   *
+   *   Limitation
+   *   1. how to know it's completed - onComplete
+   *   2. how to handle error - onError
+   */
   public static void main(String[] args) {
 
+    testObserver(new IntObservable());
+    testObserver(new Person("Jinhoon", 29));
+  }
+
+  private static void testObserver(RunnableObservable runnableObservable) {
     Observer ob1 = (o, arg) -> System.out.println(Thread.currentThread().getName() + " ob1 consume " + arg); // override `void update` method
     Observer ob2 = (o, arg) -> System.out.println(Thread.currentThread().getName() + " ob2 consume " + arg);
 
-    IntObservable runnableObservable = new IntObservable();
     runnableObservable.addObserver(ob1);
     runnableObservable.addObserver(ob2);
 
@@ -34,11 +39,15 @@ public static void main(String[] args) {
     executorService.shutdown();
   }
 
-  static class IntObservable extends Observable implements Runnable {
+  abstract static class RunnableObservable extends Observable implements Runnable {
+
+  }
+
+  static class IntObservable extends RunnableObservable {
 
     // override from Runnable
     public void run() {
-      Stream.iterate(0,  i -> i+1).limit(10)
+      Stream.iterate(1, i -> i + 1).limit(10)
           .forEach(
               i -> {
                 setChanged();
@@ -57,4 +66,24 @@ public void run() {
     }
   }
 
+  @ToString
+  @AllArgsConstructor
+  static class Person extends RunnableObservable {
+
+    String name;
+    int age;
+
+    void setAge(int newAge) {
+      if (this.age != newAge) {
+        this.age = newAge;
+        setChanged();
+        notifyObservers(this);
+      }
+    }
+
+    public void run() {
+      setAge(30);
+      setAge(31);
+    }
+  }
 }