/*

  * 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.handler;

 import org.apache.giraph.comm.netty.NettyClient;

 import org.apache.hadoop.conf.Configuration;

 import org.apache.log4j.Logger;

 import io.netty.buffer.ByteBuf;

 import io.netty.channel.ChannelHandlerContext;

 import io.netty.channel.ChannelInboundHandlerAdapter;

 import io.netty.util.ReferenceCountUtil;

 import static org.apache.giraph.conf.GiraphConstants.NETTY_SIMULATE_FIRST_RESPONSE_FAILED;

 /**

  * Generic handler of responses.

  */

 public class ResponseClientHandler extends ChannelInboundHandlerAdapter {

   /** Class logger */

   private static final Logger LOG =

       Logger.getLogger(ResponseClientHandler.class);

   /** Already dropped first response? (used if dropFirstResponse == true) */

   private static volatile boolean ALREADY_DROPPED_FIRST_RESPONSE = false;

   /** Drop first response (used for simulating failure) */

   private final boolean dropFirstResponse;

   /** Netty client that does the actual I/O and keeps track of open requests */

   private final NettyClient nettyClient;

   /**

    * Constructor.

    * @param nettyClient Client that does the actual I/O

    * @param conf Configuration

    */

   public ResponseClientHandler(NettyClient nettyClient, Configuration conf) {

     this.nettyClient = nettyClient;

     dropFirstResponse = NETTY_SIMULATE_FIRST_RESPONSE_FAILED.get(conf);

   }

   @Override

   public void channelRead(ChannelHandlerContext ctx, Object msg)

     throws Exception {

     if (!(msg instanceof ByteBuf)) {

       throw new IllegalStateException("channelRead: Got a " +

           "non-ByteBuf message " + msg);

     }

     ByteBuf buf = (ByteBuf) msg;

     int senderId = -1;

     long requestId = -1;

     int response = -1;

     try {

       senderId = buf.readInt();

       requestId = buf.readLong();

       response = buf.readInt();

     } catch (IndexOutOfBoundsException e) {

       throw new IllegalStateException(

           "channelRead: Got IndexOutOfBoundsException ", e);

     }

     ReferenceCountUtil.release(buf);

     boolean shouldDrop = false;

     // Simulate a failed response on the first response (if desired)

     if (dropFirstResponse && !ALREADY_DROPPED_FIRST_RESPONSE) {

       LOG.info("channelRead: Simulating dropped response " + response +

           " for request " + requestId);

       setAlreadyDroppedFirstResponse();

       shouldDrop = true;

     }

     nettyClient.messageReceived(senderId, requestId, response, shouldDrop);

   }

   /**

    * Set already dropped first response flag

    */

   private static void setAlreadyDroppedFirstResponse() {

     ALREADY_DROPPED_FIRST_RESPONSE = true;

   }

   @Override

   public void channelInactive(ChannelHandlerContext ctx) throws Exception {

     if (LOG.isDebugEnabled()) {

       LOG.debug("channelClosed: Closed the channel on " +

           ctx.channel().remoteAddress());

     }

     ctx.fireChannelInactive();

   }

   @Override

   public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)

     throws Exception {

     LOG.warn("exceptionCaught: Channel channelId=" +

         ctx.channel().hashCode() + " failed with remote address " +

         ctx.channel().remoteAddress(), cause);

   }

 }