/**
 * Copyright (C) 2015 The Gravitee team (http://gravitee.io)
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *         http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package io.gravitee.gateway.core.reactor.handler.impl;

import io.gravitee.common.http.GraviteeHttpHeader;
import io.gravitee.common.http.HttpHeaders;
import io.gravitee.common.http.HttpHeadersValues;
import io.gravitee.gateway.api.*;
import io.gravitee.gateway.api.handler.Handler;
import io.gravitee.gateway.api.http.client.HttpClient;
import io.gravitee.gateway.core.definition.Api;
import io.gravitee.gateway.core.http.StringBodyPart;
import io.gravitee.gateway.core.policy.Policy;
import io.gravitee.gateway.core.policy.StreamType;
import io.gravitee.gateway.core.policy.impl.AbstractPolicyChain;
import io.gravitee.gateway.core.policy.impl.ExecutionContextImpl;
import io.gravitee.gateway.core.reactor.handler.ContextReactorHandler;
import io.gravitee.policy.api.PolicyResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.List;

/**
 * @author David BRASSELY (brasseld at gmail.com)
 */
public class ApiReactorHandler extends ContextReactorHandler {

    private final Logger LOGGER = LoggerFactory.getLogger(ApiReactorHandler.class);

    @Autowired
    private Api api;

    @Autowired
    private HttpClient httpClient;

    @Override
    public void handle(Request serverRequest, Response serverResponse, Handler<Response> handler) {
        // Do we need API name or API id ? (this information is transferred to target API)
        serverRequest.headers().set(GraviteeHttpHeader.X_GRAVITEE_API_NAME, api.getId());

        // Set specific metrics for API handler
        serverResponse.metrics().setApi(api.getId());

        // Calculate policies
        List<Policy> policies = getPolicyResolver().resolve(serverRequest, StreamType.REQUEST);

        // Prepare execution context
        ExecutionContext executionContext = createExecutionContext();

        // Apply request policies
        AbstractPolicyChain requestPolicyChain = getRequestPolicyChainBuilder().newPolicyChain(policies, executionContext);
        requestPolicyChain.setResultHandler(requestPolicyResult -> {
            if (requestPolicyResult.isFailure()) {
                writePolicyResult(requestPolicyResult, serverResponse);

                handler.handle(serverResponse);
            } else {
                // Use the default invoker (call the remote API using HTTP client)
                Invoker invoker = httpClient;

                long serviceInvocationStart = System.currentTimeMillis();
                ClientRequest clientRequest = invoker.invoke(serverRequest, responseStream -> {

                    // Set the status
                    serverResponse.status(responseStream.status());

                    // Copy HTTP headers
                    responseStream.headers().forEach((headerName, headerValues) -> serverResponse.headers().put(headerName, headerValues));

                    responseStream.bodyHandler(bodyPart -> {
                        LOGGER.debug("{} proxying content to downstream: {} bytes", serverRequest.id(), bodyPart.length());
                        serverResponse.write(bodyPart);
                    });

                    responseStream.endHandler(result -> {
                        serverResponse.end();

                        serverResponse.metrics().setApiResponseTimeMs(System.currentTimeMillis() - serviceInvocationStart);
                        LOGGER.debug("Remote API invocation took {} ms [request={}]", serverResponse.metrics().getApiResponseTimeMs(), serverRequest.id());

                        // Transfer proxy response to the initial consumer
                        handler.handle(serverResponse);
                    });
                });

                serverRequest
                        .bodyHandler(clientRequest::write)
                        .endHandler(result -> clientRequest.end());
            }
        });

        requestPolicyChain.doNext(serverRequest, serverResponse);
    }

    private void writePolicyResult(PolicyResult policyResult, Response response) {
        response.status(policyResult.httpStatusCode());

        response.headers().set(HttpHeaders.CONNECTION, HttpHeadersValues.CONNECTION_CLOSE);

        if (policyResult.message() != null) {
            StringBodyPart responseBody = new StringBodyPart(policyResult.message());
            response.headers().set(HttpHeaders.CONTENT_LENGTH, Integer.toString(responseBody.length()));
            response.write(responseBody);
        }

        response.end();
    }

    private ExecutionContext createExecutionContext() {
        return new ExecutionContextImpl(applicationContext);
    }

    @Override
    public String getContextPath() {
        return api.getProxy().getContextPath();
    }

    /*
    @Override
    public String getVirtualHost() {
        return api.getProxy().getTarget().getAuthority();
    }
    */

    @Override
    protected void doStart() throws Exception {
        super.doStart();
        httpClient.start();
    }

    @Override
    protected void doStop() throws Exception {
        super.doStop();
        httpClient.stop();
    }

    @Override
    public String toString() {
        final StringBuilder sb = new StringBuilder("ApiReactorHandler{");
        sb.append("contextPath=").append(api.getProxy().getContextPath());
        sb.append('}');
        return sb.toString();
    }

    public Api getApi() {
        return api;
    }
}

@@ -2267,16 +2267,24 @@
 lculate 
+request 
 policies
@@ -2305,17 +2305,24 @@
 Policy%3E 
-p
+requestP
 olicies 
@@ -2625,17 +2625,24 @@
 cyChain(
-p
+requestP
 olicies,
@@ -3509,24 +3509,673 @@
 rValues));%0A%0A
+                    // Calculate response policies%0A                    List%3CPolicy%3E responsePolicies = getPolicyResolver().resolve(serverRequest, StreamType.RESPONSE);%0A                    AbstractPolicyChain responsePolicyChain = getResponsePolicyChainBuilder().newPolicyChain(responsePolicies, executionContext);%0A%0A                    responsePolicyChain.setResultHandler(responsePolicyResult -%3E %7B%0A                        if (responsePolicyResult.isFailure()) %7B%0A                            writePolicyResult(responsePolicyResult, serverResponse);%0A%0A                            handler.handle(serverResponse);%0A                        %7D else %7B%0A        
             
@@ -4219,24 +4219,32 @@
 dyPart -%3E %7B%0A
+        
             
@@ -4371,32 +4371,40 @@
                 
+        
 serverResponse.w
@@ -4431,37 +4431,53 @@
                 
+       
+ 
 %7D);%0A%0A
+        
                 
@@ -4534,32 +4534,40 @@
                 
+     
+   
 serverResponse.e
@@ -4569,24 +4569,32 @@
 nse.end();%0A%0A
+        
             
@@ -4721,32 +4721,40 @@
                 
+  
+      
 LOGGER.debug(%22Re
@@ -4894,16 +4894,24 @@
         
+        
 // Trans
@@ -4969,32 +4969,40 @@
                 
+        
 handler.handle(s
@@ -5037,17 +5037,155 @@
         
+     
-%7D
+   %7D);%0A                        %7D%0A                    %7D);%0A%0A                    responsePolicyChain.doNext(serverRequest, serverResponse
 );%0A     

Add response policy chain in the API handler