/*
    * Copyright 2012 The Netty Project
    *
    * The Netty Project 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:
    *
    *   https://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 io.netty5.handler.timeout;
  
  import io.netty5.bootstrap.ServerBootstrap;
  import io.netty5.channel.Channel;
  import io.netty5.channel.ChannelHandler;
  import io.netty5.channel.ChannelHandlerContext;
  import io.netty5.channel.ChannelInitializer;
  import io.netty5.util.concurrent.Future;
  import io.netty5.util.concurrent.FutureListener;
  
  import java.util.concurrent.TimeUnit;
  
  import static java.util.Objects.requireNonNull;
  
  /**
   * Triggers an {@link IdleStateEvent} when a {@link Channel} has not performed
   * read, write, or both operation for a while.
   *
   * <h3>Supported idle states</h3>
   * <table border="1">
   * <tr>
   * <th>Property</th><th>Meaning</th>
   * </tr>
   * <tr>
   * <td>{@code readerIdleTime}</td>
   * <td>an {@link IdleStateEvent} whose state is {@link IdleState#READER_IDLE}
   *     will be triggered when no read was performed for the specified period of
   *     time.  Specify {@code 0} to disable.</td>
   * </tr>
   * <tr>
   * <td>{@code writerIdleTime}</td>
   * <td>an {@link IdleStateEvent} whose state is {@link IdleState#WRITER_IDLE}
   *     will be triggered when no write was performed for the specified period of
   *     time.  Specify {@code 0} to disable.</td>
   * </tr>
   * <tr>
   * <td>{@code allIdleTime}</td>
   * <td>an {@link IdleStateEvent} whose state is {@link IdleState#ALL_IDLE}
   *     will be triggered when neither read nor write was performed for the
   *     specified period of time.  Specify {@code 0} to disable.</td>
   * </tr>
   * </table>
   *
   * <pre>
   * // An example that sends a ping message when there is no outbound traffic
   * // for 30 seconds.  The connection is closed when there is no inbound traffic
   * // for 60 seconds.
   *
   * public class MyChannelInitializer extends {@link ChannelInitializer}&lt;{@link Channel}&gt; {
   *     {@code @Override}
   *     public void initChannel({@link Channel} channel) {
   *         channel.pipeline().addLast("idleStateHandler", new {@link IdleStateHandler}(60, 30, 0));
   *         channel.pipeline().addLast("myHandler", new MyHandler());
   *     }
   * }
   *
   * // Handler should handle the {@link IdleStateEvent} triggered by {@link IdleStateHandler}.
   * public class MyHandler implements {@link ChannelHandler} {
   *     {@code @Override}
   *     public void userEventTriggered({@link ChannelHandlerContext} ctx, {@link Object} evt) throws {@link Exception} {
   *         if (evt instanceof {@link IdleStateEvent}) {
   *             {@link IdleStateEvent} e = ({@link IdleStateEvent}) evt;
   *             if (e.state() == {@link IdleState}.READER_IDLE) {
   *                 ctx.close();
   *             } else if (e.state() == {@link IdleState}.WRITER_IDLE) {
   *                 ctx.writeAndFlush(new PingMessage());
   *             }
   *         }
   *     }
   * }
   *
   * {@link ServerBootstrap} bootstrap = ...;
   * ...
   * bootstrap.childHandler(new MyChannelInitializer());
   * ...
   * </pre>
   *
   * @see ReadTimeoutHandler
   * @see WriteTimeoutHandler
   */
  public class IdleStateHandler implements ChannelHandler {
      private static final long MIN_TIMEOUT_NANOS = TimeUnit.MILLISECONDS.toNanos(1);
  
      // Not create a new ChannelFutureListeners per write operation to reduce GC pressure.
     private final FutureListener<Void> writeListener = future -> {
         lastWriteTime = ticksInNanos();
         firstWriterIdleEvent = firstAllIdleEvent = true;
     };
 
     private final long readerIdleTimeNanos;
     private final long writerIdleTimeNanos;
     private final long allIdleTimeNanos;
 
     private Future<?> readerIdleTimeout;
     private long lastReadTime;
     private boolean firstReaderIdleEvent = true;
 
     private Future<?> writerIdleTimeout;
     private long lastWriteTime;
     private boolean firstWriterIdleEvent = true;
 
     private Future<?> allIdleTimeout;
     private boolean firstAllIdleEvent = true;
 
     private byte state; // 0 - none, 1 - initialized, 2 - destroyed
     private boolean reading;
 
     /**
      * Creates a new instance firing {@link IdleStateEvent}s.
      *
      * @param readerIdleTimeSeconds
      *        an {@link IdleStateEvent} whose state is {@link IdleState#READER_IDLE}
      *        will be triggered when no read was performed for the specified
      *        period of time.  Specify {@code 0} to disable.
      * @param writerIdleTimeSeconds
      *        an {@link IdleStateEvent} whose state is {@link IdleState#WRITER_IDLE}
      *        will be triggered when no write was performed for the specified
      *        period of time.  Specify {@code 0} to disable.
      * @param allIdleTimeSeconds
      *        an {@link IdleStateEvent} whose state is {@link IdleState#ALL_IDLE}
      *        will be triggered when neither read nor write was performed for
      *        the specified period of time.  Specify {@code 0} to disable.
      */
     public IdleStateHandler(
             int readerIdleTimeSeconds,
             int writerIdleTimeSeconds,
             int allIdleTimeSeconds) {
 
         this(readerIdleTimeSeconds, writerIdleTimeSeconds, allIdleTimeSeconds,
              TimeUnit.SECONDS);
     }
 
     /**
      * Creates a new instance firing {@link IdleStateEvent}s.
      * @param readerIdleTime
      *        an {@link IdleStateEvent} whose state is {@link IdleState#READER_IDLE}
      *        will be triggered when no read was performed for the specified
      *        period of time.  Specify {@code 0} to disable.
      * @param writerIdleTime
      *        an {@link IdleStateEvent} whose state is {@link IdleState#WRITER_IDLE}
      *        will be triggered when no write was performed for the specified
      *        period of time.  Specify {@code 0} to disable.
      * @param allIdleTime
      *        an {@link IdleStateEvent} whose state is {@link IdleState#ALL_IDLE}
      *        will be triggered when neither read nor write was performed for
      *        the specified period of time.  Specify {@code 0} to disable.
      * @param unit
      *        the {@link TimeUnit} of {@code readerIdleTime},
      *        {@code writeIdleTime}, and {@code allIdleTime}
      */
     public IdleStateHandler(long readerIdleTime, long writerIdleTime, long allIdleTime,
             TimeUnit unit) {
         requireNonNull(unit, "unit");
 
         if (readerIdleTime <= 0) {
             readerIdleTimeNanos = 0;
         } else {
             readerIdleTimeNanos = Math.max(unit.toNanos(readerIdleTime), MIN_TIMEOUT_NANOS);
         }
         if (writerIdleTime <= 0) {
             writerIdleTimeNanos = 0;
         } else {
             writerIdleTimeNanos = Math.max(unit.toNanos(writerIdleTime), MIN_TIMEOUT_NANOS);
         }
         if (allIdleTime <= 0) {
             allIdleTimeNanos = 0;
         } else {
             allIdleTimeNanos = Math.max(unit.toNanos(allIdleTime), MIN_TIMEOUT_NANOS);
         }
     }
 
     /**
      * Return the readerIdleTime that was given when instance this class in milliseconds.
      *
      */
     public long getReaderIdleTimeInMillis() {
         return TimeUnit.NANOSECONDS.toMillis(readerIdleTimeNanos);
     }
 
     /**
      * Return the writerIdleTime that was given when instance this class in milliseconds.
      *
      */
     public long getWriterIdleTimeInMillis() {
         return TimeUnit.NANOSECONDS.toMillis(writerIdleTimeNanos);
     }
 
     /**
      * Return the allIdleTime that was given when instance this class in milliseconds.
      *
      */
     public long getAllIdleTimeInMillis() {
         return TimeUnit.NANOSECONDS.toMillis(allIdleTimeNanos);
     }
 
     @Override
     public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
         if (ctx.channel().isActive() && ctx.channel().isRegistered()) {
             // channelActive() event has been fired already, which means this.channelActive() will
             // not be invoked. We have to initialize here instead.
             initialize(ctx);
         } else {
             // channelActive() event has not been fired yet.  this.channelActive() will be invoked
             // and initialization will occur there.
         }
     }
 
     @Override
     public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
         destroy();
     }
 
     @Override
     public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
         // Initialize early if channel is active already.
         if (ctx.channel().isActive()) {
             initialize(ctx);
         }
         ctx.fireChannelRegistered();
     }
 
     @Override
     public void channelActive(ChannelHandlerContext ctx) throws Exception {
         // This method will be invoked only if this handler was added
         // before channelActive() event is fired.  If a user adds this handler
         // after the channelActive() event, initialize() will be called by beforeAdd().
         initialize(ctx);
         ctx.fireChannelActive();
     }
 
     @Override
     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
         destroy();
         ctx.fireChannelInactive();
     }
 
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
         if (readerIdleTimeNanos > 0 || allIdleTimeNanos > 0) {
             reading = true;
             firstReaderIdleEvent = firstAllIdleEvent = true;
         }
         ctx.fireChannelRead(msg);
     }
 
     @Override
     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
         if ((readerIdleTimeNanos > 0 || allIdleTimeNanos > 0) && reading) {
             lastReadTime = ticksInNanos();
             reading = false;
         }
         ctx.fireChannelReadComplete();
     }
 
     @Override
     public Future<Void> write(ChannelHandlerContext ctx, Object msg) {
         Future<Void> future = ctx.write(msg);
         // Allow writing with void promise if handler is only configured for read timeout events.
         if (writerIdleTimeNanos > 0 || allIdleTimeNanos > 0) {
             future.addListener(writeListener);
         }
         return future;
     }
 
     private void initialize(ChannelHandlerContext ctx) {
         // Avoid the case where destroy() is called before scheduling timeouts.
         // See: https://github.com/netty/netty/issues/143
         switch (state) {
         case 1:
         case 2:
             return;
         default:
              break;
         }
 
         state = 1;
 
         lastReadTime = lastWriteTime = ticksInNanos();
         if (readerIdleTimeNanos > 0) {
             readerIdleTimeout = schedule(ctx, new ReaderIdleTimeoutTask(ctx),
                     readerIdleTimeNanos, TimeUnit.NANOSECONDS);
         }
         if (writerIdleTimeNanos > 0) {
             writerIdleTimeout = schedule(ctx, new WriterIdleTimeoutTask(ctx),
                     writerIdleTimeNanos, TimeUnit.NANOSECONDS);
         }
         if (allIdleTimeNanos > 0) {
             allIdleTimeout = schedule(ctx, new AllIdleTimeoutTask(ctx),
                     allIdleTimeNanos, TimeUnit.NANOSECONDS);
         }
     }
 
     /**
      * This method is visible for testing!
      */
     long ticksInNanos() {
         return System.nanoTime();
     }
 
     /**
      * This method is visible for testing!
      */
     Future<?> schedule(ChannelHandlerContext ctx, Runnable task, long delay, TimeUnit unit) {
         return ctx.executor().schedule(task, delay, unit);
     }
 
     private void destroy() {
         state = 2;
 
         if (readerIdleTimeout != null) {
             readerIdleTimeout.cancel();
             readerIdleTimeout = null;
         }
         if (writerIdleTimeout != null) {
             writerIdleTimeout.cancel();
             writerIdleTimeout = null;
         }
         if (allIdleTimeout != null) {
             allIdleTimeout.cancel();
             allIdleTimeout = null;
         }
     }
 
     /**
      * Is called when an {@link IdleStateEvent} should be fired. This implementation calls
      * {@link ChannelHandlerContext#fireChannelInboundEvent(Object)}.
      */
     protected void channelIdle(ChannelHandlerContext ctx, IdleStateEvent evt) throws Exception {
         ctx.fireChannelInboundEvent(evt);
     }
 
     /**
      * Returns a {@link IdleStateEvent}.
      */
     protected IdleStateEvent newIdleStateEvent(IdleState state, boolean first) {
         switch (state) {
             case ALL_IDLE:
                 return first ? IdleStateEvent.FIRST_ALL_IDLE_STATE_EVENT : IdleStateEvent.ALL_IDLE_STATE_EVENT;
             case READER_IDLE:
                 return first ? IdleStateEvent.FIRST_READER_IDLE_STATE_EVENT : IdleStateEvent.READER_IDLE_STATE_EVENT;
             case WRITER_IDLE:
                 return first ? IdleStateEvent.FIRST_WRITER_IDLE_STATE_EVENT : IdleStateEvent.WRITER_IDLE_STATE_EVENT;
             default:
                 throw new IllegalArgumentException("Unhandled: state=" + state + ", first=" + first);
         }
     }
 
     private abstract static class AbstractIdleTask implements Runnable {
 
         private final ChannelHandlerContext ctx;
 
         AbstractIdleTask(ChannelHandlerContext ctx) {
             this.ctx = ctx;
         }
 
         @Override
         public void run() {
             if (!ctx.channel().isOpen()) {
                 return;
             }
 
             run(ctx);
         }
 
         protected abstract void run(ChannelHandlerContext ctx);
     }
 
     private final class ReaderIdleTimeoutTask extends AbstractIdleTask {
 
         ReaderIdleTimeoutTask(ChannelHandlerContext ctx) {
             super(ctx);
         }
 
         @Override
         protected void run(ChannelHandlerContext ctx) {
             long nextDelay = readerIdleTimeNanos;
             if (!reading) {
                 nextDelay -= ticksInNanos() - lastReadTime;
             }
 
             if (nextDelay <= 0) {
                 // Reader is idle - set a new timeout and notify the callback.
                 readerIdleTimeout = schedule(ctx, this, readerIdleTimeNanos, TimeUnit.NANOSECONDS);
 
                 boolean first = firstReaderIdleEvent;
                 firstReaderIdleEvent = false;
 
                 try {
                     IdleStateEvent event = newIdleStateEvent(IdleState.READER_IDLE, first);
                     channelIdle(ctx, event);
                 } catch (Throwable t) {
                     ctx.fireChannelExceptionCaught(t);
                 }
             } else {
                 // Read occurred before the timeout - set a new timeout with shorter delay.
                 readerIdleTimeout = schedule(ctx, this, nextDelay, TimeUnit.NANOSECONDS);
             }
         }
     }
 
     private final class WriterIdleTimeoutTask extends AbstractIdleTask {
 
         WriterIdleTimeoutTask(ChannelHandlerContext ctx) {
             super(ctx);
         }
 
         @Override
         protected void run(ChannelHandlerContext ctx) {
 
             long lastWriteTime = IdleStateHandler.this.lastWriteTime;
             long nextDelay = writerIdleTimeNanos - (ticksInNanos() - lastWriteTime);
             if (nextDelay <= 0) {
                 // Writer is idle - set a new timeout and notify the callback.
                 writerIdleTimeout = schedule(ctx, this, writerIdleTimeNanos, TimeUnit.NANOSECONDS);
 
                 boolean first = firstWriterIdleEvent;
                 firstWriterIdleEvent = false;
 
                 try {
                     IdleStateEvent event = newIdleStateEvent(IdleState.WRITER_IDLE, first);
                     channelIdle(ctx, event);
                 } catch (Throwable t) {
                     ctx.fireChannelExceptionCaught(t);
                 }
             } else {
                 // Write occurred before the timeout - set a new timeout with shorter delay.
                 writerIdleTimeout = schedule(ctx, this, nextDelay, TimeUnit.NANOSECONDS);
             }
         }
     }
 
     private final class AllIdleTimeoutTask extends AbstractIdleTask {
 
         AllIdleTimeoutTask(ChannelHandlerContext ctx) {
             super(ctx);
         }
 
         @Override
         protected void run(ChannelHandlerContext ctx) {
 
             long nextDelay = allIdleTimeNanos;
             if (!reading) {
                 nextDelay -= ticksInNanos() - Math.max(lastReadTime, lastWriteTime);
             }
             if (nextDelay <= 0) {
                 // Both reader and writer are idle - set a new timeout and
                 // notify the callback.
                 allIdleTimeout = schedule(ctx, this, allIdleTimeNanos, TimeUnit.NANOSECONDS);
 
                 boolean first = firstAllIdleEvent;
                 firstAllIdleEvent = false;
 
                 try {
                     IdleStateEvent event = newIdleStateEvent(IdleState.ALL_IDLE, first);
                     channelIdle(ctx, event);
                 } catch (Throwable t) {
                     ctx.fireChannelExceptionCaught(t);
                 }
             } else {
                 // Either read or write occurred before the timeout - set a new
                 // timeout with shorter delay.
                 allIdleTimeout = schedule(ctx, this, nextDelay, TimeUnit.NANOSECONDS);
             }
         }
     }
 }