rxjava2-extras

Utilities for use with RxJava 2

Features

• Strings - create/manipulate streams of String, conversions to and from
• Bytes - create/manipulate streams of byte[]
• StateMachine - a more expressive form of scan that can emit multiple events for each source event
• onBackpressureBufferToFile - high throughput with memory-mapped files
• Flowable Transformers
• Observable Transformers
• Serialized
• tests pass on Linux, Windows 10, Solaris 10
• supports Java 1.8+
• requires rxjava 2.0.7+

Status: released to Maven Central

Maven site reports are here including javadoc.

Getting started

Add this to your pom.xml:

<dependency>
  <groupId>com.github.davidmoten</groupId>
  <artifactId>rxjava2-extras</artifactId>
  <version>VERSION_HERE</version>
</dependency>

Or add this to your build.gradle:

repositories {
    mavenCentral()
}

dependencies {
    compile 'com.github.davidmoten:rxjava2-extras:VERSION_HERE'
}

Android

To use rxjava2-extras on Android you need these proguard rules.

Migration

• Primary target type is Flowable (the backpressure supporting stream)
• Operators will be implemented initially without fusion support (later)
• Where applicable Single, Maybe and Completable will be used
• To cross types (say from Flowable to Maybe) it is necessary to use to rather than compose
• Transformers (for use with compose and to) are clustered within the primary owning class rather than bunched together in the Transformers class. For example, Strings.join:

//produces a stream of "ab"
Maybe<String> o = Flowable
  .just("a","b")
  .to(Strings.join()); 

Strings

concat, join

decode

from(Reader),from(InputStream),from(File), ..

fromClasspath(String, Charset), ..

split(String), split(Pattern)

splitSimple(String)

toInputStream

trim

strings

splitLinesSkipComments

Bytes

collect

from(InputStream), from(File)

unzip(File), unzip(InputStream)

RetryWhen

Builder for .retryWhen()

IO

serverSocket(port)

Flowables

fetchPagesByRequest

match

repeat

mergeInterleaved

Transformers (Flowable)

buffer with size and timeout

collectStats

doOnEmpty

collectWhile

flatMapInterleaved

insert

mapLast

match, matchWith

maxRequest

minRequest

onBackpressureBufferToFile

rebatchRequests

reverse

stateMachine

toListWhile

windowMin

windowMax

Transformers (Observable)

onBackpressureBufferToFile

SchedulerHelper

blockUntilWorkFinished

withThreadId

withThreadIdFromCallSite

Maybes

fromNullable

Actions

doNothing setToTrue throwing

BiFunctions

constant throwing

BiPredicates

alwaysTrue alwaysFalse throwing

Callables

constant throwing

Consumers

addLongTo addTo assertBytesEquals close decrement doNothing increment printStackTrace println set setToTrue

Functions

constant identity throwing

Predicates

alwaysFalse alwaysTrue

Serialized

read

write

kryo().read

kryo().write

Documentation

buffer

To buffer on maximum size and time since last source emission:

flowable.compose(
  Transformers.buffer(maxSize, 10, TimeUnit.SECONDS));

You can also make the timeout dependent on the last emission:

flowable.compose(
  Transformers.buffer(maxSize, x -> timeout(x), TimeUnit.SECONDS));

collectStats

Accumulate statistics, emitting the accumulated results with each item.

collectWhile

Behaves as per toListWhile but allows control over the data structure used.

This operator supports request-one micro-fusion.

doOnEmpty

Performs an action only if a stream completes without emitting an item.

flowable.compose(
    Transformers.doOnEmpty(action));

fetchPagesByRequest

This is a Flowable creation method that is aimed at supporting calls to a service that provides data in pages where the page sizes are determined by requests from downstream (requests are a part of the backpressure machinery of RxJava).

Here's an example.

Suppose you have a stateless web service, say a rest service that returns JSON/XML and supplies you with

• the most popular movies of the last 24 hours sorted by descending popularity

The service supports paging in that you can pass it a start number and a page size and it will return just that slice from the list.

Now I want to give a library with a Flowable definition of this service to my colleagues that they can call in their applications whatever they may be. For example,

• Fred may just want to know the most popular movie each day,
• Greta wants to get the top 20 and then have the ability to keep scrolling down the list in her UI.

Let's see how we can efficiently support those use cases. I'm going to assume that the movie data returned by the service are mapped conveniently to objects by whatever framework I'm using (JAXB, Jersey, etc.). The fetch method looks like this:

