As of version 0.11.0 Scala support is provided through the use of implicits. Conversations on Twitter are bringing up other possible improvements. Let's use this issue to discuss.

With the upcoming RxJava2 release one of the important changes is that null is no longer accepted as a stream element.

Honestly, I have mixed feelings about this change and part of me understands that it will enforce clean APIs, but I can see a number of use cases when this might be a problem.

For instance, in my app I have an in-memory cache:

@Nullable CacheItem findCacheItem(long id);

CacheItem might not be present in cache, so method might return null value.

The way it is used with Rx* - is as following:

Observable<CacheItem> getStream(final long id) {
    return Observable.fromCallable(new Callable<CacheItem>() {
        @Override public CacheItem call() throws Exception {
            return findCacheItem(id);
        }
    });
}

So with this approach, I might get null in my stream which is totally valid situation, so it is handled properly on receiving side - let's say UI changes its state if item is not present in cache:

Observable.just(user)
          .map(user -> user.getName())
          .map(name -> convertNameToId(name))
          .flatMap(id -> getStream(id))
          .map(cacheItem -> getUserInfoFromCacheItem(cacheItem))
          .subscribe(
              userInfo -> {
                  if(userInfo != null) showUserInfo();
                  else showPrompt();
              }
          );

With RxJava2 I am no longer allowed to post null down the stream, so I either need to wrap my CacheItem into some other class and make my stream produce that wrapper instead or make quite big architectural changes.

Wrapping every single stream element into nullable counterpart doesn't look right to me.

Am I missing something fundamental here?

It seems like the situation like mine is quite popular, so Im curious what is the recommended strategy to tackle this problem given new "no null" policy in RxJava2?

While searching for the cause of #2863 I bumped into this race condition (which doesn't fix #2863):

If a request is made between L238 and L239 then consumerCapacity may become Long.MAX_VALUE on arriving at L239 in which case we don't wish to decrement it. To fix, used compareAndSet.

What is interesting about this fix is that in the test loop of 5000 in OperatorRetryTest I see many more occurrences of the failure on average (3 -> 50) presumably because the extra time to perform the compareAndSet action has expanded the window for the race condition causing the failures.

With the following code http://pastebin.com/VUZ5aApe and using the latest version of rxjava (0.18.4),

rotations should result in the activity calling unsubscribe on my subscription.

This should release the reference to my observer, which should in turn release the reference to my activity.

However, the memory tests I've done show that as I rotate the phone, my activity count keeps going up, which means the reference to my activity is not being released.

