added the reactive stream examples using java 9

README.md

-** Java 9 Feature  **
-
-1.ReactiveStreams
-2.DiamondOperator
-3.FactoryMethodsForUnmodifiableCollections
-4.PrivateMethods
-5.ProcessAPI
-6.StreamApI
-7.TryWithResourcesEnhancements
\ No newline at end of file
+#Java 9 Feature
+
+
+    ReactiveStreams
+    DiamondOperator
+    FactoryMethodsForUnmodifiableCollections
+    PrivateMethods
+    ProcessAPI
+    StreamApI
+    TryWithResourcesEnhancements
+    
+ Java 9 is providing standard of the Reactive programming 
+ 
+  ___Publisher
+  Subscriber
+  Processor
+  Subscription___
+  
+  
+    asynchronous
+    coldpublisher
+    drop
+    error
+    hotpublisher
+    rxjavaflowable
+    slowsubscriber
+    submissionpublisher
+    synchronous
+
+

build.gradle

@@ -11,8 +11,12 @@ repositories {
     mavenCentral()
 }
 
+
+
 dependencies {
     compileOnly 'org.projectlombok:lombok:1.18.12'
+    compile group: 'io.reactivex.rxjava2', name: 'rxjava', version: '2.2.16'
+
 
     testCompile group: 'junit', name: 'junit', version: '4.12'
 }

src/main/java/com/nisum/java9Features/DiamondOperator/MyGenClass.java → src/main/java/com/nisum/java9Features/diamondoperator/MyGenClass.java

-package com.nisum.java9Features.DiamondOperator;
+package com.nisum.java9Features.diamondoperator;
 
 import java.util.logging.Logger;
 

src/main/java/com/nisum/java9Features/DiamondOperator/Test.java → src/main/java/com/nisum/java9Features/diamondoperator/Test.java

