package org.jboss.resteasy.test.response;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;

import jakarta.ws.rs.ext.Provider;

import org.jboss.resteasy.spi.AsyncResponseProvider;

import io.reactivex.Single;
import io.reactivex.disposables.Disposable;

@Provider
public class SingleProvider implements AsyncResponseProvider<Single<?>> {

    private static class SingleAdaptor<T> extends CompletableFuture<T> {
        private final Disposable subscription;

        SingleAdaptor(final Single<T> observable) {
            this.subscription = observable.subscribe(this::complete, this::completeExceptionally);
        }

        @Override
        public boolean cancel(boolean mayInterruptIfRunning) {
            subscription.dispose();
            return super.cancel(mayInterruptIfRunning);
        }
    }

    @Override
    public CompletionStage toCompletionStage(Single<?> asyncResponse) {
        return new SingleAdaptor<>(asyncResponse);
    }

}