(let me know if the pastebin expires for some reason, I'll get something back up)

I want everyone to know where we're heading and what's left before we hit 1.0.

From the beginning we have allowed ourselves to do breaking changes on each 0.x release as we were aware that the design was not finished when we started this project. We are getting close to being done. Our goal is to release 1.0 in the coming months after stabilizing the API so that it can be relied upon without breaking every couple months.

Project Structure

When we hit 1.0 we intend on splitting out the language adaptors into their own top-level projects such as RxScala, RxClojure, RxGroovy, RxKotlin, RxJRuby etc.

This will allow each project to iterate as needed at their own pace, especially since some will need to continue iterating while the RxJava core stabilizes. For example, if RxScala needs breaking changes it can bump it's major version while RxJava does not. This is particularly important to RxScala for handling changes such as Scala 2.10 -> 2.11 -> 2.12 etc.

Major contrib modules will also be moved out, such as RxAndroid which also needs its own life-cycle.

Outstanding Work

The major items of work to be finished before 1.0 are:

• ~~Backpressure: https://github.com/Netflix/RxJava/issues/1000~~ [Completed]
• ~~Serialization Behavior: https://github.com/Netflix/RxJava/issues/998~~ [Completed in new merge implementation]
• ~~Scheduler API: https://github.com/Netflix/RxJava/issues/997~~ Completed
• ~~Remove all deprecated methods/classes~~ [Completed] #1053 #1621
• ~~Finish migrating all operators to using lift and chaining the Subscription via Subscriber (current priority)~~ [Completed]

The primary goal is to nail down the public API. New functionality can come in 1.1, 1.2, etc. The secondary goal is for all operators to work as advertised (regarding unsubscribe, back pressure and non-blocking). There will always be bugs, that's why 1.x.y will still be active after release, but the desire is to not need to ship 2.x soon after 1.x as this is a low level library that once entrenched becomes hard to migrate (we create significant pain at Netflix on each 0.x release).

Going Forward

Please comment if you feel there are other critical things to achieve before 1.0. The fastest way to getting us to 1.0 is helping us achieve the work stated above.

#0.17.0 Release Notes

Version 0.17.0 contains some significant signature changes that allow us to significantly improve handling of synchronous Observables and simplify Schedulers. Many of the changes have backwards compatible deprecated methods to ease the migration while some are breaking.

The new signatures related to Observable in this release are:

// A new create method takes `OnSubscribe` instead of `OnSubscribeFunc`
public final static <T> Observable<T> create(OnSubscribe<T> f)

// The new OnSubscribe type accepts a Subscriber instead of Observer and does not return a Subscription
public static interface OnSubscribe<T> extends Action1<Subscriber<? super T>>

// Subscriber is an Observer + Subscription
public abstract class Subscriber<T> implements Observer<T>, Subscription

// The main `subscribe` behavior receives a Subscriber instead of Observer
public final Subscription subscribe(Subscriber<? super T> observer)

// Subscribing with an Observer however is still appropriate
// and the Observer is automatically converted into a Subscriber
public final Subscription subscribe(Observer<? super T> observer)

// A new 'lift' function allows composing Operator implementations together
public <R> Observable<R> lift(Func1<Subscriber<? super R>, Subscriber<? super T>> lift)

Also changed is the Scheduler interface which is much simpler:

public abstract class Scheduler {
    public Subscription schedule(Action1<Scheduler.Inner> action);
    public Subscription schedule(Action1<Scheduler.Inner> action, long delayTime, TimeUnit unit);
    public Subscription schedulePeriodically(Action1<Scheduler.Inner> action, long initialDelay, long period, TimeUnit unit);
    public long now();
    public int degreeOfParallelism();

    public static class Inner implements Subscription {
        public abstract void schedule(Action1<Scheduler.Inner> action, long delayTime, TimeUnit unit);
        public abstract void schedule(Action1<Scheduler.Inner> action);
        public long now();
    }
}

This release applies many lessons learned over the past year and seeks to streamline the API before we hit 1.0.

As shown in the code above the changes fall into 2 major sections:

1) Lift/OnSubscribe/Subscriber

Changes that allow unsubscribing from synchronous Observables without needing to add concurrency.

2) Schedulers

Simplification of the Scheduler interface and make clearer the concept of "outer" and "inner" Schedulers for recursion.

Lift/OnSubscribe/Subscriber

New types Subscriber and OnSubscribe along with the new lift operator have been added. The reasons and benefits are as follows:

1) Synchronous Unsubscribe

RxJava versions up until 0.16.x are unable to unsubscribe from a synchronous Observable such as this:

Observable<Integer> oi = Observable.create(new OnSubscribe<Integer>() {

    @Override
    public void call(Observer<? super Integer> Observer) {
        for (int i = 1; i < 1000000; i++) {
            subscriber.onNext(i);
        }
        subscriber.onCompleted();
    }
});

Subscribing to this Observable will always emit all 1,000,000 values even if unsubscribed such as via oi.take(10).

Version 0.17.0 fixes this issue by injecting the Subscription into the OnSubscribe function to allow code like this:

