// Path: reactive-core/src/main/java/net/oneandone/reactive/utils/Utils.java
// public class Utils {
//     
//     private Utils() { }
//     
// 
//     
//     
//     public static <T> T get(Future<T> future) {
//         try {
//             return future.get();
//         } catch (InterruptedException | ExecutionException e) {
//             throw propagate(e);
//         }
//     }
//     
//     
// 
//     
//     
//     /**
//      * unwraps an exception 
//      *  
//      * @param ex         the exception to unwrap
//      * @return the unwrapped exception
//      */
//     public static RuntimeException propagate(Throwable ex)  {
//         Throwable t = unwrap(ex);
//         return (t instanceof RuntimeException) ? (RuntimeException) t : new RuntimeException(t);
//     }
//     
//     
//     /**
//      * unwraps an exception 
//      *  
//      * @param ex         the exception to unwrap
//      * @return the unwrapped exception
//      */
//     public static Throwable unwrap(Throwable ex)  {
//         return unwrap(ex, 9);
//     }
//     
//     
//     /**
//      * unwraps an exception 
//      *  
//      * @param ex         the exception to unwrap
//      * @param maxDepth   the max depth
//      * @return the unwrapped exception
//      */
//     private static Throwable unwrap(Throwable ex, int maxDepth)  {
//         if (isCompletionException(ex) || isExecutionException(ex)) {
//             Throwable e = ex.getCause();
//             if (e != null) {
//                 if (maxDepth > 1) {
//                     return unwrap(e, maxDepth - 1);
//                 } else {
//                     return e;
//                 }
//             }
//         }
//             
//         return ex;
//     }
//     
//     
//     private static boolean isCompletionException(Throwable t) {
//         return CompletionException.class.isAssignableFrom(t.getClass());
//     }
//     
//     
//     private static boolean isExecutionException(Throwable t) {
//         return ExecutionException.class.isAssignableFrom(t.getClass());
//     }
// }


import java.util.concurrent.atomic.AtomicReference;
import javax.ws.rs.ClientErrorException;
import javax.ws.rs.NotFoundException;
import javax.ws.rs.container.AsyncResponse;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.Response.Status;
import net.oneandone.reactive.utils.Utils;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
     *  If no element is available, a 404 response is returned. If more than 1 element is available a 
     *  409 Conflict response is returned  
     *  
     * @param asyncResponse   the async response 
     * @return the subscriber 
     */
    public static final <T> Subscriber<T> toConsumeSingleSubscriber(AsyncResponse asyncResponse) {
        return new SingleSubscriber<T>(asyncResponse);
    }


    private static class SingleSubscriber<T> implements Subscriber<T> {
        
        private final AsyncResponse asyncResponse;
        private final AtomicReference<Subscription> subscriptionRef = new AtomicReference<>();
        private final AtomicReference<T> elementRef = new AtomicReference<>();
        
        
        private SingleSubscriber(AsyncResponse asyncResponse) {
            this.asyncResponse = asyncResponse;
        }
        
        @Override
        public void onSubscribe(Subscription subscription) {
            subscriptionRef.set(subscription);
            subscription.request(1);
        }
        
        @Override
        public void onError(Throwable error) {
// Path: reactive-core/src/main/java/net/oneandone/reactive/utils/Utils.java
// public class Utils {
//     
//     private Utils() { }
//     
// 
//     
//     
//     public static <T> T get(Future<T> future) {
//         try {
//             return future.get();
//         } catch (InterruptedException | ExecutionException e) {
//             throw propagate(e);
//         }
//     }
//     
//     
// 
//     
//     
//     /**
//      * unwraps an exception 
//      *  
//      * @param ex         the exception to unwrap
//      * @return the unwrapped exception
//      */
//     public static RuntimeException propagate(Throwable ex)  {
//         Throwable t = unwrap(ex);
//         return (t instanceof RuntimeException) ? (RuntimeException) t : new RuntimeException(t);
//     }
//     
//     
//     /**
//      * unwraps an exception 
//      *  
//      * @param ex         the exception to unwrap
//      * @return the unwrapped exception
//      */
//     public static Throwable unwrap(Throwable ex)  {
//         return unwrap(ex, 9);
//     }
//     
//     
//     /**
//      * unwraps an exception 
//      *  
//      * @param ex         the exception to unwrap
//      * @param maxDepth   the max depth
//      * @return the unwrapped exception
//      */
//     private static Throwable unwrap(Throwable ex, int maxDepth)  {
//         if (isCompletionException(ex) || isExecutionException(ex)) {
//             Throwable e = ex.getCause();
//             if (e != null) {
//                 if (maxDepth > 1) {
//                     return unwrap(e, maxDepth - 1);
//                 } else {
//                     return e;
//                 }
//             }
//         }
//             
//         return ex;
//     }
//     
//     
//     private static boolean isCompletionException(Throwable t) {
//         return CompletionException.class.isAssignableFrom(t.getClass());
//     }
//     
//     
//     private static boolean isExecutionException(Throwable t) {
//         return ExecutionException.class.isAssignableFrom(t.getClass());
//     }
// }


// Path: reactive-http/src/main/java/net/oneandone/reactive/rest/container/ResultSubscriber.java
import java.util.concurrent.atomic.AtomicReference;
import javax.ws.rs.ClientErrorException;
import javax.ws.rs.NotFoundException;
import javax.ws.rs.container.AsyncResponse;
import javax.ws.rs.core.Response;
import javax.ws.rs.core.Response.Status;
import net.oneandone.reactive.utils.Utils;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

     *  If no element is available, a 404 response is returned. If more than 1 element is available a 
     *  409 Conflict response is returned  
     *  
     * @param asyncResponse   the async response 
     * @return the subscriber 
     */
    public static final <T> Subscriber<T> toConsumeSingleSubscriber(AsyncResponse asyncResponse) {
        return new SingleSubscriber<T>(asyncResponse);
    }


    private static class SingleSubscriber<T> implements Subscriber<T> {
        
        private final AsyncResponse asyncResponse;
        private final AtomicReference<Subscription> subscriptionRef = new AtomicReference<>();
        private final AtomicReference<T> elementRef = new AtomicReference<>();
        
        
        private SingleSubscriber(AsyncResponse asyncResponse) {
            this.asyncResponse = asyncResponse;
        }
        
        @Override
        public void onSubscribe(Subscription subscription) {
            subscriptionRef.set(subscription);
            subscription.request(1);
        }
        
        @Override
        public void onError(Throwable error) {
            error = Utils.unwrap(error);