/*
    * 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 org.jboss.netty.channel.socket.nio;
  
  import static org.jboss.netty.channel.Channels.*;
  
  import java.io.IOException;
  import java.net.SocketAddress;
  import java.net.SocketTimeoutException;
  import java.nio.channels.CancelledKeyException;
  import java.nio.channels.ClosedChannelException;
  import java.nio.channels.ClosedSelectorException;
  import java.nio.channels.SelectionKey;
  import java.nio.channels.Selector;
  import java.nio.channels.SocketChannel;
  import java.util.concurrent.Executor;
  import java.util.concurrent.atomic.AtomicInteger;
  
  import org.jboss.netty.channel.Channel;
  import org.jboss.netty.channel.ChannelEvent;
  import org.jboss.netty.channel.ChannelFuture;
  import org.jboss.netty.channel.ChannelPipeline;
  import org.jboss.netty.channel.ChannelState;
  import org.jboss.netty.channel.ChannelStateEvent;
  import org.jboss.netty.channel.MessageEvent;
  import org.jboss.netty.logging.InternalLogger;
  import org.jboss.netty.logging.InternalLoggerFactory;
  import org.jboss.netty.util.ThreadRenamingRunnable;
  import org.jboss.netty.util.internal.DeadLockProofWorker;
  
  class NioServerSocketPipelineSink extends AbstractNioChannelSink {
  
      private static final AtomicInteger nextId = new AtomicInteger();
  
      static final InternalLogger logger =
          InternalLoggerFactory.getInstance(NioServerSocketPipelineSink.class);
  
      final int id = nextId.incrementAndGet();
  
      private final WorkerPool<NioWorker> workerPool;
  
      NioServerSocketPipelineSink(WorkerPool<NioWorker> workerPool) {
          this.workerPool = workerPool;
      }
  
  
      public void eventSunk(
              ChannelPipeline pipeline, ChannelEvent e) throws Exception {
          Channel channel = e.getChannel();
          if (channel instanceof NioServerSocketChannel) {
              handleServerSocket(e);
          } else if (channel instanceof NioSocketChannel) {
              handleAcceptedSocket(e);
          }
      }
  
      private void handleServerSocket(ChannelEvent e) {
          if (!(e instanceof ChannelStateEvent)) {
              return;
          }
  
          ChannelStateEvent event = (ChannelStateEvent) e;
          NioServerSocketChannel channel =
              (NioServerSocketChannel) event.getChannel();
          ChannelFuture future = event.getFuture();
          ChannelState state = event.getState();
          Object value = event.getValue();
  
          switch (state) {
          case OPEN:
              if (Boolean.FALSE.equals(value)) {
                  close(channel, future);
              }
              break;
          case BOUND:
              if (value != null) {
                  bind(channel, future, (SocketAddress) value);
              } else {
                  close(channel, future);
              }
              break;
          default:
              break;
          }
      }
  
     private static void handleAcceptedSocket(ChannelEvent e) {
         if (e instanceof ChannelStateEvent) {
             ChannelStateEvent event = (ChannelStateEvent) e;
             NioSocketChannel channel = (NioSocketChannel) event.getChannel();
             ChannelFuture future = event.getFuture();
             ChannelState state = event.getState();
             Object value = event.getValue();
 
             switch (state) {
             case OPEN:
                 if (Boolean.FALSE.equals(value)) {
                     channel.worker.close(channel, future);
                 }
                 break;
             case BOUND:
             case CONNECTED:
                 if (value == null) {
                     channel.worker.close(channel, future);
                 }
                 break;
             case INTEREST_OPS:
                 channel.worker.setInterestOps(channel, future, ((Integer) value).intValue());
                 break;
             }
         } else if (e instanceof MessageEvent) {
             MessageEvent event = (MessageEvent) e;
             NioSocketChannel channel = (NioSocketChannel) event.getChannel();
             boolean offered = channel.writeBufferQueue.offer(event);
             assert offered;
             channel.worker.writeFromUserCode(channel);
         }
     }
 
     private void bind(
             NioServerSocketChannel channel, ChannelFuture future,
             SocketAddress localAddress) {
 
         boolean bound = false;
         boolean bossStarted = false;
         try {
             channel.socket.socket().bind(localAddress, channel.getConfig().getBacklog());
             bound = true;
 
             future.setSuccess();
             fireChannelBound(channel, channel.getLocalAddress());
 
             Executor bossExecutor =
                 ((NioServerSocketChannelFactory) channel.getFactory()).bossExecutor;
             DeadLockProofWorker.start(bossExecutor,
                     new ThreadRenamingRunnable(new Boss(channel),
                             "New I/O server boss #" + id + " (" + channel + ')'));
             bossStarted = true;
         } catch (Throwable t) {
             future.setFailure(t);
             fireExceptionCaught(channel, t);
         } finally {
             if (!bossStarted && bound) {
                 close(channel, future);
             }
         }
     }
 
     private static void close(NioServerSocketChannel channel, ChannelFuture future) {
         boolean bound = channel.isBound();
         try {
             if (channel.socket.isOpen()) {
                 channel.socket.close();
                 Selector selector = channel.selector;
                 if (selector != null) {
                     selector.wakeup();
                 }
             }
 
             // Make sure the boss thread is not running so that that the future
             // is notified after a new connection cannot be accepted anymore.
             // See NETTY-256 for more information.
             channel.shutdownLock.lock();
             try {
                 if (channel.setClosed()) {
                     future.setSuccess();
                     if (bound) {
                         fireChannelUnbound(channel);
                     }
                     fireChannelClosed(channel);
                 } else {
                     future.setSuccess();
                 }
             } finally {
                 channel.shutdownLock.unlock();
             }
         } catch (Throwable t) {
             future.setFailure(t);
             fireExceptionCaught(channel, t);
         }
     }
 
     NioWorker nextWorker() {
         return workerPool.nextWorker();
     }
 
     private final class Boss implements Runnable {
         private final Selector selector;
         private final NioServerSocketChannel channel;
 
         Boss(NioServerSocketChannel channel) throws IOException {
             this.channel = channel;
 
             selector = Selector.open();
 
             boolean registered = false;
             try {
                 channel.socket.register(selector, SelectionKey.OP_ACCEPT);
                 registered = true;
             } finally {
                 if (!registered) {
                     closeSelector();
                 }
             }
 
             channel.selector = selector;
         }
 
         public void run() {
             final Thread currentThread = Thread.currentThread();
 
             channel.shutdownLock.lock();
             try {
                 for (;;) {
                     try {
                         // Just do a blocking select without any timeout
                         // as this thread does not execute anything else.
                         selector.select();
                         // There was something selected if we reach this point, so clear
                         // the selected keys
                         selector.selectedKeys().clear();
 
                         // accept connections in a for loop until no new connection is ready
                         for (;;) {
                             SocketChannel acceptedSocket = channel.socket.accept();
                             if (acceptedSocket == null) {
                                 break;
                             }
                             registerAcceptedChannel(acceptedSocket, currentThread);
 
                         }
 
                     } catch (SocketTimeoutException e) {
                         // Thrown every second to get ClosedChannelException
                         // raised.
                     } catch (CancelledKeyException e) {
                         // Raised by accept() when the server socket was closed.
                     } catch (ClosedSelectorException e) {
                         // Raised by accept() when the server socket was closed.
                     } catch (ClosedChannelException e) {
                         // Closed as requested.
                         break;
                     } catch (Throwable e) {
                         if (logger.isWarnEnabled()) {
                             logger.warn(
                                     "Failed to accept a connection.", e);
                         }
 
                         try {
                             Thread.sleep(1000);
                         } catch (InterruptedException e1) {
                             // Ignore
                         }
                     }
                 }
             } finally {
                 channel.shutdownLock.unlock();
                 closeSelector();
             }
         }
 
         private void registerAcceptedChannel(SocketChannel acceptedSocket, Thread currentThread) {
             try {
                 ChannelPipeline pipeline =
                     channel.getConfig().getPipelineFactory().getPipeline();
                 NioWorker worker = nextWorker();
                 worker.register(new NioAcceptedSocketChannel(
                         channel.getFactory(), pipeline, channel,
                         NioServerSocketPipelineSink.this, acceptedSocket,
                         worker, currentThread), null);
             } catch (Exception e) {
                 if (logger.isWarnEnabled()) {
                     logger.warn(
                             "Failed to initialize an accepted socket.", e);
                 }
 
                 try {
                     acceptedSocket.close();
                 } catch (IOException e2) {
                     if (logger.isWarnEnabled()) {
                         logger.warn(
                                 "Failed to close a partially accepted socket.",
                                 e2);
                     }
 
                 }
             }
         }
 
         private void closeSelector() {
             channel.selector = null;
             try {
                 selector.close();
             } catch (Exception e) {
                 if (logger.isWarnEnabled()) {
                     logger.warn("Failed to close a selector.", e);
                 }
             }
         }
     }
 }