Observable<Integer> oi = Observable.create(new OnSubscribe<Integer>() {

    @Override
    public void call(Subscriber<? super Integer> subscriber) {
        // we now receive a Subscriber instead of Observer
        for (int i = 1; i < 1000000; i++) {
            // the OnSubscribe can now check for isUnsubscribed
            if (subscriber.isUnsubscribed()) {
                return;
            }
            subscriber.onNext(i);
        }
        subscriber.onCompleted();
    }

});

Subscribing to this will now correctly only emit 10 onNext and unsubscribe:

// subscribe with an Observer
oi.take(10).subscribe(new Observer<Integer>() {

    @Override
    public void onCompleted() {

    }

    @Override
    public void onError(Throwable e) {

    }

    @Override
    public void onNext(Integer t) {
        println("Received: " + t);
    }

})

Or the new Subscriber type can be used and the Subscriber itself can unsubscribe:

// or subscribe with a Subscriber which supports unsubscribe
oi.subscribe(new Subscriber<Integer>() {

    @Override
    public void onCompleted() {

    }

    @Override
    public void onError(Throwable e) {

    }

    @Override
    public void onNext(Integer t) {
        println("Received: " + t);
        if(t >= 10) {
            // a Subscriber can unsubscribe
            this.unsubscribe();
        }
    }

})

2) Custom Operator Chaining

Because Java doesn't support extension methods, the only approach to applying custom operators without getting them added to rx.Observable is using static methods. This has meant code like this:

MyCustomerOperators.operate(observable.map(...).filter(...).take(5)).map(...).subscribe()

In reality we want:

observable.map(...).filter(...).take(5).myCustomOperator().map(...).subscribe()

Using the newly added lift we can get quite close to this:

observable.map(...).filter(...).take(5).lift(MyCustomOperator.operate()).map(...).subscribe()

Here is how the proposed lift method looks if all operators were applied with it:

Observable<String> os = OBSERVABLE_OF_INTEGERS.lift(TAKE_5).lift(MAP_INTEGER_TO_STRING);

Along with the lift function comes a new Operator signature:

public interface Operator<R, T> extends Func1<Subscriber<? super R>, Subscriber<? super T>>

All operator implementations in the rx.operators package will over time be migrated to this new signature.

3) Simpler Operator Implementations

The lift operator injects the necessary Observer and Subscription instances (via the new Subscriber type) and eliminates (for most use cases) the need for manual subscription management. Because the Subscription is available in-scope there are no awkward coding patterns needed for creating a Subscription, closing over it and returning and taking into account synchronous vs asynchronous.

For example, the body of fromIterable is simply:

public void call(Subscriber<? super T> o) {
    for (T i : is) {
        if (o.isUnsubscribed()) {
            return;
        }
        o.onNext(i);
    }
    o.onCompleted();
}

The take operator is:

public final class OperatorTake<T> implements Operator<T, T> {

    final int limit;

    public OperatorTake(int limit) {
        this.limit = limit;
    }

    @Override
    public Subscriber<? super T> call(final Subscriber<? super T> o) {
        CompositeSubscription parent = new CompositeSubscription();
        if (limit == 0) {
            o.onCompleted();
            parent.unsubscribe();
        }
        return new Subscriber<T>(parent) {

            int count = 0;
            boolean completed = false;

            @Override
            public void onCompleted() {
                if (!completed) {
                    o.onCompleted();
                }
            }

            @Override
            public void onError(Throwable e) {
                if (!completed) {
                    o.onError(e);
                }
            }

            @Override
            public void onNext(T i) {
                if (!isUnsubscribed()) {
                    o.onNext(i);
                    if (++count >= limit) {
                        completed = true;
                        o.onCompleted();
                        unsubscribe();
                    }
                }
            }

        };
    }

}

4) Recursion/Loop Performance

