/*
    * 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.netty.channel.nio;
  
  import io.netty.buffer.ByteBuf;
  import io.netty.buffer.ByteBufAllocator;
  import io.netty.buffer.ByteBufUtil;
  import io.netty.buffer.Unpooled;
  import io.netty.channel.AbstractChannel;
  import io.netty.channel.Channel;
  import io.netty.channel.ChannelException;
  import io.netty.channel.ChannelFuture;
  import io.netty.channel.ChannelFutureListener;
  import io.netty.channel.ChannelPromise;
  import io.netty.channel.ConnectTimeoutException;
  import io.netty.channel.EventLoop;
  import io.netty.channel.IoEvent;
  import io.netty.channel.IoEventLoop;
  import io.netty.channel.IoEventLoopGroup;
  import io.netty.channel.IoRegistration;
  import io.netty.util.ReferenceCountUtil;
  import io.netty.util.ReferenceCounted;
  import io.netty.util.concurrent.Future;
  import io.netty.util.internal.ObjectUtil;
  import io.netty.util.internal.logging.InternalLogger;
  import io.netty.util.internal.logging.InternalLoggerFactory;
  
  import java.io.IOException;
  import java.net.SocketAddress;
  import java.nio.channels.CancelledKeyException;
  import java.nio.channels.ClosedChannelException;
  import java.nio.channels.ConnectionPendingException;
  import java.nio.channels.SelectableChannel;
  import java.nio.channels.SelectionKey;
  import java.util.concurrent.TimeUnit;
  
  /**
   * Abstract base class for {@link Channel} implementations which use a Selector based approach.
   */
  public abstract class AbstractNioChannel extends AbstractChannel {
  
      private static final InternalLogger logger =
              InternalLoggerFactory.getInstance(AbstractNioChannel.class);
  
      private final SelectableChannel ch;
      protected final int readInterestOp;
      protected final NioIoOps readOps;
      volatile NioIoRegistration registration;
      boolean readPending;
      private final Runnable clearReadPendingRunnable = new Runnable() {
          @Override
          public void run() {
              clearReadPending0();
          }
      };
  
      /**
       * The future of the current connection attempt.  If not null, subsequent
       * connection attempts will fail.
       */
      private ChannelPromise connectPromise;
      private Future<?> connectTimeoutFuture;
      private SocketAddress requestedRemoteAddress;
  
      /**
       * Create a new instance
       *
       * @param parent            the parent {@link Channel} by which this instance was created. May be {@code null}
       * @param ch                the underlying {@link SelectableChannel} on which it operates
       * @param readOps           the ops to set to receive data from the {@link SelectableChannel}
       */
      protected AbstractNioChannel(Channel parent, SelectableChannel ch, int readOps) {
          this(parent, ch, NioIoOps.valueOf(readOps));
      }
  
      protected AbstractNioChannel(Channel parent, SelectableChannel ch, NioIoOps readOps) {
          super(parent);
          this.ch = ch;
          this.readInterestOp = ObjectUtil.checkNotNull(readOps, "readOps").value;
          this.readOps = readOps;
          try {
              ch.configureBlocking(false);
          } catch (IOException e) {
              try {
                  ch.close();
              } catch (IOException e2) {
                 logger.warn(
                             "Failed to close a partially initialized socket.", e2);
             }
 
             throw new ChannelException("Failed to enter non-blocking mode.", e);
         }
     }
 
     protected void addAndSubmit(NioIoOps addOps) {
         int interestOps = selectionKey().interestOps();
         if (!addOps.isIncludedIn(interestOps)) {
             try {
                 registration().submit(NioIoOps.valueOf(interestOps).with(addOps));
             } catch (Exception e) {
                 throw new ChannelException(e);
             }
         }
     }
 
     protected void removeAndSubmit(NioIoOps removeOps) {
         int interestOps = selectionKey().interestOps();
         if (removeOps.isIncludedIn(interestOps)) {
             try {
                 registration().submit(NioIoOps.valueOf(interestOps).without(removeOps));
             } catch (Exception e) {
                 throw new ChannelException(e);
             }
         }
     }
 
     @Override
     public boolean isOpen() {
         return ch.isOpen();
     }
 
     @Override
     public NioUnsafe unsafe() {
         return (NioUnsafe) super.unsafe();
     }
 
     protected SelectableChannel javaChannel() {
         return ch;
     }
 
     /**
      * Return the current {@link SelectionKey}
      *
      * @deprecated use {@link #registration}.
      */
     @Deprecated
     protected SelectionKey selectionKey() {
         return registration().selectionKey();
     }
 
     @SuppressWarnings("unchecked")
     protected NioIoRegistration registration() {
         assert registration != null;
         return registration;
     }
 
     /**
      * @deprecated No longer supported.
      * No longer supported.
      */
     @Deprecated
     protected boolean isReadPending() {
         return readPending;
     }
 
     /**
      * @deprecated Use {@link #clearReadPending()} if appropriate instead.
      * No longer supported.
      */
     @Deprecated
     protected void setReadPending(final boolean readPending) {
         if (isRegistered()) {
             EventLoop eventLoop = eventLoop();
             if (eventLoop.inEventLoop()) {
                 setReadPending0(readPending);
             } else {
                 eventLoop.execute(new Runnable() {
                     @Override
                     public void run() {
                         setReadPending0(readPending);
                     }
                 });
             }
         } else {
             // Best effort if we are not registered yet clear readPending.
             // NB: We only set the boolean field instead of calling clearReadPending0(), because the SelectionKey is
             // not set yet so it would produce an assertion failure.
             this.readPending = readPending;
         }
     }
 
     /**
      * Set read pending to {@code false}.
      */
     protected final void clearReadPending() {
         if (isRegistered()) {
             EventLoop eventLoop = eventLoop();
             if (eventLoop.inEventLoop()) {
                 clearReadPending0();
             } else {
                 eventLoop.execute(clearReadPendingRunnable);
             }
         } else {
             // Best effort if we are not registered yet clear readPending. This happens during channel initialization.
             // NB: We only set the boolean field instead of calling clearReadPending0(), because the SelectionKey is
             // not set yet so it would produce an assertion failure.
             readPending = false;
         }
     }
 
     private void setReadPending0(boolean readPending) {
         this.readPending = readPending;
         if (!readPending) {
             ((AbstractNioUnsafe) unsafe()).removeReadOp();
         }
     }
 
     private void clearReadPending0() {
         readPending = false;
         ((AbstractNioUnsafe) unsafe()).removeReadOp();
     }
 
     /**
      * Special {@link Unsafe} sub-type which allows to access the underlying {@link SelectableChannel}
      */
     public interface NioUnsafe extends Unsafe {
         /**
          * Return underlying {@link SelectableChannel}
          */
         SelectableChannel ch();
 
         /**
          * Finish connect
          */
         void finishConnect();
 
         /**
          * Read from underlying {@link SelectableChannel}
          */
         void read();
 
         void forceFlush();
     }
 
     protected abstract class AbstractNioUnsafe extends AbstractUnsafe implements NioUnsafe, NioIoHandle {
         @Override
         public void close() {
             close(voidPromise());
         }
 
         @Override
         public SelectableChannel selectableChannel() {
             return ch();
         }
 
         Channel channel() {
             return AbstractNioChannel.this;
         }
 
         protected final void removeReadOp() {
             NioIoRegistration registration = registration();
             // Check first if the key is still valid as it may be canceled as part of the deregistration
             // from the EventLoop
             // See https://github.com/netty/netty/issues/2104
             if (!registration.isValid()) {
                 return;
             }
             removeAndSubmit(readOps);
         }
 
         @Override
         public final SelectableChannel ch() {
             return javaChannel();
         }
 
         @Override
         public final void connect(
                 final SocketAddress remoteAddress, final SocketAddress localAddress, final ChannelPromise promise) {
             // Don't mark the connect promise as uncancellable as in fact we can cancel it as it is using
             // non-blocking io.
             if (promise.isDone() || !ensureOpen(promise)) {
                 return;
             }
 
             try {
                 if (connectPromise != null) {
                     // Already a connect in process.
                     throw new ConnectionPendingException();
                 }
 
                 boolean wasActive = isActive();
                 if (doConnect(remoteAddress, localAddress)) {
                     fulfillConnectPromise(promise, wasActive);
                 } else {
                     connectPromise = promise;
                     requestedRemoteAddress = remoteAddress;
 
                     // Schedule connect timeout.
                     final int connectTimeoutMillis = config().getConnectTimeoutMillis();
                     if (connectTimeoutMillis > 0) {
                         connectTimeoutFuture = eventLoop().schedule(new Runnable() {
                             @Override
                             public void run() {
                                 ChannelPromise connectPromise = AbstractNioChannel.this.connectPromise;
                                 if (connectPromise != null && !connectPromise.isDone()
                                         && connectPromise.tryFailure(new ConnectTimeoutException(
                                                 "connection timed out after " + connectTimeoutMillis + " ms: " +
                                                         remoteAddress))) {
                                     close(voidPromise());
                                 }
                             }
                         }, connectTimeoutMillis, TimeUnit.MILLISECONDS);
                     }
 
                     promise.addListener(new ChannelFutureListener() {
                         @Override
                         public void operationComplete(ChannelFuture future) {
                             // If the connect future is cancelled we also cancel the timeout and close the
                             // underlying socket.
                             if (future.isCancelled()) {
                                 if (connectTimeoutFuture != null) {
                                     connectTimeoutFuture.cancel(false);
                                 }
                                 connectPromise = null;
                                 close(voidPromise());
                             }
                         }
                     });
                 }
             } catch (Throwable t) {
                 promise.tryFailure(annotateConnectException(t, remoteAddress));
                 closeIfClosed();
             }
         }
 
         private void fulfillConnectPromise(ChannelPromise promise, boolean wasActive) {
             if (promise == null) {
                 // Closed via cancellation and the promise has been notified already.
                 return;
             }
 
             // Get the state as trySuccess() may trigger an ChannelFutureListener that will close the Channel.
             // We still need to ensure we call fireChannelActive() in this case.
             boolean active = isActive();
 
             // trySuccess() will return false if a user cancelled the connection attempt.
             boolean promiseSet = promise.trySuccess();
 
             // Regardless if the connection attempt was cancelled, channelActive() event should be triggered,
             // because what happened is what happened.
             if (!wasActive && active) {
                 pipeline().fireChannelActive();
             }
 
             // If a user cancelled the connection attempt, close the channel, which is followed by channelInactive().
             if (!promiseSet) {
                 close(voidPromise());
             }
         }
 
         private void fulfillConnectPromise(ChannelPromise promise, Throwable cause) {
             if (promise == null) {
                 // Closed via cancellation and the promise has been notified already.
                 return;
             }
 
             // Use tryFailure() instead of setFailure() to avoid the race against cancel().
             promise.tryFailure(cause);
             closeIfClosed();
         }
 
         @Override
         public final void finishConnect() {
             // Note this method is invoked by the event loop only if the connection attempt was
             // neither cancelled nor timed out.
 
             assert eventLoop().inEventLoop();
 
             try {
                 boolean wasActive = isActive();
                 doFinishConnect();
                 fulfillConnectPromise(connectPromise, wasActive);
             } catch (Throwable t) {
                 fulfillConnectPromise(connectPromise, annotateConnectException(t, requestedRemoteAddress));
             } finally {
                 // Check for null as the connectTimeoutFuture is only created if a connectTimeoutMillis > 0 is used
                 // See https://github.com/netty/netty/issues/1770
                 if (connectTimeoutFuture != null) {
                     connectTimeoutFuture.cancel(false);
                 }
                 connectPromise = null;
             }
         }
 
         @Override
         protected final void flush0() {
             // Flush immediately only when there's no pending flush.
             // If there's a pending flush operation, event loop will call forceFlush() later,
             // and thus there's no need to call it now.
             if (!isFlushPending()) {
                 super.flush0();
             }
         }
 
         @Override
         public final void forceFlush() {
             // directly call super.flush0() to force a flush now
             super.flush0();
         }
 
         private boolean isFlushPending() {
             NioIoRegistration registration = registration();
             return registration.isValid() && NioIoOps.WRITE.isIncludedIn(registration.selectionKey().interestOps());
         }
 
         @Override
         public void handle(IoRegistration registration, IoEvent event) {
             try {
                 NioIoRegistration nioRegistration = (NioIoRegistration) registration;
                 NioIoEvent nioEvent = (NioIoEvent) event;
                 NioIoOps nioReadyOps = nioEvent.ops();
                 // We first need to call finishConnect() before try to trigger a read(...) or write(...) as otherwise
                 // the NIO JDK channel implementation may throw a NotYetConnectedException.
                 if (nioReadyOps.contains(NioIoOps.CONNECT)) {
                     // remove OP_CONNECT as otherwise Selector.select(..) will always return without blocking
                     // See https://github.com/netty/netty/issues/924
                     removeAndSubmit(NioIoOps.CONNECT);
 
                     unsafe().finishConnect();
                 }
 
                 // Process OP_WRITE first as we may be able to write some queued buffers and so free memory.
                 if (nioReadyOps.contains(NioIoOps.WRITE)) {
                     // Call forceFlush which will also take care of clear the OP_WRITE once there is nothing left to
                     // write
                     forceFlush();
                 }
 
                 // Also check for readOps of 0 to workaround possible JDK bug which may otherwise lead
                 // to a spin loop
                 if (nioReadyOps.contains(NioIoOps.READ_AND_ACCEPT) || nioReadyOps.equals(NioIoOps.NONE)) {
                     read();
                 }
             } catch (CancelledKeyException ignored) {
                 close(voidPromise());
             }
         }
     }
 
     @Override
     protected boolean isCompatible(EventLoop loop) {
         return loop instanceof IoEventLoop && ((IoEventLoopGroup) loop).isCompatible(AbstractNioUnsafe.class);
     }
 
     @SuppressWarnings("unchecked")
     @Override
     protected void doRegister(ChannelPromise promise) {
         assert registration == null;
         ((IoEventLoop) eventLoop()).register((AbstractNioUnsafe) unsafe()).addListener(f -> {
             if (f.isSuccess()) {
                 registration = (NioIoRegistration) f.getNow();
                 promise.setSuccess();
             } else {
                 promise.setFailure(f.cause());
             }
         });
     }
 
     @Override
     protected void doDeregister() throws Exception {
         NioIoRegistration registration = this.registration;
         if (registration != null) {
             this.registration = null;
             registration.cancel();
         }
     }
 
     @Override
     protected void doBeginRead() throws Exception {
         // Channel.read() or ChannelHandlerContext.read() was called
         NioIoRegistration registration = registration();
         if (registration == null || !registration.isValid()) {
             return;
         }
 
         readPending = true;
 
         addAndSubmit(readOps);
     }
 
     /**
      * Connect to the remote peer
      */
     protected abstract boolean doConnect(SocketAddress remoteAddress, SocketAddress localAddress) throws Exception;
 
     /**
      * Finish the connect
      */
     protected abstract void doFinishConnect() throws Exception;
 
     /**
      * Returns an off-heap copy of the specified {@link ByteBuf}, and releases the original one.
      * Note that this method does not create an off-heap copy if the allocation / deallocation cost is too high,
      * but just returns the original {@link ByteBuf}..
      */
     protected final ByteBuf newDirectBuffer(ByteBuf buf) {
         final int readableBytes = buf.readableBytes();
         if (readableBytes == 0) {
             ReferenceCountUtil.safeRelease(buf);
             return Unpooled.EMPTY_BUFFER;
         }
 
         final ByteBufAllocator alloc = alloc();
         if (alloc.isDirectBufferPooled()) {
             ByteBuf directBuf = alloc.directBuffer(readableBytes);
             directBuf.writeBytes(buf, buf.readerIndex(), readableBytes);
             ReferenceCountUtil.safeRelease(buf);
             return directBuf;
         }
 
         final ByteBuf directBuf = ByteBufUtil.threadLocalDirectBuffer();
         if (directBuf != null) {
             directBuf.writeBytes(buf, buf.readerIndex(), readableBytes);
             ReferenceCountUtil.safeRelease(buf);
             return directBuf;
         }
 
         // Allocating and deallocating an unpooled direct buffer is very expensive; give up.
         return buf;
     }
 
     /**
      * Returns an off-heap copy of the specified {@link ByteBuf}, and releases the specified holder.
      * The caller must ensure that the holder releases the original {@link ByteBuf} when the holder is released by
      * this method.  Note that this method does not create an off-heap copy if the allocation / deallocation cost is
      * too high, but just returns the original {@link ByteBuf}..
      */
     protected final ByteBuf newDirectBuffer(ReferenceCounted holder, ByteBuf buf) {
         final int readableBytes = buf.readableBytes();
         if (readableBytes == 0) {
             ReferenceCountUtil.safeRelease(holder);
             return Unpooled.EMPTY_BUFFER;
         }
 
         final ByteBufAllocator alloc = alloc();
         if (alloc.isDirectBufferPooled()) {
             ByteBuf directBuf = alloc.directBuffer(readableBytes);
             directBuf.writeBytes(buf, buf.readerIndex(), readableBytes);
             ReferenceCountUtil.safeRelease(holder);
             return directBuf;
         }
 
         final ByteBuf directBuf = ByteBufUtil.threadLocalDirectBuffer();
         if (directBuf != null) {
             directBuf.writeBytes(buf, buf.readerIndex(), readableBytes);
             ReferenceCountUtil.safeRelease(holder);
             return directBuf;
         }
 
         // Allocating and deallocating an unpooled direct buffer is very expensive; give up.
         if (holder != buf) {
             // Ensure to call holder.release() to give the holder a chance to release other resources than its content.
             buf.retain();
             ReferenceCountUtil.safeRelease(holder);
         }
 
         return buf;
     }
 
     @Override
     protected void doClose() throws Exception {
         ChannelPromise promise = connectPromise;
         if (promise != null) {
             // Use tryFailure() instead of setFailure() to avoid the race against cancel().
             promise.tryFailure(new ClosedChannelException());
             connectPromise = null;
         }
 
         Future<?> future = connectTimeoutFuture;
         if (future != null) {
             future.cancel(false);
             connectTimeoutFuture = null;
         }
     }
 }