blueflood-http/src/main/java/com/rackspacecloud/blueflood/inputs/handlers/HttpAggregatedMultiIngestionHandler.java
@Override
    public void handle(ChannelHandlerContext ctx, FullHttpRequest request) {

        Tracker.getInstance().track(request);

        requestCount.inc();

        final Timer.Context timerContext = handlerTimer.time();
        long ingestTime = clock.now().getMillis();

        // this is all JSON.
        String body = null;
        try {
            String submitterTenantId = request.headers().get(HttpMetricsIngestionServer.TENANT_ID_HEADER);

            body = request.content().toString(Constants.DEFAULT_CHARSET);
            List<AggregatedPayload> bundleList = createBundleList(body);

            if (bundleList.size() > 0) {
                // has aggregated metric bundle in body
                // convert and add metric bundle to MetricsCollection if valid
                MetricsCollection collection = new MetricsCollection();
                List<ErrorResponse.ErrorData> errors = new ArrayList<ErrorResponse.ErrorData>();

                // for each metric bundle
                int delayedMetricsCount = 0;
                int metricsCount = 0;
                for (AggregatedPayload bundle : bundleList) {
                    // validate, convert, and add to collection
                    List<ErrorResponse.ErrorData> bundleValidationErrors = bundle.getValidationErrors();
                    if (bundleValidationErrors.isEmpty()) {
                        // no validation error, add to collection
                        collection.add(PreaggregateConversions.buildMetricsCollection(bundle));
                    } else {
                        // failed validation, add to error
                        errors.addAll(bundleValidationErrors);
                    }

                    if (bundle.hasDelayedMetrics(ingestTime)) {
                        Tracker.getInstance().trackDelayedAggregatedMetricsTenant(bundle.getTenantId(),
                                bundle.getTimestamp(),
                                bundle.getDelayTime(ingestTime),
                                bundle.getAllMetricNames());
                        bundle.markDelayMetricsReceived(ingestTime);
                        delayedMetricsCount += bundle.getAllMetricNames().size();
                    } else {
                        metricsCount += bundle.getAllMetricNames().size();
                    }
                }

                // if has validation errors and no valid metrics
                if (!errors.isEmpty() && collection.size() == 0) {
                    // return BAD_REQUEST and error
                    DefaultHandler.sendErrorResponse(ctx, request,
                            errors,
                            HttpResponseStatus.BAD_REQUEST);
                    return;
                }

                // process valid metrics in collection
                ListenableFuture<List<Boolean>> futures = processor.apply(collection);
                List<Boolean> persisteds = futures.get(timeout.getValue(), timeout.getUnit());
                for (Boolean persisted : persisteds) {
                    if (!persisted) {
                        DefaultHandler.sendErrorResponse(ctx, request, "Internal error persisting data", HttpResponseStatus.INTERNAL_SERVER_ERROR);
                        return;
                    }
                }

                recordPerTenantMetrics(submitterTenantId, metricsCount, delayedMetricsCount);

                // return OK or MULTI_STATUS response depending if there were validation errors
                if (errors.isEmpty()) {
                    // no validation error, response OK
                    DefaultHandler.sendResponse(ctx, request, null, HttpResponseStatus.OK);
                    return;
                } else {
                    // has some validation errors, response MULTI_STATUS
                    DefaultHandler.sendErrorResponse(ctx, request, errors, HttpResponseStatus.MULTI_STATUS);
                    return;
                }

            } else {
                // no aggregated metric bundles in body, response OK
                DefaultHandler.sendResponse(ctx, request, "No valid metrics", HttpResponseStatus.BAD_REQUEST);
                return;
            }
        } catch (JsonParseException ex) {
            log.debug(String.format("BAD JSON: %s", body));
            DefaultHandler.sendErrorResponse(ctx, request, ex.getMessage(), HttpResponseStatus.BAD_REQUEST);
        } catch (InvalidDataException ex) {
            log.debug(String.format("Invalid request body: %s", body));
            DefaultHandler.sendErrorResponse(ctx, request, ex.getMessage(), HttpResponseStatus.BAD_REQUEST);
        } catch (TimeoutException ex) {
            DefaultHandler.sendErrorResponse(ctx, request, "Timed out persisting metrics", HttpResponseStatus.ACCEPTED);
        } catch (Exception ex) {
            log.debug(String.format("Exception processing: %s", body));
            log.error("Other exception while trying to parse content", ex);
            DefaultHandler.sendErrorResponse(ctx, request, "Internal error saving data", HttpResponseStatus.INTERNAL_SERVER_ERROR);
        } finally {
            timerContext.stop();
            requestCount.dec();
        }

    }
@Override
    public void handle(ChannelHandlerContext ctx, FullHttpRequest request) {

        Tracker.getInstance().track(request);

        requestCount.inc();

        final Timer.Context timerContext = handlerTimer.time();
        long ingestTime = clock.now().getMillis();

        // this is all JSON.
        String body = null;
        try {
            String submitterTenantId = request.headers().get(HttpMetricsIngestionServer.TENANT_ID_HEADER);

            body = request.content().toString(Constants.DEFAULT_CHARSET);
            List<AggregatedPayload> bundleList = createBundleList(body);

            if (bundleList.size() > 0) {
                // has aggregated metric bundle in body
                // convert and add metric bundle to MetricsCollection if valid
                MetricsCollection collection = new MetricsCollection();
                List<ErrorResponse.ErrorData> errors = new ArrayList<ErrorResponse.ErrorData>();

                // for each metric bundle
                int delayedMetricsCount = 0;
                int metricsCount = 0;
                for (AggregatedPayload bundle : bundleList) {
                    // validate, convert, and add to collection
                    List<ErrorResponse.ErrorData> bundleValidationErrors = bundle.getValidationErrors();
                    if (bundleValidationErrors.isEmpty()) {
                        // no validation error, add to collection
                        collection.add(PreaggregateConversions.buildMetricsCollection(bundle));
                    } else {
                        // failed validation, add to error
                        errors.addAll(bundleValidationErrors);
                    }

                    if (bundle.hasDelayedMetrics(ingestTime)) {
                        Tracker.getInstance().trackDelayedAggregatedMetricsTenant(bundle.getTenantId(),
                                bundle.getTimestamp(),
                                bundle.getDelayTime(ingestTime),
                                bundle.getAllMetricNames());
                        bundle.markDelayMetricsReceived(ingestTime);
                        delayedMetricsCount += bundle.getAllMetricNames().size();
                    } else {
                        metricsCount += bundle.getAllMetricNames().size();
                    }
                }

                // if has validation errors and no valid metrics
                if (!errors.isEmpty() && collection.size() == 0) {
                    // return BAD_REQUEST and error
                    DefaultHandler.sendErrorResponse(ctx, request,
                            errors,
                            HttpResponseStatus.BAD_REQUEST);
                    return;
                }

                // process valid metrics in collection
                ListenableFuture<List<Boolean>> futures = processor.apply(collection);
                List<Boolean> persisteds = futures.get(timeout.getValue(), timeout.getUnit());
                for (Boolean persisted : persisteds) {
                    if (!persisted) {
                        DefaultHandler.sendErrorResponse(ctx, request, "Internal error persisting data", HttpResponseStatus.INTERNAL_SERVER_ERROR);
                        return;
                    }
                }

                recordPerTenantMetrics(submitterTenantId, metricsCount, delayedMetricsCount);

                // return OK or MULTI_STATUS response depending if there were validation errors
                if (errors.isEmpty()) {
                    // no validation error, response OK
                    DefaultHandler.sendResponse(ctx, request, null, HttpResponseStatus.OK);
                    return;
                } else {
                    // has some validation errors, response MULTI_STATUS
                    DefaultHandler.sendErrorResponse(ctx, request, errors, HttpResponseStatus.MULTI_STATUS);
                    return;
                }

            } else {
                // no aggregated metric bundles in body, response OK
                DefaultHandler.sendResponse(ctx, request, "No valid metrics", HttpResponseStatus.BAD_REQUEST);
                return;
            }
        } catch (JsonParseException ex) {
            log.debug(String.format("BAD JSON: %s", body));
            DefaultHandler.sendErrorResponse(ctx, request, ex.getMessage(), HttpResponseStatus.BAD_REQUEST);
        } catch (InvalidDataException ex) {
            log.debug(String.format("Invalid request body: %s", body));
            DefaultHandler.sendErrorResponse(ctx, request, ex.getMessage(), HttpResponseStatus.BAD_REQUEST);
        } catch (TimeoutException ex) {
            DefaultHandler.sendErrorResponse(ctx, request, "Timed out persisting metrics", HttpResponseStatus.ACCEPTED);
        } catch (Exception ex) {
            log.debug(String.format("Exception processing: %s", body));
            log.error("Other exception while trying to parse content", ex);
            DefaultHandler.sendErrorResponse(ctx, request, "Internal error saving data", HttpResponseStatus.INTERNAL_SERVER_ERROR);
        } finally {
            timerContext.stop();
            requestCount.dec();
        }

    }
our own stuff.