The fromIterable use case is 20x faster when implemented as a loop instead of recursive scheduler (see https://github.com/Netflix/RxJava/commit/a18b8c1a572b7b9509b7a7fe1a5075ce93657771).

Several places we can remove recursive scheduling used originally for unsubscribe support and use a loop instead.

Schedulers

Schedulers were greatly simplified to a design based around Action1<Inner>.

public abstract class Scheduler {
    public Subscription schedule(Action1<Scheduler.Inner> action);
    public Subscription schedule(Action1<Scheduler.Inner> action, long delayTime, TimeUnit unit);
    public Subscription schedulePeriodically(Action1<Scheduler.Inner> action, long initialDelay, long period, TimeUnit unit);
    public long now();
    public int degreeOfParallelism();

    public static class Inner implements Subscription {
        public abstract void schedule(Action1<Scheduler.Inner> action, long delayTime, TimeUnit unit);
        public abstract void schedule(Action1<Scheduler.Inner> action);
        public long now();
    }
}

This design change originated from three findings:

1. It was very easy to cause memory leaks or inadvertent parallel execution since the distinction between outer and inner scheduling was not obvious.

To solve this the new design explicitly has the outer Scheduler and then Scheduler.Inner for recursion.

2. The passing of state is not useful since scheduling over network boundaries with this model does not work.

In this new design all state passing signatures have been removed. This was determined while implementing a RemoteScheduler that attempted to use observeOn to transition execution from one machine to another. This does not work because of the requirement for serializing/deserializing the state of the entire execution stack. Migration of work over the network has been bound to be better suited to explicit boundaries established by Subjects. Thus, the complications within the Schedulers are unnecessary.

3. The number of overloads with different ways of doing the same things were confusing.

This new design removes all but the essential and simplest methods.

4. A scheduled task could not do work in a loop and easily be unsubscribed which generally meant less efficient recursive scheduling.

This new design applies similar principles as done with lift/create/OnSubscribe/Subscriber and injects the Subscription via the Inner interface so a running task can check isUnsubscribed().

WIth this new design, the simplest execution of a single task is:

Schedulers.newThread().schedule(new Action1<Inner>() {

    @Override
    public void call(Inner inner) {
        doWork();
    }

});

Recursion is easily invoked:

Schedulers.newThread().schedule(new Action1<Inner>() {

    @Override
    public void call(Inner inner) {
        doWork();
        // recurse until unsubscribed (the schedule will do nothing if unsubscribed)
        inner.schedule(this);
    }

});

The use of Action1<Inner> on both the outer and inner levels makes it so recursion that refer to this and it works easily.

Similar to the new lift/create pattern with Subscriber the Inner is also a Subscription so it allows efficient loops with unsubscribe support:

Schedulers.newThread().schedule(new Action1<Inner>() {

    @Override
    public void call(Inner inner) {
        while(!inner.isUnsubscribed()) {
            doWork();
        }
    }

});

An action can now unsubscribe the Scheduler.Inner:

Schedulers.newThread().schedule(new Action1<Inner>() {

    @Override
    public void call(Inner inner) {
        while(!inner.isUnsubscribed()) {
            int i = doOtherWork();
            if(i > 100) {
                // an Action can cause the Scheduler to unsubscribe and stop
                inner.unsubscribe();
            }
        }
    }

});

Typically just stopping is sufficient:

Schedulers.newThread().schedule(new Action1<Inner>() {

    @Override
    public void call(Inner inner) {
        int i = doOtherWork();
        if (i < 10) {
            // recurse until done 10
            inner.schedule(this);
        }
    }

});

but if other work in other tasks is being done and you want to unsubscribe conditionally you could:

Schedulers.newThread().schedule(new Action1<Inner>() {

    @Override
    public void call(Inner inner) {
        int i = doOtherWork();
        if (i < 10) {
            // recurse until done 10
            inner.schedule(this);
        } else {
            inner.unsubscribe();
        }
    }

});

and the recursion can be delayed:

Schedulers.newThread().schedule(new Action1<Inner>() {

    @Override
    public void call(Inner inner) {
        doWork();
        // recurse until unsubscribed ... but delay the recursion
        inner.schedule(this, 500, TimeUnit.MILLISECONDS);
    }

});

The methods on the Inner never return a Subscription because they are always a single thread/event-loop/actor/etc and controlled by the Subscription returned by the initial Scheduler.schedule method. This is part of clarifying the contract.

Thus an unsubscribe controlled from the outside would be done like this:

Subscription s = Schedulers.newThread().schedule(new Action1<Inner>() {

    @Override
    public void call(Inner inner) {
        while(!inner.isUnsubscribed()) {
            doWork();
        }
    }

});

// unsubscribe from outside
s.unsubscribe();

Migration Path

1) Lift/OnSubscribe/Subscriber

