reactivex/rxjavaasyncutil

rxjavaasyncutil's Issues

Deprecate Async.fromCallable in favor Observable.fromCallable

Recently Observable.fromCallable operator was added to core package in 1.0.15(ReactiveX/RxJava#3154). It serves the same function as Async.fromCallable which means fromCallable in this package can be deprecated.

cc @akarnokd @artem-zinnatullin

Enhancement Request: Add new function type that can fail

Hello,

Perhaps this has already been discussed, if so I apologize in advance for rehashing an existing discussion.

So, now that we have java 8 with lambdas, I was wondering if you would consider adding a new function type, the name could be debated, but lets just call it ThrowingFunc{0, 1, ... , N}. The interface for ThrowingFunc1 would be:

public interface ThrowingFunc1<I, R>
{
    R call( I input ) throws Exception;
}

A series of static methods could be added to the Async class, as follows:

 ....
public static <I, R> Func1<I, Observable<R>> toAsync( ThrowingFunc1<I, R> throwingFunc1 )
 ....

Using this new type and the added helper methods to Async it would be possible to do the following, for example:

public class Example
{
    public String convert( String stringToConvert ) throws ConversionException
    {
        return stringToConvert;
    }

    public static String convertStatic( String stringToConvert ) throws ConversionException
    {
        return stringToConvert;
    }

    public static main( String[] args )
    {
        Example example = new Example();

        Func1<String, Observable<String>> conversionTask1 = Async.toAsync( example::convert );

        Func1<String, Observable<String>> conversionTask2 = Async.toAsync( Example::convertStatic );
    }
}

I left out the implementation of toAsync() for a ThrowingFunc1 since it is trivial, but essentially, if the execution of the function fails with some exception, we would just call onError() with that exception.

The advantages of this interface as I see it are:
1. You could make use of method references and not be forced to wrap a throwing method call in a Func1. It is also reads really nicely since you can see directly in the call to toAsync what method is being called without the noise of a wrapping type.
2. You would not be forced to wrap the thrown exception in a RuntimeException just to tranform the method call into a Func1.

Perhaps there already exists a convenient way to do what I describe if so, I'd be interested to hear how.

Mike

Generic Exception types for ThrowableFuncX

ThrowableFunc could be made generic in the Exception it throws, this would eliminate the need for broad catch(Exception e) when you are e.g only dealing with IOExceptions (ThrowableFunc & calling method throws IOException)

public interface ThrowingFunc0<R, E extends Exception> extends Function {
        R call() throws E;
}

If you think this is a good idea, we are happy to prepare a pull request

toAsync() seems different from one of Rx.NET

As I'm trying to port RxKoans from C# To Java, I found that there are some problems that cannot be ported (for me) to RxJava due to the difference of toAsync() behaviour between C# and Java.

See "AsyncLongRuningTimeout()" function in [1]. I don't think this is portable to Java because Async.toAsync() binds given function to specific scheduler when it converts the function to an Observable. In the .NET version on the other hand, it seems like toAsync() leaves the returning observable scheduler-agnostic. If I understand the behaviour correctly, I feel .NET version is more flexible than current Async.toAsync() and I'd like to have that variant in this library.

In other words, I'd like to make following code work:

  @Test(timeout = 1000L)
  void asyncLongRunningTimeout() {
    def highFive = { Integer x ->
      Thread.sleep(200)
      return "Give me " + x
    }

    def result = null
    def incAsync = Async.toAsync(highFive as Func1<Integer, String>)
    incAsync(5).timeout(
      100, TimeUnit.MILLISECONDS,
      Observable.just("Too Slow Joe")
    ).subscribeOn(Schedulers.immediate()).subscribe({ n -> result = n })

    assert "Too Slow Joe" == result
  }

What do you think? Am I missing anything?

[1] http://rxkoans.codeplex.com/SourceControl/latest#trunk/ReactiveKoans/Koans/Lessons/Lesson7AsyncInvoke.cs

Upgrade RxJava dependency to version 2.x

Would it be possible to upgrade this library dependency on RxJava from RxJava 1.x (1.0.+) to RxJava 2.x (2.+)?