-package com.nisum.java9Features.DiamondOperator;
+package com.nisum.java9Features.diamondoperator;
 
 public class Test {
     public static void main(String[] args) {

src/main/java/com/nisum/java9Features/FactoryMethodsForUnmodifiableCollections/Employee.java → src/main/java/com/nisum/java9Features/factorymethodsforunmodifiablecollections/Employee.java

-package com.nisum.java9Features.FactoryMethodsForUnmodifiableCollections;
+package com.nisum.java9Features.factorymethodsforunmodifiablecollections;
 
 public class Employee {
     private int eno;

src/main/java/com/nisum/java9Features/FactoryMethodsForUnmodifiableCollections/UnmodifiableList.java → src/main/java/com/nisum/java9Features/factorymethodsforunmodifiablecollections/UnmodifiableList.java

-package com.nisum.java9Features.FactoryMethodsForUnmodifiableCollections;
+package com.nisum.java9Features.factorymethodsforunmodifiablecollections;
 
 
 import java.util.List;

src/main/java/com/nisum/java9Features/FactoryMethodsForUnmodifiableCollections/UnmodifiableMap.java → src/main/java/com/nisum/java9Features/factorymethodsforunmodifiablecollections/UnmodifiableMap.java

-package com.nisum.java9Features.FactoryMethodsForUnmodifiableCollections;
+package com.nisum.java9Features.factorymethodsforunmodifiablecollections;
 
 
 

src/main/java/com/nisum/java9Features/FactoryMethodsForUnmodifiableCollections/UnmodifiableSet.java → src/main/java/com/nisum/java9Features/factorymethodsforunmodifiablecollections/UnmodifiableSet.java

-package com.nisum.java9Features.FactoryMethodsForUnmodifiableCollections;
+package com.nisum.java9Features.factorymethodsforunmodifiablecollections;
 
 
 import java.util.Set;

src/main/java/com/nisum/java9Features/PrivateMethods/Java9Interf.java → src/main/java/com/nisum/java9Features/privatemethod/Java9Interf.java

-package com.nisum.java9Features.PrivateMethods;
+package com.nisum.java9Features.privatemethod;
 
 import java.util.logging.Logger;
 

src/main/java/com/nisum/java9Features/PrivateMethods/Java9InterfStatic.java → src/main/java/com/nisum/java9Features/privatemethod/Java9InterfStatic.java

-package com.nisum.java9Features.PrivateMethods;
+package com.nisum.java9Features.privatemethod;
 
 import java.util.logging.Logger;
 

src/main/java/com/nisum/java9Features/PrivateMethods/PrivateStaticMethod.java → src/main/java/com/nisum/java9Features/privatemethod/PrivateStaticMethod.java

-package com.nisum.java9Features.PrivateMethods;
+package com.nisum.java9Features.privatemethod;
 
 public class PrivateStaticMethod implements Java9InterfStatic {
     public static void main(String[] args)

src/main/java/com/nisum/java9Features/ProcessAPI/CurrentProcessInfo.java → src/main/java/com/nisum/java9Features/processAPI/CurrentProcessInfo.java

-package com.nisum.java9Features.ProcessAPI;
+package com.nisum.java9Features.processAPI;
 
 import java.util.logging.Logger;
 

src/main/java/com/nisum/java9Features/ProcessAPI/DestroyProcessFromProcessBuilder.java → src/main/java/com/nisum/java9Features/processAPI/DestroyProcessFromProcessBuilder.java

-package com.nisum.java9Features.ProcessAPI;
-
-import lombok.extern.slf4j.Slf4j;
+package com.nisum.java9Features.processAPI;
 
 import java.util.logging.Logger;
 

src/main/java/com/nisum/java9Features/ProcessAPI/NotepadFromProcessBuilder.java → src/main/java/com/nisum/java9Features/processAPI/NotepadFromProcessBuilder.java

-package com.nisum.java9Features.ProcessAPI;
+package com.nisum.java9Features.processAPI;
 
 public class NotepadFromProcessBuilder {
     public static void main(String[] args) throws Exception {

src/main/java/com/nisum/java9Features/ProcessAPI/ProcessBasedOnID.java → src/main/java/com/nisum/java9Features/processAPI/ProcessBasedOnID.java

-package com.nisum.java9Features.ProcessAPI;
+package com.nisum.java9Features.processAPI;
 
 
 import java.util.Optional;

src/main/java/com/nisum/java9Features/ProcessAPI/ProcessID.java → src/main/java/com/nisum/java9Features/processAPI/ProcessID.java

-package com.nisum.java9Features.ProcessAPI;
+package com.nisum.java9Features.processAPI;
 
 import java.util.logging.Logger;
 

src/main/java/com/nisum/java9Features/reactiveapi/asynchronous/Sample.java

+package com.nisum.java9Features.reactiveapi.asynchronous;
+
+import io.reactivex.BackpressureStrategy;
+import io.reactivex.Flowable;
+import io.reactivex.FlowableEmitter;
+import io.reactivex.schedulers.Schedulers;
+
+/**
+ * creating asynchronous subscriber
+ * by putting delay
+ * Here back pressure is handled using BackpressureStrategy.BUFFER
+ */
+public class Sample {
+  public static void main(String[] args) {
+    Flowable.<Integer>create(emitter -> emit(emitter), BackpressureStrategy.BUFFER)
+        .map(data -> data * 1.0)
+        .filter(data -> data > 4)
+        .observeOn(Schedulers.io())
+        .subscribe(Sample::printIt,
+            err -> System.out.println("ERROR: " + err),
+            () -> System.out.println("DONE"));
+  }
+
+  private static void printIt(Double value) throws InterruptedException {
+    System.out.println(value + " -- " + Thread.currentThread());
+    Thread.sleep(1000);
+  }
+
+  private static void emit(FlowableEmitter<Integer> emitter) throws InterruptedException {
+    int count = 0;
+
+    while(count < 20) {
+      System.out.println("emitting " + count + " --" + Thread.currentThread());
+      emitter.onNext(count++);
+
+      Thread.sleep(500);
+    }
+  }
+}
\ No newline at end of file

src/main/java/com/nisum/java9Features/reactiveapi/coldpublisher/Sample.java

+package com.nisum.java9Features.reactiveapi.coldpublisher;
+
+import io.reactivex.Flowable;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Cold publisher will have defferent different session for different subscribers;
+ */
+public class Sample {
+  public static void main(String[] args) throws InterruptedException {
+    Flowable<Long> interval =
+        Flowable.interval(1, TimeUnit.SECONDS);
+
+    interval.subscribe(data -> printMessage("s1:" + data));
+
+    Thread.sleep(5000);
+
+    interval.subscribe(data -> printMessage("s2:" + data));
+
+    Thread.sleep(10000);
+  }
+
+  private static void printMessage(String message) {
+    System.out.println(message);
+  }
+}
\ No newline at end of file

src/main/java/com/nisum/java9Features/reactiveapi/drop/Sample.java

+package com.nisum.java9Features.reactiveapi.drop;
+
+import io.reactivex.BackpressureStrategy;
+import io.reactivex.Flowable;
+import io.reactivex.FlowableEmitter;
+import io.reactivex.schedulers.Schedulers;
+
+
+/**
+ * BackpressureStrategy.DROP
+ * here all the previous data produced by the publisher is droped.
+ *
+ */
+public class Sample {
+  public static void main(String[] args) {
+    Flowable.<Integer>create(emitter -> emit(emitter), BackpressureStrategy.DROP)
+        .map(data -> data * 1.0)
+        .filter(data -> data > 4)
+        .observeOn(Schedulers.io(), false, 2)
+        .subscribe(Sample::printIt,
+            err -> System.out.println("ERROR: " + err),
+            () -> System.out.println("DONE"));
+  }
+
+  private static void printIt(Double value) throws InterruptedException {
+    System.out.println(value + " -- " + Thread.currentThread());
+    Thread.sleep(2000);
+  }
+
+  private static void emit(FlowableEmitter<Integer> emitter) throws InterruptedException {
+    int count = 0;
+
+    while(count < 20) {
+      System.out.println("emitting " + count + " --" + Thread.currentThread());
+      emitter.onNext(count++);
+
+      Thread.sleep(500);
+    }
+
+    System.out.println("DONE emitting");
+    Thread.sleep(10000);
+  }
+}
\ No newline at end of file

src/main/java/com/nisum/java9Features/reactiveapi/error/Sample.java

+package com.nisum.java9Features.reactiveapi.error;
+
+import io.reactivex.BackpressureStrategy;
+import io.reactivex.Flowable;
+import io.reactivex.FlowableEmitter;
+
+/**
+ * handling error using the reactively;
+ * Exceptions and the functional programming are mutually exclusive;
+ * exceptions are imparative programming idea
+ * so in reactive api it  deals it with downstream;
+ *
+ */
+public class Sample {
+  public static void main(String[] args) {
+    Flowable.<Integer>create(emitter -> emit(emitter), BackpressureStrategy.BUFFER)
+        .map(data -> data * 1.0)
+        .filter(data -> data > 4)
+        .subscribe(System.out::println,
+            err -> System.out.println("ERROR: " + err),
+            () -> System.out.println("DONE"));
+  }
+
+  private static void emit(FlowableEmitter<Integer> emitter) throws InterruptedException {
+    int count = 0;
+
+    while(count < 20) {
+      emitter.onNext(count++);
+
+      if(count == 7) throw new RuntimeException("something went wrong");
+      Thread.sleep(500);
+    }
+  }
+}
\ No newline at end of file

src/main/java/com/nisum/java9Features/reactiveapi/hotpublisher/Sample.java

+package com.nisum.java9Features.reactiveapi.hotpublisher;
+
+import io.reactivex.Flowable;
+
+import java.util.concurrent.TimeUnit;
+
+/**
+ * hot publisher uses to share the subscription amoung all the subscriber;
+ * Any subscriber joining late will miss some amount of data;
+ */
+public class Sample {
+  public static void main(String[] args) throws InterruptedException {
+    Flowable<Long> interval =
+         Flowable.interval(1, TimeUnit.SECONDS)
+        .share();
+
+    interval.subscribe(data -> printMessage("s1:" + data));
+
+    Thread.sleep(5000);
+
+    interval.subscribe(data -> printMessage("s2:" + data));
+
+    Thread.sleep(10000);
+  }
+
+  private static void printMessage(String message) {
+    System.out.println(message);
+  }
+}
\ No newline at end of file

src/main/java/com/nisum/java9Features/reactiveapi/latest/Sample.java

+package com.nisum.java9Features.reactiveapi.latest;
+
+import io.reactivex.BackpressureStrategy;
+import io.reactivex.Flowable;
+import io.reactivex.FlowableEmitter;
+import io.reactivex.schedulers.Schedulers;
+
+/**
+ *
+ * BackpressureStrategy.LATES
+ * It is one of the backpressure strategy to handle the flow of the data;
+ * It gives the latest data to the subscriber rather if any subscriber subscribes;
+ */
+public class Sample {
+  public static void main(String[] args) {
+    Flowable.<Integer>create(emitter -> emit(emitter), BackpressureStrategy.LATEST)
+        .map(data -> data * 1.0)
+        .filter(data -> data > 4)
+        .observeOn(Schedulers.io(), false, 2)
+        .subscribe(Sample::printIt,
+            err -> System.out.println("ERROR: " + err),
+            () -> System.out.println("DONE"));
+  }
+
+  private static void printIt(Double value) throws InterruptedException {
+    System.out.println(value + " -- " + Thread.currentThread());
+    Thread.sleep(2000);
+  }
+
+  private static void emit(FlowableEmitter<Integer> emitter) throws InterruptedException {
+    int count = 0;
+
+    while(count < 20) {
+      System.out.println("emitting " + count + " --" + Thread.currentThread());
+      emitter.onNext(count++);
+
+      Thread.sleep(500);
+    }
+
+    System.out.println("DONE emitting");
+    Thread.sleep(10000);
+  }
+}
\ No newline at end of file

src/main/java/com/nisum/java9Features/reactiveapi/rxjavaflowable/Sample.java

+package com.nisum.java9Features.reactiveapi.rxjavaflowable;
+
+import io.reactivex.BackpressureStrategy;
+import io.reactivex.Flowable;
+import io.reactivex.FlowableEmitter;
+
+public class Sample {
+  public static void main(String[] args) {
+    Flowable.<Integer>create(emitter -> emit(emitter), BackpressureStrategy.BUFFER)
+        .map(data -> data * 1.0)
+        .filter(data -> data > 4)
+        .subscribe(System.out::println);
+  }
+
+  private static void emit(FlowableEmitter<Integer> emitter) throws InterruptedException {
+    int count = 0;
+
+    while(count < 20) {
+      emitter.onNext(count++);
+      Thread.sleep(500);
+    }
+  }
+}
\ No newline at end of file

src/main/java/com/nisum/java9Features/reactiveapi/slowsubscriber/Sample.java

+package com.nisum.java9Features.reactiveapi.slowsubscriber;
+
+import io.reactivex.BackpressureStrategy;
+import io.reactivex.Flowable;
+import io.reactivex.FlowableEmitter;
+
+public class Sample {
+  public static void main(String[] args) {
+    Flowable.<Integer>create(emitter -> emit(emitter), BackpressureStrategy.BUFFER)
+        .map(data -> data * 1.0)
+        .filter(data -> data > 4)
+        .subscribe(Sample::printIt,
+            err -> System.out.println("ERROR: " + err),
+            () -> System.out.println("DONE"));
+  }
+
+  private static void printIt(Double value) throws InterruptedException {
+    System.out.println(value + " -- " + Thread.currentThread());
+    Thread.sleep(1000);
+  }
+
+  private static void emit(FlowableEmitter<Integer> emitter) throws InterruptedException {
+    int count = 0;
+
+    while(count < 20) {
+      System.out.println("emitting " + count + " --" + Thread.currentThread());
+      emitter.onNext(count++);
+
+      Thread.sleep(500);
+    }
+  }
+}
\ No newline at end of file

src/main/java/com/nisum/java9Features/reactiveapi/submissionpublisher/PrintResultSubscriber.java

+package com.nisum.java9Features.reactiveapi.submissionpublisher;
+
+import java.util.concurrent.Flow;
+
+/**
+ * creating the subscriber implementing the Flow.Subscriber given by jdk 9
+ * here we are maintaining the flow of data as well using request method of Subscription Interface;;
+ */
+
+public class PrintResultSubscriber implements Flow.Subscriber<String> {
+  private Flow.Subscription subscription;
+
+  @Override
+  public void onSubscribe(Flow.Subscription subscription) {
+    System.out.println("subscribed...");
+    this.subscription = subscription;
+    subscription.request(5);
+  }
+
+  @Override
+  public void onNext(String data) {
+    System.out.println(data);
+    subscription.request(1);
+    if(data.equals("CoronaPatient9")) subscription.cancel();
+  }
+
+  @Override
+  public void onError(Throwable throwable) {
+
+  }
+
+  @Override
+  public void onComplete() {
+    
+  }
+}
\ No newline at end of file

src/main/java/com/nisum/java9Features/reactiveapi/submissionpublisher/SampleUserOfSubmissionPublisher.java

+package com.nisum.java9Features.reactiveapi.submissionpublisher;
+
+
+
+import java.util.concurrent.SubmissionPublisher;
+
+/**
+ *Use of SubmissionPublisher provided by java 9 Flow API
+ * it provides submit method  to publish the data
+ * It is implementing the Flow.Publisher
+ * x
+ */
+public class SampleUserOfSubmissionPublisher {
+  public static void main(String[] args) throws InterruptedException {
+    SubmissionPublisher<String> publisher =
+        new SubmissionPublisher<>();
+
+    PrintResultSubscriber subscriber = new PrintResultSubscriber();
+    publisher.subscribe(subscriber);
+
+    int count = 0;
+    while(count < 20) {
+      Thread.sleep(500);
+      if(!publisher.hasSubscribers()) {
+        System.out.println("Subscription is removed:  as we are getting more than nine corona patients in hyderabd");
+        break;
+      }
+      String res="CoronaPatient"+count++;
+      publisher.submit(res);
+    }
+
+    Thread.sleep(2000);
+  }
+}
\ No newline at end of file

src/main/java/com/nisum/java9Features/reactiveapi/synchronous/Sample.java

+package com.nisum.java9Features.reactiveapi.synchronous;
+
+import io.reactivex.BackpressureStrategy;
+import io.reactivex.Flowable;
+import io.reactivex.FlowableEmitter;
+//sync by default
+public class Sample {
+  public static void main(String[] args) {
+    Flowable.<Integer>create(emitter -> emit(emitter), BackpressureStrategy.BUFFER)
+        .map(data -> data * 1.0)
+        .filter(data -> data > 4)
+        .subscribe(Sample::printIt,
+            err -> System.out.println("ERROR: " + err),
+            () -> System.out.println("DONE"));
+  }
+
+  private static void printIt(Double value) {
+    System.out.println(value + " -- " + Thread.currentThread());
+  }
+
+  private static void emit(FlowableEmitter<Integer> emitter) throws InterruptedException {
+    int count = 0;
+
+    while(count < 20) {
+      System.out.println("emitting " + count + " --" + Thread.currentThread());
+      emitter.onNext(count++);
+
+      Thread.sleep(500);
+    }
+  }
+}
\ No newline at end of file

src/main/java/com/nisum/java9Features/ReactiveStreams/EmpHelper.java → src/main/java/com/nisum/java9Features/reactivestreams/EmpHelper.java

-package com.nisum.java9Features.ReactiveStreams;
+package com.nisum.java9Features.reactivestreams;
 
 
 import java.util.ArrayList;

src/main/java/com/nisum/java9Features/ReactiveStreams/Employee.java → src/main/java/com/nisum/java9Features/reactivestreams/Employee.java

-package com.nisum.java9Features.ReactiveStreams;
+package com.nisum.java9Features.reactivestreams;
 
 public class Employee {
     private int id;

src/main/java/com/nisum/java9Features/ReactiveStreams/MyReactiveApp.java → src/main/java/com/nisum/java9Features/reactivestreams/MyReactiveApp.java

-package com.nisum.java9Features.ReactiveStreams;
+package com.nisum.java9Features.reactivestreams;
 
 
 import java.util.List;

src/main/java/com/nisum/java9Features/ReactiveStreams/MySubscriber.java → src/main/java/com/nisum/java9Features/reactivestreams/MySubscriber.java

-package com.nisum.java9Features.ReactiveStreams;
+package com.nisum.java9Features.reactivestreams;
 
 
 import java.util.concurrent.Flow.Subscriber;

src/main/java/com/nisum/java9Features/StreamApi/TestDropWhile.java → src/main/java/com/nisum/java9Features/streamapi/TestDropWhile.java

-package com.nisum.java9Features.StreamApi;
+package com.nisum.java9Features.streamapi;
 
 
 import java.util.ArrayList;
@@ -11,7 +11,7 @@ public class TestDropWhile {
     private final static Logger log =Logger.getLogger(Logger.GLOBAL_LOGGER_NAME);
 
     public static void main(String[] args) {
-        ArrayList<Integer> l1 = new ArrayList<Integer>();
+        List<Integer> l1 = new ArrayList<Integer>();
         l1.add(2);
         l1.add(4);
         l1.add(1);

src/main/java/com/nisum/java9Features/StreamApi/TestIterate.java → src/main/java/com/nisum/java9Features/streamapi/TestIterate.java

-package com.nisum.java9Features.StreamApi;
+package com.nisum.java9Features.streamapi;
 
 
 import java.util.logging.Logger;
@@ -8,7 +8,6 @@ public class TestIterate {
     private final static Logger log =Logger.getLogger(Logger.GLOBAL_LOGGER_NAME);
 
     public static void main(String[] args) {
-        //Stream.iterate(1, x->x+1).forEach(System.out::println); // return infinite elements
         Stream.iterate(1, x->x+1).limit(5).forEach(System.out::println);
         log.info("3argument iterate");
         Stream.iterate(1,x->x<10, x->x+1).forEach(System.out::println);

src/main/java/com/nisum/java9Features/StreamApi/TestOfNullable.java → src/main/java/com/nisum/java9Features/streamapi/TestOfNullable.java

-package com.nisum.java9Features.StreamApi;
+package com.nisum.java9Features.streamapi;
 
 
 import java.util.ArrayList;

src/main/java/com/nisum/java9Features/StreamApi/TestTakeWhile.java → src/main/java/com/nisum/java9Features/streamapi/TestTakeWhile.java

-package com.nisum.java9Features.StreamApi;
+package com.nisum.java9Features.streamapi;
 
 import java.util.ArrayList;
 

src/main/java/com/nisum/java9Features/TryWithResourcesEnhancements/MyResources.java → src/main/java/com/nisum/java9Features/trywithresourcesenhancements/MyResources.java

-package com.nisum.java9Features.TryWithResourcesEnhancements;
+package com.nisum.java9Features.trywithresourcesenhancements;
 
 import java.util.logging.Logger;
 

src/main/java/com/nisum/java9Features/TryWithResourcesEnhancements/TryWithResourceApp.java → src/main/java/com/nisum/java9Features/trywithresourcesenhancements/TryWithResourceApp.java

-package com.nisum.java9Features.TryWithResourcesEnhancements;
-
-import lombok.extern.flogger.Flogger;
-import lombok.extern.slf4j.Slf4j;
+package com.nisum.java9Features.trywithresourcesenhancements;
 
 import java.util.logging.Logger;