Reactive Extensions for Java [closed]

Closed. This question does not meet Stack Overflow guidelines. It is not currently accepting answers.

---

We don’t allow questions seeking recommendations for books, tools, software libraries, and more. You can edit the question开发者_运维百科 so it can be answered with facts and citations.

Closed 11 months ago.

Improve this question

Is there any equivalent of Reactive Extensions (.NET) for Java?

About Rx (Reactive Extensions)

Rx is a library for composing asynchronous and event-based programs using observable collections.

I am aware of rule engines such as Drools from JBOSS, but is there some other way that is closer to the Microsoft .NET approach?

---

According to this NetFlix blog post, they have ported Rx to Java. It doesn't seem to be available yet, but they do have a history of releasing such internally developed tools as open source, so I expect it will be released eventually.

---

I'm not aware of one - and frankly it would be hard to do in such a neat fashion. The threading side is fine; Java's perfectly capable when it comes to threading... but at the moment, the language just doesn't support the appropriate features.

One of the neat design features of LINQ is that everything is based on interfaces and then extension methods are used to add extra functionality. That allows code to read fluently - imagine if you had to write:

IObservable<int> = Observable.Select(
                       Observable.Where(
                           source, x => x.SomeCondition)
                       x => x.SomeProjection);

Ick. The extension method is much more graceful:

IObservable<int> = source.Where(x => x.SomeCondition)
                         .Select(x => x.SomeProjection);

Now a Java version could be based on an abstract base class instead, but that would lose some of the elegance.

Next come lambda expressions and method group conversions - these really are fundamental to LINQ in general; the closest equivalent in Java (anonymous inner classes) are too ugly to be used everywhere you'd use a lambda expression in C#.

Basically, a Java version of Rx would be feasible, but it wouldn't be nearly as elegant as the C# version - which is probably why it hasn't been done as far as I'm aware. There may be other "asynchrony-driven" libraries around for Java, but they're unlikely to be as comprehensive and neat as Rx.

It's possible that the changes in Java 7 will make this more feasible; as far as I know extension methods aren't being introduced, but an abstract base class version would be reasonable if the lambda syntax ends up being okay...

---

If you are still interested, I've been working on a Java library lately, which offers most of the Rx operators on reactive and interactive side as well.

It is definitely not as nice as the .NET. Java does not have extension methods or function types, therefore, everything must be called via a static method and inner classes. In .NET:

Observable.Range(0, 10).SelectMany(
    o => Observable.Range(0, o)
).Run(Console.WriteLine);

My library:

Reactive.run(
    Reactive.selectMany(
        Reactive.range(0, 10), 
        new Func1<Observable<Integer>, Integer>() {
            public Observable<Integer> invoke(Integer param1) {
               return Reactive.range(0, param1);
        }
    }       
), Reactive.println());

Interactive.run(
    Interactive.selectMany(
        Interactive.range(0, 10), 
        new Func1<Iterable<Integer>, Integer>() {
            public Iterable<Integer> invoke(Integer param1) {
               return Interactive.range(0, param1);
        }
    }       
), Interactive.println());

I use Func0, Func1, Func2 types for lambdas and Closeable type for unregistering. Once Java gets first class function citizens, these FuncX types can be simply replaced. Closeable can be used by a Java 7's try-with-resources enhancement. The third uglyness is due the weak type inference at various places.

---

RxJava (RX ported by Netflix) is available and stable here: https://github.com/ReactiveX/RxJava.

It even works with Java 8's lambda syntax pretty well! You can do stuff like that:

Observable
  .from(Arrays.asList(1, 2, 3, 4, 5))
  .filter(val -> val % 2 == 0)
  .map(val -> val * val)
  .scan(0, (prev, curr) -> prev + curr)
  .subscribe(System.out::println)

Pretty cool a?

---

You might consider porting raix (wiki), written in ActionScript 3, to Java 7 when it arrives. The project (my own) is opensource and includes black-box unit tests of the .NET version of the framework to ensure compatibility.

Like Java, AS3 supports neither extension methods nor lambda expressions so the implementation may be similar. Having said that, you would need to be careful with race conditions since the AVM doesn't support threads so there is no locking code in RxAs.

---

You can use the Project Reactor, which implements the Reactive Streams specification. It offers two Publisher interfaces: Mono and Flux.

Mono can emit 0 or 1 items, while Flux can emit 0…n items via the onNext signal.

A good tutorial on Reactive Programming in Java that covers Project Reactor is here.