Using Consumer interface of Reactivex

Question

I'm new to ReactiveX. I was learning it from reading source-code. Everything was so clear but suddenly I got this word named "Consumer" which was an Interface. It was used in place of Observer.

Can someone let me know what it exactly does?

I followed several links but they just all said just one statement Consumer is a functional interface (callback) that accepts a single value.

I want to know the exact working of it.

What is it?
Why do we need it?
How do you use it?
Does it take the place of Observer? If YES, how and why?

same. i have tosubscribe like this subscribe(Consumer). dont get it? — filthy_wizard, Oct 23 '17 at 14:15

score 18 · Accepted Answer · edited Dec 17 '19 at 12:02

Consumer is a simple Java interface that accepts variable of type T. Like you said it is used for callbacks.

Example:

import io.reactivex.functions.Consumer;

Flowable.just("Hello world").subscribe(new Consumer<String>() {
      @Override public void accept(String s) {
          System.out.println(s);
      }
  });

Why does it work? How can we use a Consumer instead of an Observer?

RxJava simply creates an Observer, passes the Consumer to it an it gets called in onNext

Update

LambdaObserver is a kind of observer that is created out of four functional interfaces and uses them as callbacks. It's mostly for using java 8 lambda expressions. It looks like this:

Observable.just(new Object())
                .subscribe(
                        o -> processOnNext(o),
                        throwable -> processError(throwable),
                        () -> processCompletion(),
                        disposable -> processSubscription()
                );

How is the dispose done when using consumer interface? sample code snippet would help, thanx — abby, Jan 15 '21 at 03:40

score 7 · Answer 2 · edited Apr 11 '20 at 19:03

A Consumer consumes the values you receive when subscribing. It's like a Subscriber who passes the emitted data as callback.

The Consumer is a simple Interface which has a callback for a generic Type and is needed to receive the emitted items by the Observable.

Take care that if you only have a Consumer that you don't catch errors and you may get problems by debugging.

You can solve that by using another Consumer as second parameter which receives a Throwable.

Flowable.just("Hello world")
  .subscribe(
            emittedData -> System.out.println(emittedData), // onNext
            throwable -> throwable.printStackTrace() // onError
);

score 0 · Answer 3 · answered Dec 06 '18 at 01:41

In my humble opinion, consumer is for reverse / bi-directional streams.

For example, your data source emits a "Y" of complex time-dependent operations executed from a parameter "X" in the form of a "hot" flowable (HF).

Suppose the parameter X is emitted through a "hot" observable (HO), so, your data source can be a consumer that subscribes to the "HO" and emits the result of the complex operations through the HF.

In this case, you have the bi-directional stream and you used the consumer to push the date provided through the HO in the data source.

I am not sure if my answer is really correct... rx is a little bit complex :B

Using Consumer interface of Reactivex

3 Answers3