/**
 * Copyright 2016 Smoke Turner, LLC.
 *
 * 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.smoketurner.uploader.application.handler;

import java.util.concurrent.atomic.AtomicReference;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.codahale.metrics.Meter;
import com.codahale.metrics.MetricRegistry;
import com.codahale.metrics.SharedMetricRegistries;
import com.smoketurner.uploader.application.core.Batch;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;

public class BatchHandler extends SimpleChannelInboundHandler<byte[]> {

    private static final Logger LOGGER = LoggerFactory
            .getLogger(BatchHandler.class);
    private final long maxUploadBytes;
    private final Meter eventMeter;
    private final AtomicReference<Batch> curBatch = new AtomicReference<>();

    /**
     * Constructor
     *
     * @param maxUploadBytes
     *            Maximum size of AWS S3 file to upload
     */
    public BatchHandler(final long maxUploadBytes) {
        this.maxUploadBytes = maxUploadBytes;

        final MetricRegistry registry = SharedMetricRegistries
                .getOrCreate("default");
        this.eventMeter = registry
                .meter(MetricRegistry.name(BatchHandler.class, "event-rate"));
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        LOGGER.debug("channelActive: creating new batch");
        curBatch.set(new Batch());
    }

    @Override
    public void channelRead0(ChannelHandlerContext ctx, byte[] msg)
            throws Exception {

        LOGGER.trace("channelRead0: received message");

        eventMeter.mark();

        final Batch batch = curBatch.get();
        batch.add(msg);

        if (batch.size() > maxUploadBytes) {
            LOGGER.debug(
                    "Batch size {} bytes exceeds max upload size of {} bytes",
                    batch.size(), maxUploadBytes);

            batch.finish();
            ctx.fireChannelRead(batch);
            curBatch.set(new Batch());
        }
    }

    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) {
        ctx.flush();
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        final Batch batch = curBatch.get();
        if (batch != null && !batch.isEmpty()) {
            LOGGER.debug(
                    "Channel inactive, sending remaining batch of {} events",
                    batch.getCount());
            batch.finish();
            ctx.fireChannelRead(batch);
            curBatch.set(null);
        }
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt)
            throws Exception {
        if (evt instanceof IdleStateEvent) {
            final IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state() == IdleState.READER_IDLE) {
                LOGGER.debug("No data received on channel, closing");
                ctx.close();
            }
        }
    }
}

@@ -1311,82 +1311,8 @@
 s);%0A
-    private final long maxUploadBytes;%0A    private final Meter eventMeter;
 %0A   
@@ -1384,16 +1384,107 @@
 nce%3C%3E();
+%0A    private final long maxUploadBytes;%0A%0A    // metrics%0A    private final Meter eventMeter;
 %0A%0A    /*

Move some code around