DeferredResultSubscriber

DeferredResultSubscriber.java

import org.springframework.web.context.request.async.DeferredResult;
import rx.Observable;
import rx.Observer;
import rx.Subscription;

/**
 * A {@link DeferredResult} that subscribes to an observable
 *
 * @see DeferredResult
 */
public class DeferredResultSubscriber<T> extends DeferredResult<T> implements Observer<T>, Runnable{

    private final Subscription subscription;

    private boolean completed;

    public DeferredResultSubscriber(Observable<T> observable) {
        onTimeout(this);
        onCompletion(this);
        this.subscription = observable.subscribe(this);
    }

    @Override
    public void onNext(T value) {
        if (!completed) {
            setResult(value);
        }
    }

    @Override
    public void onError(Throwable e) {
        setErrorResult(e);
    }

    @Override
    public void onCompleted() {
        completed = true;
    }

    @Override
    public void run() {
        this.subscription.unsubscribe();
    }
}