The lift function will not be used by most and is additive so will not affect backwards compatibility. The Subscriber type is also additive and for most use cases does not need to be used directly, the Observer interface can continue being used.

The previous create(OnSubscribeFunc f) signature has been deprecated so code will work but now have warnings. Please begin migrating code as this will be deleted prior to the 1.0 release.

Code such as this:

Observable.create(new OnSubscribeFunc<Integer>() {

    @Override
    public Subscription onSubscribe(Observer<? super Integer> o) {
        o.onNext(1);
        o.onNext(2);
        o.onCompleted();
        return Subscriptions.empty();
    }
});

should change to this:

Observable.create(new OnSubscribe<Integer>() {

    @Override
    public void call(Subscriber<? super Integer> subscriber) {
        subscriber.onNext(1);
        subscriber.onNext(2);
        subscriber.onCompleted();
    }
});

If concurrency was being injected:

Observable.create(new OnSubscribeFunc<Integer>() {

    @Override
    public Subscription onSubscribe(final Observer<? super Integer> o) {
        final BooleanSubscription s = new BooleanSubscription();
        Thread t = new Thread(new Runnable() {

            @Override
            public void run() {
                int i = 0;
                while (s.isUnsubscribed()) {
                    o.onNext(i++);
                }
            }

        });
        t.start();
        return s;
    }
});

you may no longer need it and can implement like this instead:

Observable.create(new OnSubscribe<Integer>() {

    @Override
    public void call(Subscriber<? super Integer> subscriber) {
        int i = 0;
        while (subscriber.isUnsubscribed()) {
            subscriber.onNext(i++);
        }
    }
});

or can just simplify the Subscription management:

Observable.create(new OnSubscribe<Integer>() {

    @Override
    public void call(final Subscriber<? super Integer> subscriber) {
        Thread t = new Thread(new Runnable() {

            @Override
            public void run() {
                int i = 0;
                while (subscriber.isUnsubscribed()) {
                    subscriber.onNext(i++);
                }
            }

        });
        t.start();
    }
});

or use a Scheduler:

Observable.create(new OnSubscribe<Integer>() {

    @Override
    public void call(final Subscriber<? super Integer> subscriber) {
        Schedulers.io().schedule(new Action1<Inner>() {

            @Override
            public void call(Inner inner) {
                int i = 0;
                while (subscriber.isUnsubscribed()) {
                    subscriber.onNext(i++);
                }
            }

        });
    }
});

or use subscribeOn which now works to make synchronous Observables async while supporting unsubscribe (this didn't work before):

Observable.create(new OnSubscribe<Integer>() {

    @Override
    public void call(Subscriber<? super Integer> subscriber) {
        int i = 0;
        while (subscriber.isUnsubscribed()) {
            subscriber.onNext(i++);
        }
    }
}).subscribeOn(Schedulers.newThread());

2) Schedulers

Custom Scheduler implementations will need to be re-implemented and any direct use of the Scheduler interface will also need to be updated.

3) Subscription

If you have custom Subscription implementations you will see they now need an isUnsubscribed() method.

You can either add this method, or wrap your function using Subscriptions.create and it will handle the isUnsubscribed behavior and execute your function when unsubscribe() is called.

The Future...