// note that start is 0-based
List<Movie> mostPopularMovies(int start, int size);

Now I'm going to wrap this synchronous call as a Flowable to give to my colleagues:

Flowable<Movie> mostPopularMovies(int start) {
    return Flowables.fetchPagesByRequest(
          (position, n) -> Flowable.fromIterable(mostPopular(position, n)),
          start)
        // rebatch requests so that they are always between 
        // 5 and 100 except for the first request
      .compose(Transformers.rebatchRequests(5, 100, false));
}

Flowable<Movie> mostPopularMovies() {
    return mostPopularMovies(0);
}

Note particularly that the method above uses a variant of rebatchRequests to limit both minimum and maximum requests. We particularly don't want to allow a single call requesting the top 100,000 popular movies because of the memory and network pressures that arise from that call.

Righto, Fred now uses the new API like this:

Movie top = mostPopularMovies()
    .compose(Transformers.maxRequest(1))
    .first()
    .blockingFirst();

The use of maxRequest above may seem unnecessary but strangely enough the first operator requests Long.MAX_VALUE of upstream and cancels as soon as one arrives. The take, elemnentAt and firstXXX operators all have this counter-intuitive characteristic.

Greta uses the new API like this:

mostPopularMovies()
    .rebatchRequests(20)
    .doOnNext(movie -> addToUI(movie))
    .subscribe(subscriber);

A bit more detail about fetchPagesByRequest: * if the fetch function returns a Flowable that delivers fewer than the requested number of items then the overall stream completes.

insert

Inserts zero or one items into a stream if the given Maybe succeeds before the next source emission.

Example:

Flowable
  .interval(1, TimeUnit.SECONDS)
  .compose(Transformers.insert(
     Maybe.just(-1L).delay(500, TimeUnit.MILLISECONDS))) 
  .forEach(System.out::println);

produces (with 500ms intervals between every emission):

0
-1
1
-1
2
-1
3
-1
4
-1
...

mapLast

Modifies the last element of the stream via a defined Function.

Example:

Flowable
    .just(1, 2, 3)
    .compose(Transformers.mapLast(x -> x + 1))
    .forEach(System.out::println);

produces

1
2
4

match, matchWith

Finds out-of-order matches in two streams.

javadoc

You can use FlowableTranformers.matchWith or Flowables.match:

Flowable<Integer> a = Flowable.just(1, 2, 4, 3);
Flowable<Integer> b = Flowable.just(1, 2, 3, 5, 6, 4);
Flowables.match(a, b,
     x -> x, // key to match on for a
     x -> x, // key to match on for b
     (x, y) -> x // combiner
    )
   .forEach(System.out::println);

gives

1
2
3
4

Don't rely on the output order!

Under the covers elements are requested from a and b in alternating batches of 128 by default. The batch size is configurable in another overload.

maxRequest

Limits upstream requests.

• may allow requests less than the maximum
• serializes requests
• does not buffer items
• requests at start and just before emission of last item in current batch

flowable
  .compose(Transformers.maxRequest(100));

To constrain some requests then no constraint:

flowable
  .compose(Transformers.maxRequest(100, 256, 256, 256, Long.MAX_VALUE);

See also: minRequest, rebatchRequests

mergeInterleaved

When you use Flowable.merge on synchronous sources the sources are completely consumed serially. As a consequence if you want to merge two infinite sources synchronously then you only get the first source and the second source is never read.

Flowables.mergeInterleaved round-robins the requests to the window of current publishers (up to maxConcurrent) so that two or more infinite streams can be merged without resorting to apply asynchronicity via an asynchronous scheduler.

There is a sacrifice made in that the operator is less performant than merging asynchronous streams because only one stream is requested from at a time. In addition the operator does not include micro- and macro-fusion optimisations.

minRequest

Ensures requests are at least the given value(s).

• serializes requests
• may buffer items
• requests at start and just after emission of last item in current batch

flowable
  .compose(Transformers.minRequest(10));

To allow the first request through unconstrained:

flowable
  .compose(Transformers.minRequest(1, 10));

See also: maxRequest, rebatchRequests

onBackpressureBufferToFile

With this operator you can offload a stream's emissions to disk to reduce memory pressure when you have a fast producer + slow consumer (or just to minimize memory usage).

If you have used the onBackpressureBuffer operator you'll know that when a stream is p