Schedulers and Publishers

For some heavy computations, we may want to run them in the background, while rendering the result in a separate thread so as not to block the UI or rendering thread. For this case, we can use the observeOn method with a different Scheduler.

public static void runComputation() throws Exception {
   StringBuffer sb = new StringBuffer();
   Flowable<String> source = Flowable.fromCallable(() -> { //1
      Thread.sleep(1000); // imitate expensive computation
      return "Done";
   });
source.doOnComplete(() -> System.out.println("Completed runComputation"));

Flowable<String> background = source.subscribeOn(Schedulers.io()); //2

Flowable<String> foreground = background.observeOn(Schedulers.single());//3

foreground.subscribe(System.out::println, Throwable::printStackTrace);//4
}

1. Create a new Flowable from a Callable, functional interface (SAM) that simply returns a value.
2. Run the Flowable using the “IO” Scheduler, which is good for blocking code like database calls or reading files.
3. Observe the results of the Flowable using a single threaded Scheduler.
4. Finally, subscribe to the resulting foreground Flowable to initiate the flow and print the results to standard out.

Publishers

For non-trivial problems, we might need to create our own Publisher.

For the following example, imagine you want to write to a file or read from a file using a custom Publisher in RxJava.

First, we write a range of numbers to a file using the following method:

public static void writeFile(File file) {
     try (PrintWriter pw = new PrintWriter(file)) {
         Flowable.range(1, 100)
           .observeOn(Schedulers.newThread())
           .blockingSubscribe(pw::println);
     } catch (FileNotFoundException e) {
          e.printStackTrace();
     }
}

Here, we use a try-with-resources block and blockingSubscribe to write the range to the file.

Second, we want to read from a file. In this example, the contents of a file are printed to standard out using the “IO” Scheduler:

public static void readFile(File file) {
     try (final BufferedReader br = new BufferedReader(new FileReader(file))) {

        Flowable<String> flow = Flowable.fromPublisher(new FilePublisher(br));

        flow.observeOn(Schedulers.io())
.blockingSubscribe(System.out::println);

     } catch (IOException e) {
         e.printStackTrace();
     }
}

A Publisher implements the subscribe method that takes a Subscriber. The Subscriber interface has several methods on it; the first of which to call is onSubscribe(Subscription). To implement backpressure in reactive streams, the Subscription interface was created, which has only two methods: request(n) for requesting the next n elements and cancel for cancelling the subscription.

static class FilePublisher implements Publisher < String > {
    BufferedReader reader;
    public FilePublisher(BufferedReader reader) {
        this.reader = reader;
    }
    @Override
    public void subscribe(Subscriber << ? super String > subscriber) {
        subscriber.onSubscribe(
            new FilePublisherSubscription(this, subscriber));
    }
    public String readLine() throws IOException {
        return reader.readLine();
    }
}

static class FilePublisherSubscription implements Subscription {
    FilePublisher publisher;
    Subscriber << ? super String > subscriber;
    public FilePublisherSubscription(FilePublisher publisher,
        Subscriber << ? super String > subscriber) {
        this.publisher = publisher;
        this.subscriber = subscriber;
    }
    @Override
    public void request(long n) {
        try {
            String line;
            for (int i = 0; i < n && publisher != null &&
                (line = publisher.readLine()) != null; i++) {
                if (subscriber != null) subscriber.onNext(line);
            }
        } catch (IOException ex) {
            subscriber.onError(ex);
        }
        subscriber.onComplete();
    }
    @Override
    public void cancel() {
        publisher = null;
    }
}

This example shows how we might implement a Publisher for reading files including backpressure support. A similar approach could be used for any Publisher or Subscription implementation.