RxJava

1.) Reactive extension for Java
2.) Concept is, when a change happens to something, it should be informed to who ever substribed for it.
    Eg: 
    int a = 10;
    int b = a;
    Now, when a = 20; when usually b will remain 10 t=itself.
    But in real life, it should be 20 as it's refering to a itself.
    In reactive world, we can make b point to 20.
3.) Uses Observable pattern here.
4.) 3 Os:
    Observer
    Observable
    Operator
    
5.) Observable -----push to------> Substriber/ Observer
    Observer/Substriber -----listen/ follow up with----> observable
    
    Operator is intermediator who can manipulate/ transform the data before it reaches observers.
    observable -------------<operator>---------->observer
6.) Observer can just receive changes or take action accordingly.
7.) Observables
    -Hot - live performance; When a subscriber is missing to receive, it won't republish it further.
    -Cold - Recorded type; When a subscriber miss it, can replay and watch it. Not missing that event.
    -Connectable - A cold observable converted into hot. 
                  Ie; Eg: a movie dvd is cold, but if we says, at 9 I will start it and if subsriber want to watch, can come, else miss some part.
8.)- Single Observable:
      Notifies observer only once. In success or error. But atleast once for sure.(exactly 1)
      Eg: Took a ticket for a show. Gets notification if it's started or cancelled.
9.)- May be:
      Notify on success, failure or cancel. But may or may not get informed.(min 0, max 1)
10.)-Completable:
      Notify on error or completion only.
11.) Creating Observable:
     --------------------
      1.)With Observable.just:
        Observable.just(1, 2, 3, 4, 5);//This is creation. But starts emitting only after subscription.
        just returns Observable object itself.
        So, 
        Observable.just(1, 2, 3, 4, 5).subscribe(System.out::prinltn);// action to be done with changes is specified inside subscribe method.
        just() have overloaded method with 10 arguments of same type(can be object type as well. Eg: 12, 2, 3, "Asha") at max.
        Above that if we want to use, need to use something with collection.
      2.)With fromIterable:
        Observable<Integer> observable = Observable.fromIterable(Arrays.asList(1, 2, 3, 4, 5));
        observable.subscribe(System.out::println);
      3.)With create:
        This uses emitter.
        Observable<Integer> observable = Observable.create((emitter)->{// this emitter is of thype ObservableEmitter. 
                                                                       //Emitter have methods such as onNext(), onError(), onComplete() etc.
                
            });//create expects an ObservableOnSubscribe interface. It has only one method.
                                        //subscribe(ObservableEmitter). So, use lambda to provide it.
                                        //This won't feed any data to observer. To do that,
                                        // use emitter.onNext();
                                        
                                        Ie;
                                      Observable<Integer> observable = Observable.create((emitter)->{
                                            emitter.onNext(1);
                                            emitter.onNext(2);
                                      });
                                       and then do;
        observable.subscribe(System.out::println();
        
        With emitter,
        Observable<Integer> emittedObj = Observable.create(emitter->{
            emitter.onNext(1);
            emitter.onNext(1);
            emitter.onError(new Exception("Emission Erroroorororor"));
        });

        emittedObj.subscribe(System.out::println, (err)->{
            System.out.println(err.getMessage());
        });// will print 1, 2, 3, 4, 5 and then Emission Error on error.
        
        To do something on completion;
        Observable<Integer> emittedObj = Observable.create(emitter->{
            emitter.onNext(1);
            emitter.onNext(1);
            emitter.onComplete();
        });

        emittedObj.subscribe(System.out::println, (err)->{
            System.out.println(err.getMessage());
        }, ()->{
            System.out.println("Completed");
        });
        
        
        emission of null value from observable is not allowes from rxJava2.x.
        It gives error.
        
        
        Observable<Integer> emittedObj = Observable.create(emitter->{
            emitter.onNext(1);
            emitter.onNext(1);
            emitter.onNext(null);
        });

        emittedObj.subscribe(System.out::println, (err)->{
            System.out.println(err.getMessage());
        });// prints error "onNext called with null. Null values are generally not allowed in 2.x operators and sources."
12.)Observer/ Subscriber:
        Use method subscribe(observer interface);
        Eg:
        Observable<Integer> obsToAccept = Observable.fromIterable(Arrays.asList(1, 2, 3 ,4 , 5));
        obsToAccept.subscribe(new Observer<Integer>() {
            @Override
            public void onSubscribe(Disposable d) {
                System.out.println("Subscribed");
            }

            @Override
            public void onNext(Integer integer) {
                System.out.println("On Next" + integer);
            }

            @Override
            public void onError(Throwable e) {
                System.out.println("Error");
            }

            @Override
            public void onComplete() {
                System.out.println("Complete");
            }
        });
        
