/*

  * 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.flow_control.FlowControl;

 import org.apache.giraph.comm.requests.WritableRequest;

 import org.apache.giraph.conf.ImmutableClassesGiraphConfiguration;

 import org.apache.giraph.graph.TaskInfo;

 import org.apache.giraph.time.SystemTime;

 import org.apache.giraph.time.Time;

 import org.apache.giraph.time.Times;

 import org.apache.log4j.Logger;

 import io.netty.buffer.ByteBuf;

 import io.netty.channel.ChannelHandlerContext;

 import io.netty.channel.ChannelInboundHandlerAdapter;

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

 /**

  * Generic handler of requests.

  *

  * @param <R> Request type

  */

 public abstract class RequestServerHandler<R> extends

   ChannelInboundHandlerAdapter {

   /** Number of bytes in the encoded response */

   public static final int RESPONSE_BYTES = 16;

   /** Time class to use */

   private static Time TIME = SystemTime.get();

   /** Class logger */

   private static final Logger LOG =

       Logger.getLogger(RequestServerHandler.class);

   /** Already closed first request? */

   private static volatile boolean ALREADY_CLOSED_FIRST_REQUEST = false;

   /** Flow control used in sending requests */

   protected FlowControl flowControl;

   /** Close connection on first request (used for simulating failure) */

   private final boolean closeFirstRequest;

   /** Request reserved map (for exactly one semantics) */

   private final WorkerRequestReservedMap workerRequestReservedMap;

   /** My task info */

   private final TaskInfo myTaskInfo;

   /** Start nanoseconds for the processing time */

   private long startProcessingNanoseconds = -1;

   /** Handler for uncaught exceptions */

   private final Thread.UncaughtExceptionHandler exceptionHandler;

   /**

    * Constructor

    *

    * @param workerRequestReservedMap Worker request reservation map

    * @param conf Configuration

    * @param myTaskInfo Current task info

    * @param exceptionHandler Handles uncaught exceptions

    */

   public RequestServerHandler(

       WorkerRequestReservedMap workerRequestReservedMap,

       ImmutableClassesGiraphConfiguration conf,

       TaskInfo myTaskInfo,

       Thread.UncaughtExceptionHandler exceptionHandler) {

     this.workerRequestReservedMap = workerRequestReservedMap;

     closeFirstRequest = NETTY_SIMULATE_FIRST_REQUEST_CLOSED.get(conf);

     this.myTaskInfo = myTaskInfo;

     this.exceptionHandler = exceptionHandler;

   }

   @Override

   public void channelRead(ChannelHandlerContext ctx, Object msg)

     throws Exception {

     if (LOG.isTraceEnabled()) {

       LOG.trace("messageReceived: Got " + msg.getClass());

     }

     WritableRequest request = (WritableRequest) msg;

     // Simulate a closed connection on the first request (if desired)

     if (closeFirstRequest && !ALREADY_CLOSED_FIRST_REQUEST) {

       LOG.info("messageReceived: Simulating closing channel on first " +

           "request " + request.getRequestId() + " from " +

           request.getClientId());

       setAlreadyClosedFirstRequest();

       ctx.close();

       return;

     }

     // Only execute this request exactly once

     AckSignalFlag alreadyDone = AckSignalFlag.DUPLICATE_REQUEST;

     if (workerRequestReservedMap.reserveRequest(

         request.getClientId(),

         request.getRequestId())) {

       if (LOG.isDebugEnabled()) {

         startProcessingNanoseconds = TIME.getNanoseconds();

       }

       processRequest((R) request);

       if (LOG.isDebugEnabled()) {

         LOG.debug("messageReceived: Processing client " +

             request.getClientId() + ", " +

             "requestId " + request.getRequestId() +

             ", " +  request.getType() + " took " +

             Times.getNanosSince(TIME, startProcessingNanoseconds) + " ns");

       }

       alreadyDone = AckSignalFlag.NEW_REQUEST;

     } else {

       LOG.info("messageReceived: Request id " +

           request.getRequestId() + " from client " +

           request.getClientId() +

           " was already processed, " +

           "not processing again.");

     }

     // Send the response with the request id

     ByteBuf buffer = ctx.alloc().buffer(RESPONSE_BYTES);

     buffer.writeInt(myTaskInfo.getTaskId());

     buffer.writeLong(request.getRequestId());

     int signal =

         flowControl.calculateResponse(alreadyDone, request.getClientId());

     buffer.writeInt(signal);

     ctx.write(buffer);

   }

   /**

    * Set the flag indicating already closed first request

    */

   private static void setAlreadyClosedFirstRequest() {

     ALREADY_CLOSED_FIRST_REQUEST = true;

   }

   /**

    * Process request

    *

    * @param request Request to process

    */

   public abstract void processRequest(R request);

   @Override

   public void channelActive(ChannelHandlerContext ctx) throws Exception {

     if (LOG.isDebugEnabled()) {

       LOG.debug("channelActive: Connected the channel on " +

           ctx.channel().remoteAddress());

     }

     ctx.fireChannelActive();

   }

   @Override

   public void channelInactive(ChannelHandlerContext ctx) throws Exception {

     if (LOG.isDebugEnabled()) {

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

           ctx.channel().remoteAddress());

     }

     ctx.fireChannelInactive();

   }

   @Override

   public void exceptionCaught(

       ChannelHandlerContext ctx, Throwable cause) throws Exception {

     exceptionHandler.uncaughtException(Thread.currentThread(), cause);

   }

   /**

    * Factory for {@link RequestServerHandler}

    */

   public interface Factory {

     /**

      * Create new {@link RequestServerHandler}

      *

      * @param workerRequestReservedMap Worker request reservation map

      * @param conf Configuration to use

      * @param myTaskInfo Current task info

      * @param exceptionHandler Handles uncaught exceptions

      * @return New {@link RequestServerHandler}

      */

     RequestServerHandler newHandler(

         WorkerRequestReservedMap workerRequestReservedMap,

         ImmutableClassesGiraphConfiguration conf,

         TaskInfo myTaskInfo,

         Thread.UncaughtExceptionHandler exceptionHandler);

     /**

      * Inform the factory about the flow control policy used (this method should

      * be called before any call to `#newHandle()`)

      *

      * @param flowControl reference to flow control used

      */

     void setFlowControl(FlowControl flowControl);

   }

 }