/*
    * 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.channel.nio;
  
  import io.netty5.channel.Channel;
  import io.netty5.channel.ChannelMetadata;
  import io.netty5.channel.ChannelOutboundBuffer;
  import io.netty5.channel.ChannelPipeline;
  import io.netty5.channel.ChannelShutdownDirection;
  import io.netty5.channel.EventLoop;
  import io.netty5.channel.RecvBufferAllocator;
  import io.netty5.channel.ServerChannel;
  
  import java.io.IOException;
  import java.net.PortUnreachableException;
  import java.net.SocketAddress;
  import java.nio.channels.SelectableChannel;
  import java.nio.channels.SelectionKey;
  import java.util.ArrayList;
  import java.util.List;
  
  /**
   * {@link AbstractNioChannel} base class for {@link Channel}s that operate on messages.
   */
  public abstract class AbstractNioMessageChannel<P extends Channel, L extends SocketAddress, R extends SocketAddress>
          extends AbstractNioChannel<P, L, R> {
      boolean inputShutdown;
      private final List<Object> readBuf = new ArrayList<>();
  
      /**
       * @see AbstractNioChannel#AbstractNioChannel(Channel, EventLoop,
       * ChannelMetadata, RecvBufferAllocator, SelectableChannel, int)
       */
      protected AbstractNioMessageChannel(P parent, EventLoop eventLoop, ChannelMetadata metadata,
                                          RecvBufferAllocator defaultRecvAllocator,
                                          SelectableChannel ch, int readInterestOp) {
          super(parent, eventLoop, metadata, defaultRecvAllocator, ch, readInterestOp);
      }
  
      @Override
      protected void doBeginRead() throws Exception {
          if (inputShutdown) {
              return;
          }
          super.doBeginRead();
      }
  
      protected boolean continueReading(RecvBufferAllocator.Handle allocHandle) {
          return allocHandle.continueReading(isAutoRead());
      }
  
      @Override
      protected final void readNow() {
          assert executor().inEventLoop();
          final ChannelPipeline pipeline = pipeline();
          final RecvBufferAllocator.Handle allocHandle = recvBufAllocHandle();
          allocHandle.reset();
  
          boolean closed = false;
          Throwable exception = null;
          try {
              try {
                  do {
                      int localRead = doReadMessages(readBuf);
                      if (localRead == 0) {
                          break;
                      }
                      if (localRead < 0) {
                          closed = true;
                          break;
                      }
  
                      allocHandle.incMessagesRead(localRead);
                  } while (continueReading(allocHandle) && !isShutdown(ChannelShutdownDirection.Inbound));
              } catch (Throwable t) {
                  exception = t;
              }
  
              int size = readBuf.size();
              for (int i = 0; i < size; i ++) {
                  readPending = false;
                  pipeline.fireChannelRead(readBuf.get(i));
              }
              readBuf.clear();
              allocHandle.readComplete();
              pipeline.fireChannelReadComplete();
 
             if (exception != null) {
                 closed = closeOnReadError(exception);
 
                 pipeline.fireChannelExceptionCaught(exception);
             }
 
             if (closed) {
                 inputShutdown = true;
                 if (isOpen()) {
                     closeTransport(newPromise());
                 }
             } else {
                 readIfIsAutoRead();
             }
         } finally {
             // Check if there is a readPending which was not processed yet.
             // This could be for two reasons:
             // * The user called Channel.read() or ChannelHandlerContext.read() in channelRead(...) method
             // * The user called Channel.read() or ChannelHandlerContext.read() in channelReadComplete(...) method
             //
             // See https://github.com/netty/netty/issues/2254
             if (!readPending && !isAutoRead()) {
                 removeReadOp();
             }
         }
     }
 
     @Override
     protected void doWrite(ChannelOutboundBuffer in) throws Exception {
         final SelectionKey key = selectionKey();
         if (key == null) {
             return;
         }
         final int interestOps = key.interestOps();
 
         int maxMessagesPerWrite = getMaxMessagesPerWrite();
         while (maxMessagesPerWrite > 0) {
             Object msg = in.current();
             if (msg == null) {
                 break;
             }
             try {
                 boolean done = false;
                 for (int i = getWriteSpinCount() - 1; i >= 0; i--) {
                     if (doWriteMessage(msg, in)) {
                         done = true;
                         break;
                     }
                 }
 
                 if (done) {
                     maxMessagesPerWrite--;
                     in.remove();
                 } else {
                     break;
                 }
             } catch (Exception e) {
                 if (continueOnWriteError()) {
                     maxMessagesPerWrite--;
                     in.remove(e);
                 } else {
                     throw e;
                 }
             }
         }
         if (in.isEmpty()) {
             // Wrote all messages.
             if ((interestOps & SelectionKey.OP_WRITE) != 0) {
                 key.interestOps(interestOps & ~SelectionKey.OP_WRITE);
             }
         } else {
             // Did not write all messages.
             if ((interestOps & SelectionKey.OP_WRITE) == 0) {
                 key.interestOps(interestOps | SelectionKey.OP_WRITE);
             }
         }
     }
 
     /**
      * Returns {@code true} if we should continue the write loop on a write error.
      */
     protected boolean continueOnWriteError() {
         return false;
     }
 
     protected boolean closeOnReadError(Throwable cause) {
         if (!isActive()) {
             // If the channel is not active anymore for whatever reason we should not try to continue reading.
             return true;
         }
         if (cause instanceof PortUnreachableException) {
             return false;
         }
         if (cause instanceof IOException) {
             // ServerChannel should not be closed even on IOException because it can often continue
             // accepting incoming connections. (e.g. too many open files)
             return !(this instanceof ServerChannel);
         }
         return true;
     }
 
     /**
      * Read messages into the given array and return the amount which was read.
      */
     protected abstract int doReadMessages(List<Object> buf) throws Exception;
 
     /**
      * Write a message to the underlying {@link java.nio.channels.Channel}.
      *
      * @return {@code true} if and only if the message has been written
      */
     protected abstract boolean doWriteMessage(Object msg, ChannelOutboundBuffer in) throws Exception;
 }