Instantly share code, notes, and snippets.

Embed
What would you like to do?
import com.twitter.finagle.Service;
import com.twitter.util.Future;
import com.twitter.util.Promise;
import rx.Observable;
import rx.Observer;
import rx.Subscription;
public class Convert {
static public <Res> Future<Res> fromObservable(final Observable<Res> observable) {
final Observable<Res> single = observable.single();
final Promise<Res> result = new Promise<Res>() {
Promise<Res> self = this;
Res singletonValue;
final Subscription subscription = single.subscribe(new Observer<Res>() {
@Override
public void onCompleted() {
assert singletonValue != null : "At least one on next should have occured; the singletonValue should be set";
self.setValue(singletonValue);
}
@Override
public void onError(Throwable e) {
self.setException(e);
}
@Override
public void onNext(Res res) {
singletonValue = res;
}
});
@Override
public void cancel() {
subscription.unsubscribe();
}
};
return result;
}
}
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment