/*
 * Copyright 2013 Rackspace
 *
 *    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 com.rackspacecloud.blueflood.inputs.handlers;

import com.codahale.metrics.Timer;
import com.google.common.util.concurrent.ListenableFuture;
import com.rackspacecloud.blueflood.concurrent.AsyncChain;
import com.rackspacecloud.blueflood.http.HttpRequestHandler;
import com.rackspacecloud.blueflood.http.HttpResponder;
import com.rackspacecloud.blueflood.inputs.formats.JSONMetricsContainer;
import com.rackspacecloud.blueflood.io.Constants;
import com.rackspacecloud.blueflood.types.IMetric;
import com.rackspacecloud.blueflood.types.Metric;
import com.rackspacecloud.blueflood.types.MetricsCollection;
import com.rackspacecloud.blueflood.utils.Metrics;
import com.rackspacecloud.blueflood.utils.TimeValue;
import org.codehaus.jackson.JsonParseException;
import org.codehaus.jackson.map.JsonMappingException;
import org.codehaus.jackson.map.ObjectMapper;
import org.codehaus.jackson.map.type.TypeFactory;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.handler.codec.http.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeoutException;

public class HttpMetricsIngestionHandler implements HttpRequestHandler {
    private static final Logger log = LoggerFactory.getLogger(HttpMetricsIngestionHandler.class);

    protected final ObjectMapper mapper;
    protected final TypeFactory typeFactory;
    private final AsyncChain<MetricsCollection, List<Boolean>> processorChain;
    private final TimeValue timeout;

    // Metrics
    private static final Timer handlerTimer = Metrics.timer(HttpMetricsIngestionHandler.class, "HTTP metrics ingestion timer");

    public HttpMetricsIngestionHandler(AsyncChain<MetricsCollection, List<Boolean>> processorChain, TimeValue timeout) {
        this.mapper = new ObjectMapper();
        this.typeFactory = TypeFactory.defaultInstance();
        this.timeout = timeout;
        this.processorChain = processorChain;
    }

    protected JSONMetricsContainer createContainer(String body, String tenantId) throws JsonParseException, JsonMappingException, IOException {
        List<JSONMetricsContainer.JSONMetric> jsonMetrics =
                mapper.readValue(
                        body,
                        typeFactory.constructCollectionType(List.class,
                                JSONMetricsContainer.JSONMetric.class)
                );
        return new JSONMetricsContainer(tenantId, jsonMetrics);
    }

    @Override
    public void handle(ChannelHandlerContext ctx, HttpRequest request) {
        final String tenantId = request.getHeader("tenantId");
        JSONMetricsContainer jsonMetricsContainer = null;

        final Timer.Context timerContext = handlerTimer.time();
        final String body = request.getContent().toString(Constants.DEFAULT_CHARSET);
        try {
            jsonMetricsContainer = createContainer(body, tenantId);
            if (!jsonMetricsContainer.isValid()) {
                throw new IOException("Invalid JSONMetricsContainer");
            }
        } catch (JsonParseException e) {
            log.warn("Exception parsing content", e);
            sendResponse(ctx, request, "Cannot parse content", HttpResponseStatus.BAD_REQUEST);
            return;
        } catch (JsonMappingException e) {
            log.warn("Exception parsing content", e);
            sendResponse(ctx, request, "Cannot parse content", HttpResponseStatus.BAD_REQUEST);
            return;
        } catch (IOException e) {
            log.warn("IO Exception parsing content", e);
            sendResponse(ctx, request, "Cannot parse content", HttpResponseStatus.BAD_REQUEST);
            return;
        } catch (Exception e) {
            log.warn("Other exception while trying to parse content", e);
            sendResponse(ctx, request, "Failed parsing content", HttpResponseStatus.INTERNAL_SERVER_ERROR);
            return;
        }

        if (jsonMetricsContainer == null) {
            sendResponse(ctx, request, null, HttpResponseStatus.OK);
            return;
        }

        List<Metric> containerMetrics;
        try {
            containerMetrics = jsonMetricsContainer.toMetrics();
        } catch (Exception e) {
            log.warn("Exception converting JSON container to metric objects", e);
            // This could happen if clients send BigIntegers as metric values. BF doesn't handle them. So let's send a
            // BAD REQUEST message until we start handling BigIntegers.
            sendResponse(ctx, request, "Error converting JSON payload to metric objects",
                    HttpResponseStatus.BAD_REQUEST);
            return;
        }

        if (containerMetrics == null || containerMetrics.isEmpty()) {
            sendResponse(ctx, request, null, HttpResponseStatus.OK);
            return;
        }

        final MetricsCollection collection = new MetricsCollection();
        collection.add(new ArrayList<IMetric>(containerMetrics));

        try {
            ListenableFuture<List<Boolean>> futures = processorChain.apply(collection);
            List<Boolean> persisteds = futures.get(timeout.getValue(), timeout.getUnit());
            for (Boolean persisted : persisteds) {
                if (!persisted) {
                    sendResponse(ctx, request, null, HttpResponseStatus.INTERNAL_SERVER_ERROR);
                    return;
                }
            }
            sendResponse(ctx, request, null, HttpResponseStatus.OK);
        } catch (TimeoutException e) {
            sendResponse(ctx, request, "Timed out persisting metrics", HttpResponseStatus.ACCEPTED);
        } catch (Exception e) {
            log.error("Exception persisting metrics", e);
            sendResponse(ctx, request, "Error persisting metrics", HttpResponseStatus.INTERNAL_SERVER_ERROR);
        } finally {
            timerContext.stop();
        }
    }

    public static void sendResponse(ChannelHandlerContext channel, HttpRequest request, String messageBody, HttpResponseStatus status) {
        HttpResponse response = new DefaultHttpResponse(HttpVersion.HTTP_1_1, status);

        if (messageBody != null && !messageBody.isEmpty()) {
            response.setContent(ChannelBuffers.copiedBuffer(messageBody, Constants.DEFAULT_CHARSET));
        }
        HttpResponder.respond(channel, request, response);
    }
}

@@ -4674,24 +4674,105 @@
  == null) %7B%0A
+            log.warn(ctx.getChannel().getRemoteAddress() + %22 No valid metrics%22);%0A
             
@@ -4790,36 +4790,50 @@
 e(ctx, request, 
-null
+%22No valid metrics%22
 , HttpResponseSt
@@ -4829,34 +4829,43 @@
 pResponseStatus.
-OK
+BAD_REQUEST
 );%0A            r
@@ -5541,32 +5541,113 @@
 cs.isEmpty()) %7B%0A
+            log.warn(ctx.getChannel().getRemoteAddress() + %22 No valid metrics%22);%0A
             send
@@ -5661,36 +5661,50 @@
 e(ctx, request, 
-null
+%22No valid metrics%22
 , HttpResponseSt
@@ -5712,31 +5712,20 @@
 tus.
-OK);%0A            return
+BAD_REQUEST)
 ;%0A  

Return 400 when there are no valid metrics.