/*
    * Licensed to the Apache Software Foundation (ASF) under one
    * or more contributor license agreements.  See the NOTICE file
    * distributed with this work for additional information
    * regarding copyright ownership.  The ASF licenses this file
    * to you 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 org.apache.giraph.comm.netty;
  
  import org.apache.giraph.metrics.GiraphMetrics;
  import org.apache.giraph.metrics.ResetSuperstepMetricsObserver;
  import org.apache.giraph.metrics.SuperstepMetricsRegistry;
  import org.apache.log4j.Logger;
  
  import io.netty.buffer.ByteBuf;
  import io.netty.channel.ChannelHandlerContext;
  import io.netty.channel.ChannelInboundHandlerAdapter;
  import io.netty.channel.ChannelHandler.Sharable;
  
  
  /**
   * Keep track of the bytes received and provide some metrics when
   * desired as part of the Netty Channel stack.
   */
  @Sharable
  public class InboundByteCounter extends ChannelInboundHandlerAdapter implements
      ByteCounter, ResetSuperstepMetricsObserver {
    /** Class logger */
    private static final Logger LOG =
        Logger.getLogger(InboundByteCounter.class);
    /** ByteCounter delegate object */
    private final ByteCounterDelegate delegate = new ByteCounterDelegate(true);
  
    /** Constructor */
    public InboundByteCounter() {
      // Initialize Metrics
      GiraphMetrics.get().addSuperstepResetObserver(this);
    }
  
    public long getBytesReceived() {
      return delegate.getBytesProcessed();
    }
  
    /**
     * Returns bytes received per superstep.
     * @return Number of bytes.
     */
    public long getBytesReceivedPerSuperstep() {
      return delegate.getBytesProcessedPerSuperstep();
    }
  
    /**
     * Set bytes received per superstep to 0.
     */
    public void resetBytesReceivedPerSuperstep() {
      delegate.resetBytesProcessedPerSuperstep();
    }
  
    /**
     * @return Mbytes received / sec in the current interval
     */
    public double getMbytesPerSecReceived() {
      return delegate.getMbytesPerSecProcessed();
    }
  
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg)
      throws Exception {
      if (msg instanceof ByteBuf) {
        ByteBuf buf = (ByteBuf) msg;
        int receivedBytes = delegate.byteBookkeeper(buf);
        if (LOG.isDebugEnabled()) {
          LOG.debug("channelRead: " + ctx.channel().toString() + " buffer " +
              "size = " + receivedBytes + ", total bytes = " +
              getBytesReceived());
        }
      }
      ctx.fireChannelRead(msg);
    }
  
    @Override
    public void newSuperstep(SuperstepMetricsRegistry superstepMetrics) {
      delegate.newSuperstep(superstepMetrics);
    }
  
    @Override
    public void resetAll() {
      delegate.resetAll();
   }
 
   @Override
   public String getMetrics() {
     return delegate.getMetrics();
   }
 
   @Override
   public String getMetricsWindow(int minMsecsWindow) {
     return delegate.getMetricsWindow(minMsecsWindow);
   }
 }