package org.zalando.nakadi.filters;

import com.google.common.net.HttpHeaders;
import org.json.JSONObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.web.filter.OncePerRequestFilter;
import org.zalando.nakadi.service.NakadiKpiPublisher;

import javax.servlet.FilterChain;
import javax.servlet.ServletException;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
import java.io.IOException;
import java.security.Principal;
import java.util.Optional;

@Component
public class LoggingFilter extends OncePerRequestFilter {

    // We are using empty log name, cause it is used only for access log and we do not care about class name
    private static final Logger ACCESS_LOGGER = LoggerFactory.getLogger("ACCESS_LOG");

    private final NakadiKpiPublisher nakadiKpiPublisher;
    private final String accessLogEventType;

    @Autowired
    public LoggingFilter(final NakadiKpiPublisher nakadiKpiPublisher,
                         @Value("${nakadi.kpi.event-types.nakadiAccessLog}") final String accessLogEventType) {
        this.nakadiKpiPublisher = nakadiKpiPublisher;
        this.accessLogEventType = accessLogEventType;
    }

    @Override
    protected void doFilterInternal(final HttpServletRequest request,
                                    final HttpServletResponse response, final FilterChain filterChain)
            throws ServletException, IOException {
        final long start = System.currentTimeMillis();
        try {
            //execute request
            filterChain.doFilter(request, response);
        } finally {
            final long time = System.currentTimeMillis();
            final Long timing = time - start;
            final String userAgent = Optional.ofNullable(request.getHeader("User-Agent")).orElse("-");
            final String user = Optional.ofNullable(request.getUserPrincipal()).map(Principal::getName).orElse("-");
            final String method = request.getMethod();
            final String path = request.getRequestURI();
            final String query = Optional.ofNullable(request.getQueryString()).map(q -> "?" + q).orElse("");
            final String contentEncoding = Optional.ofNullable(request.getHeader(HttpHeaders.CONTENT_ENCODING))
                    .orElse("-");
            final String acceptEncoding = Optional.ofNullable(request.getHeader(HttpHeaders.ACCEPT_ENCODING))
                    .orElse("-");
            final Long contentLength = request.getContentLengthLong() == -1 ? 0 : request.getContentLengthLong();

            ACCESS_LOGGER.info("{} \"{}{}\" \"{}\" \"{}\" {} {}ms \"{}\" \"{}\" {}B",
                    method,
                    path,
                    query,
                    userAgent,
                    user,
                    response.getStatus(),
                    timing,
                    contentEncoding,
                    acceptEncoding,
                    contentLength);
            nakadiKpiPublisher.publish(accessLogEventType, () -> new JSONObject()
                    .put("method", method)
                    .put("path", path)
                    .put("query", query)
                    .put("app", user)
                    .put("app_hashed", nakadiKpiPublisher.hash(user))
                    .put("status_code", response.getStatus())
                    .put("response_time_ms", timing));
        }
    }
}

@@ -443,16 +443,85 @@
 isher;%0A%0A
+import javax.servlet.AsyncEvent;%0Aimport javax.servlet.AsyncListener;%0A
 import j
@@ -1448,16 +1448,1070 @@
 %0A    %7D%0A%0A
+    public class AsyncRequestListener implements AsyncListener %7B%0A        private final HttpServletRequest request;%0A        private final HttpServletResponse response;%0A        private long startTime;%0A%0A        public AsyncRequestListener(final HttpServletRequest request, final HttpServletResponse response,%0A                                    final long startTime) %7B%0A            this.request = request;%0A            this.response = response;%0A            this.startTime = startTime;%0A        %7D%0A        @Override%0A        public void onComplete(final AsyncEvent event) %7B%0A            writeToAccessLogAndEventType(request, response, startTime);%0A        %7D%0A%0A        @Override%0A        public void onTimeout(final AsyncEvent event) %7B%0A            writeToAccessLogAndEventType(request, response, startTime);%0A        %7D%0A%0A        @Override%0A        public void onError(final AsyncEvent event) %7B%0A            writeToAccessLogAndEventType(request, response, startTime);%0A        %7D%0A%0A        @Override%0A        public void onStartAsync(final AsyncEvent event) %7B%0A        %7D%0A    %7D%0A%0A
     @Ove
@@ -2787,32 +2787,70 @@
 ntTimeMillis();%0A
+        boolean willBeLogged = false;%0A
         try %7B%0A  
@@ -2942,44 +2942,575 @@
     
-%7D finally %7B%0A            final long t
+    if (request.isAsyncStarted()) %7B%0A                request.getAsyncContext().addListener(new AsyncRequestListener(request, response, start));%0A                willBeLogged = true;%0A            %7D%0A        %7D finally %7B%0A            if(!willBeLogged) %7B%0A                writeToAccessLogAndEventType(request, response, start);%0A            %7D%0A        %7D%0A    %7D%0A%0A    private void writeToAccessLogAndEventType(final HttpServletRequest request, final HttpServletResponse response,%0A                                             final long requestTime) %7B%0A            final long currentT
 ime 
@@ -3575,20 +3575,33 @@
 g = 
-time - start
+currentTime - requestTime
 ;%0A  
@@ -5277,26 +5277,16 @@
 ming));%0A
-        %7D%0A
     %7D%0A%7D%0A

Add async listener in logging filter