I&#39;ve read the Wikipedia article on [reactive programming](http://en.wikipedia.org/wiki/Reactive_programming). I&#39;ve also read the small article on [functional reactive programming](http://en.wikipedia.org/wiki/Functional_reactive_programming). The descriptions are quite abstract.

1. What does functional reactive programming (FRP) mean in practice? 
1. What does reactive programming (as opposed to non-reactive programming?) consist of? 

My background is in imperative/OO languages, so an explanation that relates to this paradigm would be appreciated.


What is (functional) reactive programming?

So now with swift, the [ReactiveCocoa](https://github.com/ReactiveCocoa/ReactiveCocoa) people have rewritten it in version 3.0 for swift

Also, there&#39;s been another project spun up called [RxSwift](https://github.com/ReactiveX/RxSwift).

I wonder if people could add information about what the differences in design/api/philosophy of the two frameworks are (please, in the spirit of SO, stick to things which are true, rather than opinions about which is &quot;best&quot;)

[Note for StackOverflow mods: This question DOES have definitive answers, the answer is the differences between the two frameworks. I think it is also highly on topic for SO]

To get started, my initial impression from reading their ReadMe&#39;s is:

- As someone who is familiar with the &quot;real&quot; C# Rx from microsoft, RxSwift looks a lot more recognisable.
- ReactiveCococa seems to have gone off into it&#39;s own space now, introducing new abstractions such as Signals vs SignalProducers and Lifting. On the one hand this seems to clarify some situations (what&#39;s a Hot vs Cold signal) but on the other hand this seems to increase the complexity of the framework a LOT

ReactiveCocoa vs RxSwift - pros and cons?

I&#39;ve been looking to understand those 3: 

* [Subject][1]
* [BehaviorSubject][2]
* [ReplaySubject][3]

I would like to use them and know when and why, what are the benefits of using them and although I&#39;ve read the documentation, watched tutorials and searched google I&#39;ve failed to make any sense of this.
 
So what are their purpose? A real-world case would be most appreciated it does not have to even code.

I would prefer a clean explanation not just &quot;a+b =&gt; c you are subscribed to ....&quot;

Thank you

[1]: https://rxjs.dev/api/index/class/Subject
[2]: https://rxjs.dev/api/index/class/BehaviorSubject
[3]: https://rxjs.dev/api/index/class/ReplaySubject

Subject vs BehaviorSubject vs ReplaySubject in Angular

The [rxjava doc][1] definition of switchmap is rather vague and it links to the [same page][2] as flatmap. What is the difference between the two operators?


  [1]: https://github.com/ReactiveX/RxJava/wiki/Transforming-Observables
  [2]: http://reactivex.io/documentation/operators/flatmap.html

What is the difference between flatmap and switchmap in RxJava?

I think I must be misunderstanding something fundamental, because in my mind this should be the most basic case for an observable, but for the life of my I can&#39;t figure out how to do it from the docs.

Basically, I want to be able to do this:

&lt;!-- language-all: lang-js --&gt;

    // create a dummy observable, which I would update manually
    var eventObservable = rx.Observable.create(function(observer){});
    var observer = eventObservable.subscribe(
       function(x){
         console.log(&#39;next: &#39; + x);
       }
    ...
    var my_function = function(){
      eventObservable.push(&#39;foo&#39;); 
      //&#39;push&#39; adds an event to the datastream, the observer gets it and prints 
      // next: foo
    }

But I have not been able to find a method like `push`. I&#39;m using this for a click handler, and I know they have `Observable.fromEvent` for that, but I&#39;m trying to use it with React and I&#39;d rather be able to simply update the datastream in a callback, instead of using a completely different event handling system. So basically I want this:

    $( &quot;#target&quot; ).click(function(e) {
      eventObservable.push(e.target.text()); 
    });

The closest I got was using `observer.onNext(&#39;foo&#39;)`, but that didn&#39;t seem to actually work and that&#39;s called on the observer, which doesn&#39;t seem right. The observer should be the thing reacting to the data stream, not changing it, right? 

Do I just not understand the observer/observable relationship? 

RxJS: How would I &quot;manually&quot; update an Observable?

I have used [Redux-Saga](https://github.com/yelouafi/redux-saga). Code written with it is easy to reason so far, except JS generator function is messing up my head from time to time. From my understanding, [Redux-Observable](https://github.com/redux-observable/redux-observable) can achieve the similar job that handles side effects but without using generator function. 

However, docs from Redux-Observable does not provide many opinions of why it is superior against Redux-Saga. I would want to know whether not using generator function is the only benefit of using Redux-Observable. And what could be the downsides, gotcha or compromises from using Redux-Observable instead of Redux-Saga? 



Why use Redux-Observable over Redux-Saga?

Can someone give a clear definition together with a simple example that explains what is a &quot;callback hell&quot; for someone who does not know JavaScript and node.js ?

When (in what kind of settings) does the &quot;callback hell problem&quot; occur? 

Why does it occur?

Is &quot;callback hell&quot; always related to asynchronous computations? 

Or can &quot;callback hell&quot; occur also in a single threaded application? 

I took the Reactive Course at Coursera and Erik Meijer said in one of his lectures that RX solves the problem of &quot;callback hell&quot;. I asked what is a &quot;callback hell&quot; on the Coursera forum but I got no clear answer.  

After explaining &quot;callback hell&quot; on a simple example, could you also show how RX solves the &quot;callback hell problem&quot; on that simple example?

What is &quot;callback hell&quot; and how and why does RX solve it?

In my app i have something like:

    this._personService.getName(id)
          .concat(this._documentService.getDocument())
          .subscribe((response) =&gt; {
                      console.log(response)
                      this.showForm()
           });
   
     //Output: 
     // [getnameResult]
     // [getDocumentResult]

     // I want:
     // [getnameResult][getDocumentResult]


Then i get two separated results, first of the `_personService` and then the `_documentService`. How can I wait for both results before call `this.showForm()` to finish an then manipulate the results of each one.


How to &#39;wait&#39; for two observables in RxJS

I was going through this [blog][1] and reading about Observables and couldn&#39;t figure out the difference between the Observable and a Subject.

  [1]: http://reactivex.io/rxjs/manual/overview.html

What is the difference between Observable and a Subject in rxjs?

I&#39;m new to project [Reactor][1] and reactive programming in general.

I&#39;m currently working on a piece of code similar to this:

	Mono.just(userId)
		.map(repo::findById)
		.map(user-&gt; {
			if(user == null){
				throw new UserNotFoundException();
			}
			return user;
		})
		// ... other mappings

This example is probably silly and there are surely better ways of implementing this case, but the point is: 

Is it wrong to use a `throw new` exception in a `map` block or should I replace this with a `return Mono.error(new UserNotFoundException())`?

Is there any actual difference in these two ways of doing?



  [1]: https://projectreactor.io/

Correct way of throwing exceptions with Reactor

**TL;DR**
How to convert `Task.whenAll(List&lt;Task&gt;)` into `RxJava`?

My existing code uses Bolts to build up a list of asynchronous tasks and waits until all of those tasks finish before performing other steps. Essentially, it builds up a `List&lt;Task&gt;` and returns a single `Task` which is marked as completed when *all* tasks in the list complete, as per the [example on the Bolts site][1].

I&#39;m looking to replace `Bolts` with `RxJava` and I&#39;m assuming this method of building up a list of async tasks (size not known in advance) and wrapping them all into a single `Observable` is possible, but I don&#39;t know how. 

I&#39;ve tried looking at `merge`, `zip`, `concat` etc... but can&#39;t get to work on the `List&lt;Observable&gt;` that I&#39;d be building up as they all seem geared to working on just two `Observables` at a time if I understand the docs correctly.

I&#39;m trying to learn `RxJava` and am still very new to it so forgive me if this is an obvious question or explained in the docs somewhere; I have tried searching. Any help would be much appreciated.


  [1]: https://github.com/BoltsFramework/Bolts-Android#tasks-in-parallel

Combine a list of Observables and wait until all completed

I&#39;m trying to visualize some simple automatic physical systems (such things as pendulum, robot arms,etc.) in Haskell.
Often those systems can be described by equations like

`df/dt = c*f(t) + u(t)`

where `u(t)` represents some kind of &#39;intelligent control&#39;. Those systems look to fit very nicely in the Functional Reactive Programming paradigm.

So I grabbed the book &quot;The Haskell School of Expression&quot; by Paul Hudak,
and found that the domain specific language &quot;FAL&quot; (for Functional Animation Language) presented there actually works quite pleasently for my simple toy systems (although some functions, notably `integrate`, seemed to be  a bit too lazy for an efficient use, but easily fixable).

My question is, what&#39;s the more mature, up-to-date, well-maintained, performance-tuned alternative for more advanced, or even practical applications today?

[This wiki page](http://www.haskell.org/haskellwiki/Functional_Reactive_Programming) lists several options for Haskell, but I&#39;m not clear about the following respects:

1. The status of &quot;reactive&quot;, the project from Conal Eliott who is (as I understand it) one of the inventers of this programming paradigm, looks a bit stale. I love his code, but maybe I should try other more up-to-date alternatives? What&#39;s the primary difference between them, in terms of syntax/performance/runtime-stability?

2. To quote from a [survey](http://www.cs.rit.edu/~eca7215/frp-independent-study/Survey.pdf) in 2011, Section 6, &quot;_... FRP implementations are still not efficient enough or predictable enough in performance to be used effectively in domains which require latency guarantees ..._&quot;. Alghough the survey suggests some interesting possible optimizations, given the fact that FRP is there for more than 15 years, I get the impression that this performance problem might be something _very_ or even inherently difficult to solve at least within a few years. Is this true?

3. The same author of the survey talks about &quot;time leaks&quot; in his [blog](http://blog.edwardamsden.com/2011/03/demonstrating-time-leak-in-arrowized.html). Is the problem unique to FRP, or something we are generally having when programming in a pure, non-strict language? Have you ever found it just too difficult to stabilize an FRP-based system over time, if not performant enough?

4. Is this still a research level project? Are the people like plant engineers, robotics engineers, financial engineers, etc. actually using them (in whaterver language that suits their needs)?

Although I personally prefer a Haskell implementation, I&#39;m open to other suggestions. For example, it would be particularly fun to have an Erlang implementation --- it would then be very easy to have an intelligent, adaptive, self-learning server process!


What&#39;s the status of current Functional Reactive Programming implementations?

I am chaining async operations using RxJava, and I&#39;d like to pass some variable downstream:

    Observable
       .from(modifications)
       .flatmap( (data1) -&gt; { return op1(data1); })
       ...
       .flatmap( (data2) -&gt; { 
           // How to access data1 here ?
           return op2(data2);
       })

It seems like a common pattern but I couldn&#39;t find information about it.


In RxJava, how to pass a variable along when chaining observables?

I have been doing some research into reactive programming recently and I&#39;m finding it hard to find a good description of the difference between Reactive and **Functional**-Reactive. 

Is it simply that the reactive programming is implemented using functional methods/paradigms as opposed to using declarative or OO paradigms? 

The difference between Reactive and Functional-Reactive programming

I am trying to print the result of `http` call in `Angular` using `rxjs`

Consider the following code

    import { Component, Injectable, OnInit } from &#39;@angular/core&#39;;
	import { Http, HTTP_PROVIDERS } from &#39;@angular/http&#39;;
	import &#39;rxjs/Rx&#39;;

	@Injectable()
	class myHTTPService {
	  constructor(private http: Http) {}
	  
	  configEndPoint: string = &#39;/my_url/get_config&#39;;

	  getConfig() {
	    
	    return this.http
	      .get(this.configEndPoint)
	      .map(res =&gt; res.json());
	  }
	}

	@Component({
	    selector: &#39;my-app&#39;,
	    templateUrl: &#39;./myTemplate&#39;,
	    providers: [HTTP_PROVIDERS, myHTTPService],


	})
	export class AppComponent implements OnInit {

	    constructor(private myService: myHTTPService) { }
	  
	    ngOnInit() {
	      console.log(this.myService.getConfig());
	    }
	}

Whenever I tried to print out the result of `getconfig` it always return 

    Observable {_isScalar: false, source: Observable, operator: MapOperator}

even though I return a json object instead.

How would I print out the result of `getConfig` ?

How to get data from observable in angular2

After learning about [Observables](https://github.com/jhusain/asyncgenerator), I find them quite similar to [Node.js streams][1]. Both have a mechanism of notifying the consumer whenever new data arrives, an error occurs or there is no more data (EOF).

I would love to learn about the conceptual/functional differences between the two. Thanks!


  [1]: https://nodejs.org/api/stream.html

Node.js Streams vs. Observables

I&#39;m learning RxJS and Angular 2. Let&#39;s say I have a promise chain with multiple async function calls which depend on the previous one&#39;s result which looks like: 

    var promiseChain = new Promise((resolve, reject) =&gt; {
      setTimeout(() =&gt; {
        resolve(1);
      }, 1000);
    }).then((result) =&gt; {
      console.log(result);

      return new Promise((resolve, reject) =&gt; {
        setTimeout(() =&gt; {
          resolve(result + 2);
        }, 1000);
      });
    }).then((result) =&gt; {
      console.log(result);

      return new Promise((resolve, reject) =&gt; {
          setTimeout(() =&gt; {
          resolve(result + 3);
            }, 1000);
      });
    });

    promiseChain.then((finalResult) =&gt; {
      console.log(finalResult);
    });

My attempts at doing the same solely using RxJS without the use of promises produced the following:

    var observableChain = Observable.create((observer) =&gt; {
      setTimeout(() =&gt; {
        observer.next(1);
        observer.complete();
      }, 1000);
    }).flatMap((result) =&gt; {
      console.log(result);

      return Observable.create((observer) =&gt; {
        setTimeout(() =&gt; {
          observer.next(result + 2);
          observer.complete()
        }, 1000);
      });
    }).flatMap((result) =&gt; {
      console.log(result);

      return Observable.create((observer) =&gt; {
        setTimeout(() =&gt; {
          observer.next(result + 3);
          observer.complete()
        }, 1000);
      });
    });

    observableChain.subscribe((finalResult) =&gt; {
      console.log(finalResult);
    });

It yields the same output as the promise chain. My questions are

1. Am I doing this right? Are there any RxJS related improvements that I can make to the above code

2. How do I get this observable chain to execute repeatedly? i.e. Adding another subscription at the end just produces an additional 6 though I expect it to print 1, 3 and 6.

    observableChain.subscribe((finalResult) =&gt; {
      console.log(finalResult);
    });
    
    observableChain.subscribe((finalResult) =&gt; {
      console.log(finalResult);
    });
    
    1
    3
    6
    6

Chaining Observables in RxJS

As per spring 5:

&gt;WebClient is an interface representing the main entry point for performing web requests.
&gt;
&gt;It has been created as a part of the Spring Web Reactive module and will be replacing the classic RestTemplate in these scenarios. The new client is a reactive, non-blocking solution that works over the HTTP/1.1 protocol

Does that mean, we need to recode for the old applications using RestTemplate if we want to upgrade to Spring 5?

Or there is some workaround to work with RestTemplate in Spring 5?


WebClient vs RestTemplate

When I create an observable from scratch, and have the observer error, then complete, the done part of the subscription never is invoked.

    var observer = Rx.Observable.create(function(observer){
    	observer.onError(new Error(&#39;no!&#39;));
        observer.onCompleted();
    })
    
    observer.subscribe(
    	function(x) { console.log(&#39;succeeded with &#39; + x ) },
    	function(x) { console.log(&#39;errored with &#39; + x ) },
    	function() { console.log(&#39;completed&#39;) }
    )

The output is:

    errored with Error: no!

I&#39;d expect it to be:

    errored with Error: no!
    completed

If I change the code to invoke onNext instead of onError, the observable properly completes:

    var observer = Rx.Observable.create(function(observer){
    	observer.onNext(&#39;Hi!&#39;);
        observer.onCompleted();
    })
    
    observer.subscribe(
    	function(x) { console.log(&#39;succeeded with &#39; + x ) },
    	function(x) { console.log(&#39;errored with &#39; + x ) },
    	function() { console.log(&#39;completed&#39;) }
    )

I get the expected output:

    succeeded with Hi! 
    completed

Why does it not complete when an error has occured?

RxJS - observable doesn&#39;t complete when an error occurs

Recently I heard the term *reactive programming* a lot, but when I searched for it, what I discovered was only some similarities with *observer pattern*. Actually, I cannot find any differences between them. What&#39;s conceptual difference between them and why the term *reactive programming* is getting buzzed?

What is difference between observer pattern and reactive programming?

I reviewed the following SO question:
https://stackoverflow.com/questions/2521277/what-are-the-hot-and-cold-observables

To summarize:

- a cold observable emits its values when it has an observer to consume them, i.e. the sequence of values received by observers is independent of time of subscription. All observers will consume the same sequence of values.
- a hot observable emits value independently of its subscriptions, i.e. the values received by observers are a function of the time of subscription.

Yet, I feel like hot vs. cold is still a source of confusion. So here are my questions:

- Are all rx observables cold by default (with the exception of subjects)? 

  I often read that events are the typical metaphor for hot observables, but I also read that `Rx.fromEvent(input, &#39;click&#39;)` is a cold observable(?).

- Are there/what are the Rx operators which turn a cold observables into a hot observable (apart from `publish`, and `share`)?

  For instance, how does it work with Rx operator `withLatestFrom`? Let `cold$` be a cold observable which has somewhere been subscribed to. Will `sth$.withLatestFrom(cold$,...)` be a hot observable? 

  Or if I do `sth1$.withLatestFrom(cold$,...), sth2$.withLatestFrom(cold$,...)` and subscribe to `sth1` and `sth2`, will I always see the same value for both `sth`?

- I thought `Rx.fromEvent` creates cold observables but that is not the case, as mentioned in one of the answers. However, I am still baffled by this behaviour: https://codepen.io/anon/pen/NqQMJR?editors=101. Different subscriptions get different values from the same observable. Wasn&#39;t the `click` event shared?

Hot and Cold observables: are there &#39;hot&#39; and &#39;cold&#39; operators?

As per the documentation:

&gt; Flux is a stream which can emit 0..N elements:

    Flux&lt;String&gt; fl = Flux.just(&quot;a&quot;, &quot;b&quot;, &quot;c&quot;);

&gt; Mono is a stream of 0..1 elements:

    Mono&lt;String&gt; mn = Mono.just(&quot;hello&quot;);

And as both are the implementations of the Publisher interface in the reactive stream.

Can&#39;t we use only Flux in most of the cases as it also can emit 0..1, thus satisfying the conditions of a Mono?

Or there are some specific conditions when only Mono needs to be used and Flux can not handle the operations?
Please suggest. 

Mono vs Flux in Reactive Stream

We wrote a small Spring Boot REST application, which performs a REST request on another REST endpoint.

    @RequestMapping(&quot;/api/v1&quot;)
    @SpringBootApplication
    @RestController
    @Slf4j
    public class Application
    {
        @Autowired
        private WebClient webClient;
    
        @RequestMapping(value = &quot;/zyx&quot;, method = POST)
        @ResponseBody
        XyzApiResponse zyx(@RequestBody XyzApiRequest request, @RequestHeader HttpHeaders headers)
        {
            webClient.post()
                .uri(&quot;/api/v1/someapi&quot;)
                .accept(MediaType.APPLICATION_JSON)
                .contentType(MediaType.APPLICATION_JSON)
                .body(BodyInserters.fromObject(request.getData()))
                .exchange()
                .subscribeOn(Schedulers.elastic())
                .flatMap(response -&gt;
                        response.bodyToMono(XyzServiceResponse.class).map(r -&gt;
                        {
                            if (r != null)
                            {
                                r.setStatus(response.statusCode().value());
                            }
    
                            if (!response.statusCode().is2xxSuccessful())
                            {
                                throw new ProcessResponseException(
                                        &quot;Bad status response code &quot; + response.statusCode() + &quot;!&quot;);
                            }
    
                            return r;
                        }))
                .subscribe(body -&gt;
                {
                    // Do various things
                }, throwable -&gt;
                {
                    // This section handles request errors
                });
    
            return XyzApiResponse.OK;
        }
    }

We are new to Spring and are having trouble writing a Unit Test for this small code snippet.

Is there an elegant (reactive) way to mock the webClient itself or to start a mock server that the webClient can use as an endpoint?

How to mock Spring WebFlux WebClient?

Given a list of cars (`List&lt;Car&gt; cars`), I can do:

    Observable.just(cars); //returns an Observable that emits one List&lt;Car&gt;
    Observable.from(cars); //returns an Observable that emits a squence of Car

Is there a way I can go from an `Observable` of a `List&lt;Car&gt;` to a sequence of `Observable&lt;Car&gt;`?

Something like a `from` without parameters

    Obserable.just(cars).from()

Convert Observable&lt;List&lt;Car&gt;&gt; to a sequence of Observable&lt;Car&gt; in RxJava

Previously in **rxjs4** there was a method in the BehaviorSubject called:
`getValue()` ([doc here][1]).

This method does not exist anymore in **rxjs5**.

So the only solution that I found to get the value of a BehaviorSubject was:

    let value;
    myBehaviorSubject.take(1).subscribe( (e) =&gt; value = e );

This code runs synchronously (I do not exactly understand why, but it does ...) and gets the value. It works, but it&#39;s not as clean as it could be if `getValue()` was present:

    let value = myBehaviorSubject.getValue();

Why `getValue()` was removed in **rxjs5** and what&#39;s the cleanest solution to this problem?

  [1]: https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/subjects/behaviorsubject.md#rxbehaviorsubjectprototypegetvalue


Simple way to get the current value of a BehaviorSubject with rxjs5

I seem to be a little bit confused when comparing Clojure&#39;s **core.async** to the so called **Reactive Extensions** (Rx) and **FRP** in general. They seem to tackle similar problem of async-hronicity, so I wonder what are the principal differences and in what cases is one preferred over the other. Can someone please explain?

EDIT: To encourage more in-depth answers I want to make the question more specific:

1. Core.async allows me to write synchronous-looking code. However as I understand it FRP only needs one level of nested callbacks (all the function that handle logic are passed as arguments to FRP API). This seems that both approaches make the callback *pyramids* unnecessary. It is true that in JS I have to write `function() {...}` many times, but the main problem, the *nested* callbacks, is gone in FRP also. Do I get it right?

2. &quot;**FRP** complects the communication of messages with flow of control&quot; Can you (someone) please give a more specific explanation?

3. Can&#39;t I pass around FRP&#39;s observable endpoints the same way as i pass channels?

In general I understand where both approaches historically come from and I have tried few tutorials in both of them. However I seem to be &quot;paralyzed&quot; by the non-obviousness of differences. Is there some **example of a code that would be hard to write in one of these and easy using the other**? And what is the architectural reason of that?

Comparing core.async and Functional Reactive Programming (+Rx)

I am learning reactive programming and functional reactive programming in JavaScript. I am very confused.

Wikipedia says that there are various ways to write reactive code such as imperative, OORP and functional. I want to know if event-driven is just an another way to write reactive code?

How is reactive programming related to Promises? I think promises is an alternative to event-driven and callback hell.

How is reactive programming different than event-driven programming?

I need some help in implementing parallel asynchronous calls in RxJava. I have picked up a simple use case wherein the FIRST call fetches (rather searches) a list of products (Tile) to be displayed. The subsequent calls go out and fetch (A) REVIEWS and (B) PRODUCT IMAGES

After several attempts I got to this place. 

     1    Observable&lt;Tile&gt; searchTile = searchServiceClient.getSearchResults(searchTerm);
     2    List&lt;Tile&gt; allTiles = new ArrayList&lt;Tile&gt;();
     3    ClientResponse response = new ClientResponse();
    
     4    searchTile.parallel(oTile -&gt; {
     5      return oTile.flatMap(t -&gt; {
     6        Observable&lt;Reviews&gt; reviews = reviewsServiceClient.getSellerReviews(t.getSellerId());
     7        Observable&lt;String&gt; imageUrl = reviewsServiceClient.getProductImage(t.getProductId());
    
     8        return Observable.zip(reviews, imageUrl, (r, u) -&gt; {
     9          t.setReviews(r);
    10          t.setImageUrl(u);
    
    11          return t;
    12        });
    
    13      });
    14    }).subscribe(e -&gt; {
    15      allTiles.add((Tile) e);
    16    });

Line  1: goes out and fetches the product (Tile) to be displayed

Line  4: We take the list of the Observable&lt;Tile&gt; and SHARD it to fetch reviews and imageUrls

Lie 6,7: Fetch the Observable&lt;Review&gt; review and Observable&lt;String&gt; url

Line  8: Finally the 2 observables are zipped up to return an updated Observable&lt;Tile&gt;

Line 15: finally line 15 collates all the individual products to be displayed in a collection which can be returned back to the calling layer

While the Observable&lt;Tile&gt; has been sharded and in our tests run over 4 different threads; fetching of reviews and images seems to be one after another. I suspect that the zip step on line 8 is basically causing the sequential invocation of the the 2 observables (reviews and url).

![enter image description here][1]

Does this group have any suggestion to parallely fetch reiews and image urls. In essence the waterfall chart attached above should look more vertically stacked. The calls to reviews and images should be in parallel

thanks
anand raman


  [1]: http://i.stack.imgur.com/YVERX.png

RxJava Fetching Observables In Parallel

Recently I&#39;ve started looking at &lt;a href=&quot;https://github.com/Reactive-Extensions/RxJS&quot;&gt;RxJS&lt;/a&gt; and RxJava(from Netflix) libraries which work on the concept of Reactive Programming. 

Node.js works on the basis of event loops, which provides you all the arsenal for asynchronous programming and the subsequent node libraries like &quot;cluster&quot; help you to get best out of your multi-core machine. And Node.js also provides you the EventEmitter functionality where you can subscribe to events and act upon it asynchronously. 

On the other hand if I understand correctly RxJS (and Reactive Programming in general) works on the principle of event streams, subscribing to event streams, transforming the event stream data asynchronously. 

So, the question is what does using Rx packages in Node.js mean. How different is the Node&#39;s event loop, event emitter &amp; subscriptions to the Rx&#39;s streams and subscriptions.

Reactive Programming - RxJS vs EventEmitter in Node.js

I&#39;m a starter in **Spring Web-Flux**. I wrote a controller as follows: 

    @RestController
    public class FirstController 
    {
        @GetMapping(&quot;/first&quot;)
        public Mono&lt;String&gt; getAllTweets() 
        {
            return Mono.just(&quot;I am First Mono&quot;)
        }
    }

I know one of the reactive benefits is **Backpressure**, and it can balance the request or the response rate. I want to realize how to have backpressure mechanism in **Spring Web-Flux**.

Backpressure mechanism in Spring Web-Flux

When to call the subscribeWith method rather than plain subscribe? And what is the use case?

    compositeDisposable.add(get()
        .observeOn(AndroidSchedulers.mainThread())
        .subscribeOn(Schedulers.io())
        .subscribe(this::handleResponse, this::handleError));

VS

       compositeDisposable.add(get()
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribeOn(Schedulers.io())
                  //  .subscribe(this::handleResponse, this::handleError);
                    .subscribeWith(new DisposableObserver&lt;News&gt;() {
                        @Override public void onNext(News value) {
                            handleResponse(value);
                        }
    
                        @Override public void onError(Throwable e) {
                            handleError(e);
                        }
    
                        @Override public void onComplete() {
                           // dispose here ? why? when the whole thing will get disposed later
                           //via  compositeDisposable.dispose();  in onDestroy();
                        }
                    }));


Thank you  


----------
**Added Later**
----------
According to the documentation, both return disposable SingleObserver instances:

    @CheckReturnValue
    @SchedulerSupport(SchedulerSupport.NONE)
    public final &lt;E extends SingleObserver&lt;? super T&gt;&gt; E subscribeWith(E observer) {
        subscribe(observer);
        return observer;
    }

    @SchedulerSupport(SchedulerSupport.NONE)
    public final Disposable subscribe(final Consumer&lt;? super T&gt; onSuccess, final Consumer&lt;? super Throwable&gt; onError) {
        ObjectHelper.requireNonNull(onSuccess, &quot;onSuccess is null&quot;);
        ObjectHelper.requireNonNull(onError, &quot;onError is null&quot;);
        ConsumerSingleObserver&lt;T&gt; s = new ConsumerSingleObserver&lt;T&gt;(onSuccess, onError);
        subscribe(s);
        return s;
    }

 Where ConsumerSingleObserver class implements SingleObserver and Disposable.



All Reactive Programming Solutions on Gang of Coders

Total of 31 Reactive Programming Solutions