This is hopefully the last of the major refactors to rxjava-core and we're approaching version 1.0. We have most if not all operators from Rx.Net that we want or intend to port. We think we have got the create/subscribe signatures as we want and the Subscription and Scheduler interfaces are now clean.

We need to improve on some of the Subject implementations still, particularly ReplaySubject. We are beginning to focus after this release on cleaning up all of the operator implementations, stabilizing, fixing bugs and performance tuning.

We appreciate your usage, feedback and contributions and hope the library is creating value for you!

Reviewing the Scheduler interface changes of 0.17 with @headinthebox revealed that we're not 100% happy with the outcome, particularly after learning that Java 8 does not allow referencing this from within a lambda.

The Scheduler interface as of 0.17 is:

class Scheduler {
    public abstract Subscription schedule(Action1<Scheduler.Inner> action);
    public abstract Subscription schedule(Action1<Scheduler.Inner> action, final long delayTime, final TimeUnit unit);
    public Subscription scheduleRecursive(Action1<Recurse> action);
    public Subscription schedulePeriodically(Action1<Scheduler.Inner> action, long initialDelay, long period, TimeUnit unit);
    public int degreeOfParallelism();
    public long now();

    public static final class Recurse {
        public void schedule() {
        public void schedule(long delay, TimeUnit unit) {
    }

    public abstract static class Inner implements Subscription {
        public abstract void schedule(Action1<Scheduler.Inner> action, long delayTime, TimeUnit unit);
        public abstract void schedule(Action1<Scheduler.Inner> action);
        public long now();
    }
}

We have determined two problems with this:

1. Inner/Outer Dance

In practice we have found that usage is always one of two things, either you just interact with the outer and don't care about the Inner, or you immediately need the Inner and have to do an awkward first scheduling just to get access to the Inner. (See here and weep.)

2. Recursion

The Action1<Scheduler.Inner> signature was chosen and put on both outer and inner so that an inner class could refer to itself using this to simply reschedule itself from the outer onto the inner.

It was assumed this would work in Java 8 lambdas but unfortunately we did not prove it.

This works with anonymous classes:

Schedulers.newThread().schedule(new Action1<Inner>() {

    @Override
    public void call(Inner inner) {
        System.out.println("do stuff");
        // recurse
        inner.schedule(this);
    }

});

but this does not with lambdas:

Schedulers.newThread().schedule((inner) -> {
    System.out.println("do stuff");
    inner.schedule(this); // doesn't compile
});

So we end up with this:

Schedulers.newThread().scheduleRecursive((recurse) -> {
    System.out.println("do stuff");
    recurse.schedule();
});

At that point it's clear that Inner is not working well and we have Recurse to fix the problem.

Thus, the proposed changes (breaking again) are:

class Scheduler {
    public final Subscription schedule(Action1<Recurse> action);
    public final Subscription schedule(Action1<Recurse> action, final long delayTime, final TimeUnit unit);
    public final Subscription schedulePeriodically(Action1<Recurse> action, long initialDelay, long period, TimeUnit unit);
    public abstract Inner createInner(); // for advanced use cases like `observeOn`
    public int degreeOfParallelism();
    public long now();

    // now the primary interface
    public static final class Recurse {
        public final void schedule();
        public final void schedule(long delay, TimeUnit unit);
        public final void schedule(Action1<Recurse> action);
        public final void schedule(Action1<Recurse> action, final long delayTime, final TimeUnit unit);
    }

