reactivex / rxjavajoins

RxJava Joins

Joins operators for RxJava.

Master Build Status

Communication

• Google Group: RxJava
• Twitter: @RxJava
• GitHub Issues

Binaries

Binaries and dependency information for Maven, Ivy, Gradle and others can be found at http://search.maven.org.

Example for Maven:

<dependency>
    <groupId>io.reactivex</groupId>
    <artifactId>rxjava-joins</artifactId>
    <version>x.y.z</version>
</dependency>

for Ivy:

<dependency org="io.reactivex" name="rxjava-joins" rev="x.y.z" />

and for Gradle:

    compile 'io.reactivex:rxjava-joins:0.22.0'

Build

To build:

$ git clone [email protected]:ReactiveX/RxJavaJoins.git
$ cd RxJavaJoins/
$ ./gradlew build

Bugs and Feedback

For bugs, questions and discussions please use the Github Issues.

LICENSE

Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

rxjavajoins's Issues

Backpressure: Join/GroupJoin

The 'join' and groupJoin operators need a backpressure story

Originally from ReactiveX/RxJava#1478

See ReactiveX/RxJava#1698 for where these operators are moved over.

Has this project been deprecated?

Has this project been deprecated?
Are there any plan to continue support RxJava 2.x?

Combine observables to create another observable

I´m using plan to do it

        Observable<Track> trackObservable = Track.get(pickLocation, dropLocation);
        Observable<UploadedFile> uploadedFileObservable = uploadImage(bitmapPath);
        Observable<OrderData> orderDataObservable = Observable.just(order);
        Observable<Address> pickObservable = Address.get(order.getPickAddress());
        Observable<Address> dropObservable = Address.get(order.getDropAddress());

Plan0<Observable<Order>> plan = JoinObservable.from(pickObservable)
                .and(dropObservable)
                .and(trackObservable)
                .and(uploadedFileObservable)
                .and(orderDataObservable)
                .then(Order::create);

        return JoinObservable.when(plan).toObservable().flatMap(orderObservable -> orderObservable);

...

private static Observable<Order> create(Address pickAddress, Address dropAddress, Track track,
                                            UploadedFile image, OrderData orderData) {
       ...

        return SERVICE.createOrder(newOrder)
                .map(order -> {
                    save(order);
                    return order;
                });
    }

I do toObservable().flatMap(orderObservable -> orderObservable); becuase createOrder generates an observable then .then(Order::create); creates an Observer<Observer<Order>>.

But don´t work.
I can´t find a way to chain multiple observers results into another observer :(

Split package

The "observables" package is contained both by rx itself and rx-joins. This causes a split package which is specially problematic when using OSGi.
The best solution to avoid this would be to rename the observables package to something else.

NoSuchMethodError: rx.functions.Actions.onErrorFrom(Lrx/Observer;)

The following code throws a:

java.lang.NoSuchMethodError: rx.functions.Actions.onErrorFrom(Lrx/Observer;)Lrx/functions/Action1;

What went wrong?

import rx.lang.scala.ImplicitFunctionConversions._
import rx.lang.scala.JavaConversions._

def sendIndexFromThread(s: Subject[Int], repeats: Int) = new Thread(new Runnable {
    def run() {
      var i = 0
      while (i < repeats) {
        s.onNext(i)
        i = i + 1
      }
      s.onCompleted()
    }
})

val internalSize = 1024

val s1 = Subject[Int]
val s2 = Subject[Int]

val o1 = s1.onBackpressureBuffer.observeOn(NewThreadScheduler())
val o2 = s2.onBackpressureBuffer.observeOn(NewThreadScheduler())

val gotAll = new CountDownLatch(internalSize * 2)
val latch = new CountDownLatch(1)
val counter = new AtomicInteger(0)

val p1 = JoinObservable.from(o1).then((x: Int) => {
  counter.incrementAndGet()
  if (counter.get == (internalSize * 2)) { latch.countDown }
})
val p2 = JoinObservable.from(o2).then((x: Int) => {
  counter.incrementAndGet()
  if (counter.get == (internalSize * 2)) { latch.countDown }
})

val obs = JoinObservable.when(p1, p2).toObservable

obs.subscribe((_: Unit) => gotAll.countDown, (_: Throwable) => (), () => ())

val thread1 = sendIndexFromThread(s1, internalSize)
val thread2 = sendIndexFromThread(s2, internalSize)

thread1.start()
thread2.start()

latch.await
gotAll.await

thread1.join()
thread2.join()