package org.asynchttpclient.extra;

import org.asynchttpclient.AsyncHandler;
import org.asynchttpclient.HttpResponseBodyPart;
import org.asynchttpclient.HttpResponseHeaders;
import org.asynchttpclient.HttpResponseStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.Semaphore;

public class AsyncHandlerWrapper<T> implements AsyncHandler<T> {

    private final static Logger logger = LoggerFactory.getLogger(AsyncHandlerWrapper.class);
    private final AsyncHandler<T> asyncHandler;
    private final Semaphore available;

    public AsyncHandlerWrapper(AsyncHandler<T> asyncHandler, Semaphore available) {
        this.asyncHandler = asyncHandler;
        this.available = available;
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public void onThrowable(Throwable t) {
        try {
            asyncHandler.onThrowable(t);
        } finally {
            available.release();
            if (logger.isDebugEnabled()) {
                logger.debug("Current Throttling Status after onThrowable {}", available.availablePermits());
            }
        }
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public STATE onBodyPartReceived(HttpResponseBodyPart bodyPart) throws Exception {
        return asyncHandler.onBodyPartReceived(bodyPart);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public STATE onStatusReceived(HttpResponseStatus responseStatus) throws Exception {
        return asyncHandler.onStatusReceived(responseStatus);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public STATE onHeadersReceived(HttpResponseHeaders headers) throws Exception {
        return asyncHandler.onHeadersReceived(headers);
    }

    /**
     * {@inheritDoc}
     */
    @Override
    public T onCompleted() throws Exception {
        available.release();
        if (logger.isDebugEnabled()) {
            logger.debug("Current Throttling Status {}", available.availablePermits());
        }
        return asyncHandler.onCompleted();
    }
}
@@ -310,16 +310,66 @@
 maphore;
+%0Aimport java.util.concurrent.atomic.AtomicBoolean;
 %0A%0Apublic
@@ -444,28 +444,28 @@
 ate 
-final 
 static
+ final
  Logger 
 logg
@@ -460,22 +460,22 @@
  Logger 
-logger
+LOGGER
  = Logge
@@ -607,16 +607,85 @@
 ailable;
+%0A    private final AtomicBoolean complete = new AtomicBoolean(false);
 %0A%0A    pu
@@ -855,172 +855,82 @@
     
-/**%0A     * %7B@inheritDoc%7D%0A     */%0A    @Override%0A    public void onThrowable(Throwable t) %7B%0A        try %7B%0A            asyncHandler.onThrowable(t);%0A        %7D finally %7B
+private void complete() %7B%0A        if (complete.compareAndSet(false, true))
 %0A   
@@ -963,38 +963,34 @@
 ();%0A        
-    if (logger
+if (LOGGER
 .isDebugEnab
@@ -987,34 +987,32 @@
 sDebugEnabled())
- %7B
 %0A               
@@ -1004,34 +1004,30 @@
             
-    logger
+LOGGER
 .debug(%22Curr
@@ -1106,25 +1106,211 @@
 ));%0A    
+%7D%0A%0A
     
-    %7D
+/**%0A     * %7B@inheritDoc%7D%0A     */%0A    @Override%0A    public void onThrowable(Throwable t) %7B%0A        try %7B%0A            asyncHandler.onThrowable(t);%0A        %7D finally %7B%0A            complete();
 %0A       
@@ -2035,212 +2035,119 @@
     
-available.release();%0A        if (logger.isDebugEnabled()) %7B%0A            logger.debug(%22Current Throttling Status %7B%7D%22, available.availablePermits());%0A        %7D%0A        return asyncHandler.onC
+try %7B%0A            return asyncHandler.onCompleted();%0A        %7D finally %7B%0A            c
 omplete
-d
 ();%0A
+        %7D%0A
     %7D%0A%7D
+%0A

Make ThrottleRequestFilter properly release Semaphore, close #567