    // now mostly an implementation detail except for advanced use cases
    public abstract static class Inner implements Subscription {
        public abstract void schedule(Action1<Recurse> action, long delayTime, TimeUnit unit);
        public abstract void schedule(Action1<Recurse> action);
        public long now();
    }
}

The name of Recurse is up for debate. It may be possible to merge Recurse and Inner but I haven't figured it out yet. The reason is that Inner is a single instance representing a thread or event-loop whereas Recurse represents an Action or work. Thus a given Inner could have multiple Recurse actions scheduled on to it. It is being an Action that allows it to recurse by invoking schedule() that just reschedules itself.

This would make it better support Java 8 lambdas and simply recursion, while also better supporting (via the createInner() method) the more complicated use cases like observeOn where current code is very awkward.

This needs to be the last refactor of this so we nail it down and stop breaking things and can get to 1.0.

Let the discussion begin ...

We need to spend time profiling memory and object allocation and finding places where we can improve.

I would really appreciate help diving into this and finding problem areas. Even if you don't fix them, but just identity use cases, operators, etc that would be very valuable.

This is partly a result of the fact that in Netflix production we have seen an increase in YoungGen GCs since 0.17.x.

The areas to start should probably be:

• Observable.create
• Observable.lift
• Subscriber
• CompositeSubscription
• map
• flatMap

If you can or want to get involved in this please comment here so we all can collaborate together.

Adds rx.Task as a "scalar Observable" for representing work with a single return value.

See https://github.com/ReactiveX/RxJava/issues/1594 rx.Future/Task

This provides a type similar to Future in that it represents a scalar unit of work, but it is lazy like an Observable and many Tasks can be combined into an Observable stream. Note how Task.zip returns Task<R> whereas Task.merge returns Observable<R>.

NOTE: This is for experimentation and feedback at this time.

Items requiring review and work that I'm particularly aware of:

• naming of OnExecute
• naming of TaskObserver (this one in particular I don't like)
• design and implementation of Task.Promise
• should the public lift use the Observable.Operator or should that only be for internal reuse?
• should we have a public lift that uses a Task.Operator?
• the Task.toObservable implementation right now is efficient but will likely break something so it likely needs to change to use subscribe
• implementation of this merge variant: Task<T> merge(final Task<? extends Task<? extends T>> source)
• several operators currently just wrap as Observable to reuse existing operators ... is that okay performance wise?
• Javadocs

Examples of using this class:

import rx.Observable;
import rx.Task;
import rx.Task.Promise;

public class TaskExamples {

    public static void main(String... args) {
        // scalar synchronous value
        Task<String> t1 = Task.create(t -> {
            t.onSuccess("Hello World!");
        });

        // scalar synchronous value using helper method
        Task<Integer> t2 = Task.just(1);

        // synchronous error
        Task<String> error = Task.create(t -> {
            t.onError(new RuntimeException("failed!"));
        });

        // executing
        t1.subscribe(System.out::println);
        t2.subscribe(System.out::println);
        error.subscribe(System.out::println, e -> System.out.println(e.getMessage()));

        // scalar Tasks for request/response like a Future
        getData(1).subscribe(System.out::println);
        getDataUsingPromise(2).subscribe(System.out::println);

        // combining Tasks into another Task
        Task<String> zipped = Task.zip(t1, t2, (a, b) -> a + " -- " + b);

        // combining Tasks into an Observable stream
        Observable<String> merged = Task.merge(t1, t2.map(String::valueOf), getData(3));
        Observable<String> mergeWith = t1.mergeWith(t2.map(String::valueOf));

        zipped.subscribe(v -> System.out.println("zipped => " + v));
        merged.subscribe(v -> System.out.println("merged => " + v));
        mergeWith.subscribe(v -> System.out.println("mergeWith => " + v));
    }

    /**
     * Example of an async scalar execution using Task.create
     * <p>
     * This shows the lazy, idiomatic approach for Rx exactly like an Observable except scalar.
     *
     * @param arg
     * @return
     */
    public static Task<String> getData(int arg) {
        return Task.create(s -> {
            new Thread(() -> {
                try {
                    Thread.sleep(500);
                } catch (Exception e) {
                    e.printStackTrace();
                }
                // deliver value
                    s.onSuccess("Data=" + arg);
                }).start();
        });
    }

    /**
     * Example of an async scalar execution using a Task.Promise
     * <p>
     * This shows how an eager (hot) process would work like using a Future.
     *
     * @param arg
     * @return
     */
    public static Task<String> getDataUsingPromise(int arg) {
        Task.Promise<String> p = Promise.create();

        new Thread(() -> {
            try {
                Thread.sleep(500);
            } catch (Exception e) {
                e.printStackTrace();
            }
            // deliver value
                p.onSuccess("Data=" + arg);
            }).start();

        return p.getTask();
    }
}

RxJava in not nice when it comes to stacktraces.

The worst thing is observeOn that schedules execution on different threads. Every time we use observeOn we get our stacktrace from the process root, all the call history gets erased.

The idea is to make schedulers save the caller's stack, wrap action calls and attach the stacktrace to each exception that has been thrown by scheduled actions.

Pros: traceable exceptions Cons: Performance cost Leaks because of recursion

I think that we can have Schedulers.io(boolean trace) alternative that will save the stacktrace, so we could call it Schedules.io(DEBUG) to turn stacktracing off on production for performance critical parts.

How do you guys find this idea?

Ok, so this pull request changes a lot of lines. It's mostly generalizing all the FuncXs to be used like FuncX[-T1, -T2, ..., -TX, +R] (contravariant parameters, covariant return type) and all the Observers to be used "in a contravariant way". A few of the Observable uses are covariant, now, too (mostly zip).

This is the pull request for #326.

This doesn't look very good in the code (thanks Java). Also, it doesn't seem to make Scala interop easier at all (at least not yet).

Please take a look. I'm not exactly happy with the result. - Maybe I'm doing something wrong here? - I've still got hope that there's an easier way...

The pull request compiles and tests ok for me (except for the Clojure module, but that's another story and not due to my changes).

Bumps com.vanniktech.maven.publish from 0.19.0 to 0.23.1.

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting @dependabot rebase.

Bumps testng from 7.5 to 7.7.1.

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting @dependabot rebase.

Bumps actions/cache from 3.0.11 to 3.2.2.

Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting @dependabot rebase.

src/main/java/io/reactivex/rxjava3/exceptions/Exceptions.java:70: warning: [removal] ThreadDeath in java.lang has been deprecated and marked for removal
        } else if (t instanceof ThreadDeath) {
                                ^
src/main/java/io/reactivex/rxjava3/exceptions/Exceptions.java:71: warning: [removal] ThreadDeath in java.lang has been deprecated and marked for removal
            throw (ThreadDeath) t;
                   ^
src/test/java/io/reactivex/rxjava3/exceptions/ExceptionsTest.java:143: warning: [removal] ThreadDeath in java.lang has been deprecated and marked for removal
    @Test(expected = ThreadDeath.class)
                     ^
src/test/java/io/reactivex/rxjava3/exceptions/ExceptionsTest.java:164: warning: [removal] ThreadDeath in java.lang has been deprecated and marked for removal
                throw new ThreadDeath();
                          ^
src/test/java/io/reactivex/rxjava3/exceptions/ExceptionsTest.java:179: warning: [removal] ThreadDeath in java.lang has been deprecated and marked for removal
            Exceptions.throwIfFatal(new ThreadDeath());
                                        ^
src/test/java/io/reactivex/rxjava3/exceptions/ExceptionsTest.java:181: warning: [removal] ThreadDeath in java.lang has been deprecated and marked for removal
        } catch (ThreadDeath ex) {
                 ^

Add onDropped callback overloads to most operators that drop items that can't be recovered by other means.

• [x] throttleLatest PR #7457
• [x] throttleFirst PR #7482
• [x] throttleLast / sample PR #7488
• [ ] throttleWithTimeout / debounce (PR TBD)
• [ ] onBackpressureLatest (PR TBD)
• [ ] onBackpressureBuffer (PR TBD)
  • 📓 The current overloads with Action are somewhat unhelpful, however, we'll have to add overloads in a way that avoids lambda ambiguity.