/*
    * 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:
    *
    *   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 io.netty.channel.nio;
  
  import io.netty.channel.Channel;
  import io.netty.channel.ChannelConfig;
  import io.netty.channel.ChannelOutboundBuffer;
  import io.netty.channel.ChannelPipeline;
  import io.netty.channel.ServerChannel;
  
  import java.io.IOException;
  import java.net.PortUnreachableException;
  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 extends AbstractNioChannel {
  
      /**
       * @see AbstractNioChannel#AbstractNioChannel(Channel, SelectableChannel, int)
       */
      protected AbstractNioMessageChannel(Channel parent, SelectableChannel ch, int readInterestOp) {
          super(parent, ch, readInterestOp);
      }
  
      @Override
      protected AbstractNioUnsafe newUnsafe() {
          return new NioMessageUnsafe();
      }
  
      private final class NioMessageUnsafe extends AbstractNioUnsafe {
  
          private final List<Object> readBuf = new ArrayList<Object>();
  
          @Override
          public void read() {
              assert eventLoop().inEventLoop();
              final ChannelConfig config = config();
              if (!config.isAutoRead() && !isReadPending()) {
                  // ChannelConfig.setAutoRead(false) was called in the meantime
                  removeReadOp();
                  return;
              }
  
              final int maxMessagesPerRead = config.getMaxMessagesPerRead();
              final ChannelPipeline pipeline = pipeline();
              boolean closed = false;
              Throwable exception = null;
              try {
                  try {
                      for (;;) {
                          int localRead = doReadMessages(readBuf);
                          if (localRead == 0) {
                              break;
                          }
                          if (localRead < 0) {
                              closed = true;
                              break;
                          }
  
                          // stop reading and remove op
                          if (!config.isAutoRead()) {
                              break;
                          }
  
                          if (readBuf.size() >= maxMessagesPerRead) {
                              break;
                          }
                      }
                  } catch (Throwable t) {
                      exception = t;
                  }
                  setReadPending(false);
                  int size = readBuf.size();
                  for (int i = 0; i < size; i ++) {
                      pipeline.fireChannelRead(readBuf.get(i));
                  }
  
                  readBuf.clear();
                  pipeline.fireChannelReadComplete();
  
                  if (exception != null) {
                     closed = closeOnReadError(exception);
 
                     pipeline.fireExceptionCaught(exception);
                 }
 
                 if (closed) {
                     if (isOpen()) {
                         close(voidPromise());
                     }
                 }
             } 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 (!config.isAutoRead() && !isReadPending()) {
                     removeReadOp();
                 }
             }
         }
     }
 
     @Override
     protected void doWrite(ChannelOutboundBuffer in) throws Exception {
         final SelectionKey key = selectionKey();
         final int interestOps = key.interestOps();
 
         for (;;) {
             Object msg = in.current();
             if (msg == null) {
                 // Wrote all messages.
                 if ((interestOps & SelectionKey.OP_WRITE) != 0) {
                     key.interestOps(interestOps & ~SelectionKey.OP_WRITE);
                 }
                 break;
             }
             try {
                 boolean done = false;
                 for (int i = config().getWriteSpinCount() - 1; i >= 0; i--) {
                     if (doWriteMessage(msg, in)) {
                         done = true;
                         break;
                     }
                 }
 
                 if (done) {
                     in.remove();
                 } else {
                     // Did not write all messages.
                     if ((interestOps & SelectionKey.OP_WRITE) == 0) {
                         key.interestOps(interestOps | SelectionKey.OP_WRITE);
                     }
                     break;
                 }
             } catch (Exception e) {
                 if (continueOnWriteError()) {
                     in.remove(e);
                 } else {
                     throw e;
                 }
             }
         }
     }
 
     /**
      * Returns {@code true} if we should continue the write loop on a write error.
      */
     protected boolean continueOnWriteError() {
         return false;
     }
 
     protected boolean closeOnReadError(Throwable cause) {
         // ServerChannel should not be closed even on IOException because it can often continue
         // accepting incoming connections. (e.g. too many open files)
         return cause instanceof IOException &&
                 !(cause instanceof PortUnreachableException) &&
                 !(this instanceof ServerChannel);
     }
 
     /**
      * 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;
 }