13.)All observables from just(), fromIterable(), create() are cold. Any subscribers can consume any time.
    Because;
     emittedObj.subscribe(System.out::println, (err)->{
            System.out.println(err.getMessage());
        }, ()->{
            System.out.println("Completed1");
        });

        emittedObj.subscribe(System.out::print, (err)->{
            System.out.print(err.getMessage());
        }, ()->{
            System.out.println("Completed2");
        });
        will print values emitted from emittedObj.
        Then how can we make hot observables:
 14.)Create hot observable:
        Use just().publish(); or create().pusblish() or fromIterable().publish().
        This will convert observable coming from cold methods into hot observable.
        This is called ConnectableObservable.
        
        System.out.println("Connectable");
        Observable<Integer> emittedObj = Observable.create(emitter->{
            emitter.onNext(1);
            emitter.onNext(1);
        });
        ConnectableObservable<Integer> published = emittedObj.publish();

        published.subscribe(System.out::println, (err)->{// prints 1 1 when connect() is called.
            System.out.println(err.getMessage());
        }, ()->{
            System.out.println("Completed1");
        });
        published.connect();

        published.subscribe(System.out::print, (err)->{// nothing prints here since the emission finished at the time of calling published.connect();
            System.out.print(err.getMessage());
        }, ()->{
            System.out.println("Completed2");
        });
  15.) Observable Error:
        Create errors which are observable.
        Observable<Throwable> errObs = Observable.error(new Exception("Hai"));
        errObs.subscribe(()->{//if data was there}, (err)->{Syetm.out.println("Error" + err.hashCode())});
        errObs.subscribe(()->{//if data was there}, (err)->{Syetm.out.println("Error" + err.hashCode())});
        
        If we are not giving error handling lambda, while running, error gets thrown.
        
        Both will print same hashcode. Observable pushes same err object to both.
        But;
        if we want different error objects to be pushed to different customers can use observable.error(callable);
        Eg:
        // diff error obj pushed to diff subscribers
        Observable<Throwable> observableCallable = Observable.error(()->new Exception("Obs Error"));
        observableCallable.subscribe(System.out::println, (err)->{System.out.println(err.getMessage() + err.hashCode()); });
        observableCallable.subscribe(System.out::println, (err)->{System.out.println(err.getMessage() + err.hashCode());});
        prints different hashcode for error object.
        
  16.) Empty & Never Observables:
        Empty - will return empty instead of null. Ie; it just don't give anything. Just simply ends up onComplete().
        Never is observable for testing purpose. Which neither emits or not completes.
        
        Observable empty = Observable.empty();// same way, Arrays.asList(1, 2).stream().findAny();//optional - avoid null
        empty.subscribe((object)-> System.out.println("anything"), (err)->{}, ()-> System.out.println("Finished"));
        //prints Finished.

        Observable never = Observable.never();
        never.subscribe((object)-> System.out.println("anything"), (err)->{}, ()-> System.out.println("Finished"));
        //prints nothing.
  17.) Observable with Range:
        Observable<Integer> rangeForInt = Observable.range(1, 10);
        rangeForInt.subscribe(System.out::println);
        same way as;
        IntStream.range(1, 10);
  18.) Defer:
        To make an observable change according to some value. Here, when count count changes, prints 1-20 in second case. First case prints 1-10.
        static int count. = 10;
        Observable<Integer> rangeForInt = Observable.defer(()-> Observable.range(1, count));
        rangeForInt.subscribe(System.out::println);
        count = 20;
        rangeForInt.subscribe(System.out::println);
        
        One more eg: Here output comes as 120 since count got updated. Here callable is used and is invoked everytime on subscription.
        observable<Integer> ob = Observable.defer(()-> Observable.create((emitter)->{
            int c = count;
            emitter.onNext(c+10);
        }));
        count = 100;
        ob.subscribe(System.out::println);
        
        
                
      
      
      
      
      
      
      
      
      
Notes:
------
Generics cannot be applied for Constructor declaration as well as while caling sttaic methods.
Eg: Test<T>(){
    } for Test class gives error.
    Test<T extends number>(){
    } for Test class gives error.
    
    Also,
    
For async, Caallback(void return), Future, CompletableFuture for async with return value.
REST APis are sync and blocking (200 threads per 200 req by tomcat)
Dataflow is based on event/message driven.
Request comes, returns immediately. Then events are send on result fetching completed.
3 states:
onNext() - steam events
onCompletion()
onError()

Back Pressure: Giving feedback to fast systems if needed to slow down.

Reactive Stream Specs:
4 interfaces
publisher (data source) - subscribe() returns Subscription object to the subscriber.
subscriber (data accepter) - onSubscribe(), onNext(), onError(), onComplete()
subscription - request(), cancel()
processor - subscriber + publisher

Reactor - Pivotal. With springboot
Reactive Types:
Mono - 0-1 element class
Flux - 0-N elements class


Flux/Mono.just().map().subscribe(action);