/*
    *   @(#) $Id: DatagramAcceptorDelegate.java 355016 2005-12-08 07:00:30Z trustin $
    *
    *   Copyright 2004 The Apache Software Foundation
    *
    *   Licensed 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.apache.mina.transport.socket.nio.support;
  
  import java.io.IOException;
  import java.net.InetSocketAddress;
  import java.net.SocketAddress;
  import java.nio.channels.DatagramChannel;
  import java.nio.channels.SelectionKey;
  import java.nio.channels.Selector;
  import java.util.HashMap;
  import java.util.Iterator;
  import java.util.Map;
  import java.util.Set;
  
  import org.apache.mina.common.ByteBuffer;
  import org.apache.mina.common.ExceptionMonitor;
  import org.apache.mina.common.IoAcceptor;
  import org.apache.mina.common.IoFilterChainBuilder;
  import org.apache.mina.common.IoHandler;
  import org.apache.mina.common.IoSession;
  import org.apache.mina.common.IoFilter.WriteRequest;
  import org.apache.mina.common.support.BaseIoAcceptor;
  import org.apache.mina.util.ExceptionUtil;
  import org.apache.mina.util.Queue;
  
  /***
   * {@link IoAcceptor} for datagram transport (UDP/IP).
   * 
   * @author The Apache Directory Project (dev@directory.apache.org)
   * @version $Rev: 355016 $, $Date: 2005-12-08 16:00:30 +0900 (Thu, 08 Dec 2005) $
   */
  public class DatagramAcceptorDelegate extends BaseIoAcceptor implements IoAcceptor, DatagramSessionManager
  {
      private static volatile int nextId = 0;
  
      private final IoAcceptor wrapper;
      private final int id = nextId ++ ;
      private Selector selector;
      private final Map channels = new HashMap();
      private final Queue registerQueue = new Queue();
      private final Queue cancelQueue = new Queue();
      private final Queue flushingSessions = new Queue();
      private Worker worker;
  
      /***
       * Creates a new instance.
       */
      public DatagramAcceptorDelegate( IoAcceptor wrapper )
      {
          this.wrapper = wrapper;
      }
  
      public void bind( SocketAddress address, IoHandler handler, IoFilterChainBuilder filterChainBuilder )
              throws IOException
      {
          if( address == null )
              throw new NullPointerException( "address" );
          if( handler == null )
              throw new NullPointerException( "handler" );
  
          if( !( address instanceof InetSocketAddress ) )
              throw new IllegalArgumentException( "Unexpected address type: "
                                                  + address.getClass() );
          if( ( ( InetSocketAddress ) address ).getPort() == 0 )
              throw new IllegalArgumentException( "Unsupported port number: 0" );
          
          if( filterChainBuilder == null )
          {
              filterChainBuilder = IoFilterChainBuilder.NOOP;
          }
  
          RegistrationRequest request = new RegistrationRequest( address, handler, filterChainBuilder );
          synchronized( this )
          {
              synchronized( registerQueue )
              {
                  registerQueue.push( request );
              }
              startupWorker();
          }
          selector.wakeup();
          
         synchronized( request )
         {
             while( !request.done )
             {
                 try
                 {
                     request.wait();
                 }
                 catch( InterruptedException e )
                 {
                 }
             }
         }
         
         if( request.exception != null )
         {
             ExceptionUtil.throwException( request.exception );
         }
     }
 
     public void unbind( SocketAddress address )
     {
         // TODO: DIRMINA-93
         if( address == null )
             throw new NullPointerException( "address" );
 
         CancellationRequest request = new CancellationRequest( address );
         synchronized( this )
         {
             try
             {
                 startupWorker();
             }
             catch( IOException e )
             {
                 // IOException is thrown only when Worker thread is not
                 // running and failed to open a selector.  We simply throw
                 // IllegalArgumentException here because we can simply
                 // conclude that nothing is bound to the selector.
                 throw new IllegalArgumentException( "Address not bound: " + address );
             }
 
             synchronized( cancelQueue )
             {
                 cancelQueue.push( request );
             }
         }
         selector.wakeup();
         
         synchronized( request )
         {
             while( !request.done )
             {
                 try
                 {
                     request.wait();
                 }
                 catch( InterruptedException e )
                 {
                 }
             }
         }
         
         if( request.exception != null )
         {
             request.exception.fillInStackTrace();
             throw request.exception;
         }
     }
     
     public IoSession newSession( SocketAddress remoteAddress, SocketAddress localAddress )
     {
         if( remoteAddress == null )
         {
             throw new NullPointerException( "remoteAddress" );
         }
         if( localAddress == null )
         {
             throw new NullPointerException( "localAddress" );
         }
         
         Selector selector = this.selector;
         DatagramChannel ch = ( DatagramChannel ) channels.get( localAddress );
         if( selector == null || ch == null )
         {
             throw new IllegalArgumentException( "Unknown localAddress: " + localAddress );
         }
             
         SelectionKey key = ch.keyFor( selector );
         if( key == null )
         {
             throw new IllegalArgumentException( "Unknown localAddress: " + localAddress );
         }
 
         RegistrationRequest req = ( RegistrationRequest ) key.attachment();
         DatagramSessionImpl s = new DatagramSessionImpl( wrapper, this, ch, req.handler );
         s.setRemoteAddress( remoteAddress );
         s.setSelectionKey( key );
         
         try
         {
             this.filterChainBuilder.buildFilterChain( s.getFilterChain() );
             req.filterChainBuilder.buildFilterChain( s.getFilterChain() );
             ( ( DatagramFilterChain ) s.getFilterChain() ).sessionCreated( s );
         }
         catch( Throwable t )
         {
             ExceptionMonitor.getInstance().exceptionCaught( t );
         }
         
         return s;
     }
 
     private synchronized void startupWorker() throws IOException
     {
         if( worker == null )
         {
             selector = Selector.open();
             worker = new Worker();
             worker.start();
         }
     }
 
     public void flushSession( DatagramSessionImpl session )
     {
         scheduleFlush( session );
         Selector selector = this.selector;
         if( selector != null )
         {
             selector.wakeup();
         }
     }
 
     public void closeSession( DatagramSessionImpl session )
     {
     }
 
     private void scheduleFlush( DatagramSessionImpl session )
     {
         synchronized( flushingSessions )
         {
             flushingSessions.push( session );
         }
     }
 
     private class Worker extends Thread
     {
         public Worker()
         {
             super( "DatagramAcceptor-" + id );
         }
 
         public void run()
         {
             for( ;; )
             {
                 try
                 {
                     int nKeys = selector.select();
 
                     registerNew();
 
                     if( nKeys > 0 )
                     {
                         processReadySessions( selector.selectedKeys() );
                     }
 
                     flushSessions();
                     cancelKeys();
 
                     if( selector.keys().isEmpty() )
                     {
                         synchronized( DatagramAcceptorDelegate.this )
                         {
                             if( selector.keys().isEmpty() &&
                                 registerQueue.isEmpty() &&
                                 cancelQueue.isEmpty() )
                             {
                                 worker = null;
                                 try
                                 {
                                     selector.close();
                                 }
                                 catch( IOException e )
                                 {
                                     ExceptionMonitor.getInstance().exceptionCaught( e );
                                 }
                                 finally
                                 {
                                     selector = null;
                                 }
                                 break;
                             }
                         }
                     }
                 }
                 catch( IOException e )
                 {
                     ExceptionMonitor.getInstance().exceptionCaught( e );
 
                     try
                     {
                         Thread.sleep( 1000 );
                     }
                     catch( InterruptedException e1 )
                     {
                     }
                 }
             }
         }
     }
 
     private void processReadySessions( Set keys )
     {
         Iterator it = keys.iterator();
         while( it.hasNext() )
         {
             SelectionKey key = ( SelectionKey ) it.next();
             it.remove();
 
             DatagramChannel ch = ( DatagramChannel ) key.channel();
 
             RegistrationRequest req = ( RegistrationRequest ) key.attachment();
             DatagramSessionImpl session =
                 new DatagramSessionImpl( wrapper, this, ch, req.handler );
             session.setSelectionKey( key );
             
             try
             {
                 ( ( DatagramFilterChain ) session.getFilterChain() ).sessionCreated( session );
 
                 if( key.isReadable() )
                 {
                     readSession( session );
                 }
 
                 if( key.isWritable() )
                 {
                     scheduleFlush( session );
                 }
             }
             catch( Throwable t )
             {
                 ExceptionMonitor.getInstance().exceptionCaught( t );
             }
         }
     }
 
     private void readSession( DatagramSessionImpl session )
     {
 
         ByteBuffer readBuf = ByteBuffer.allocate( 2048 );
         try
         {
             SocketAddress remoteAddress = session.getChannel().receive(
                     readBuf.buf() );
             if( remoteAddress != null )
             {
                 readBuf.flip();
                 session.setRemoteAddress( remoteAddress );
 
                 ByteBuffer newBuf = ByteBuffer.allocate( readBuf.limit() );
                 newBuf.put( readBuf );
                 newBuf.flip();
 
                 session.increaseReadBytes( newBuf.remaining() );
                 ( ( DatagramFilterChain ) session.getFilterChain() ).messageReceived( session, newBuf );
             }
         }
         catch( IOException e )
         {
             ( ( DatagramFilterChain ) session.getFilterChain() ).exceptionCaught( session, e );
         }
         finally
         {
             readBuf.release();
         }
     }
 
     private void flushSessions()
     {
         if( flushingSessions.size() == 0 )
             return;
 
         for( ;; )
         {
             DatagramSessionImpl session;
 
             synchronized( flushingSessions )
             {
                 session = ( DatagramSessionImpl ) flushingSessions.pop();
             }
 
             if( session == null )
                 break;
 
             try
             {
                 flush( session );
             }
             catch( IOException e )
             {
                 ( ( DatagramFilterChain ) session.getFilterChain() ).exceptionCaught( session, e );
             }
         }
     }
 
     private void flush( DatagramSessionImpl session ) throws IOException
     {
         DatagramChannel ch = session.getChannel();
 
         Queue writeRequestQueue = session.getWriteRequestQueue();
 
         WriteRequest req;
         for( ;; )
         {
             synchronized( writeRequestQueue )
             {
                 req = ( WriteRequest ) writeRequestQueue.first();
             }
 
             if( req == null )
                 break;
 
             ByteBuffer buf = ( ByteBuffer ) req.getMessage();
             if( buf.remaining() == 0 )
             {
                 // pop and fire event
                 synchronized( writeRequestQueue )
                 {
                     writeRequestQueue.pop();
                 }
 
                 req.getFuture().setWritten( true );
                 session.increaseWrittenWriteRequests();
                 ( ( DatagramFilterChain ) session.getFilterChain() ).messageSent( session, buf );
                 continue;
             }
 
             SelectionKey key = session.getSelectionKey();
             if( key == null )
             {
                 scheduleFlush( session );
                 break;
             }
             if( !key.isValid() )
             {
                 continue;
             }
 
             int pos = buf.position();
             int writtenBytes = ch
                     .send( buf.buf(), session.getRemoteAddress() );
 
             if( writtenBytes == 0 )
             {
                 // Kernel buffer is full
                 key.interestOps( key.interestOps() | SelectionKey.OP_WRITE );
             }
             else if( writtenBytes > 0 )
             {
                 key.interestOps( key.interestOps()
                                  & ( ~SelectionKey.OP_WRITE ) );
 
                 // pop and fire event
                 synchronized( writeRequestQueue )
                 {
                     writeRequestQueue.pop();
                 }
 
                 session.increaseWrittenBytes( writtenBytes );
                 req.getFuture().setWritten( true );
                 session.increaseWrittenWriteRequests();
                 ( ( DatagramFilterChain ) session.getFilterChain() ).messageSent( session, buf.position( pos ) );
             }
         }
     }
 
     private void registerNew()
     {
         if( registerQueue.isEmpty() )
             return;
 
         for( ;; )
         {
             RegistrationRequest req;
             synchronized( registerQueue )
             {
                 req = ( RegistrationRequest ) registerQueue.pop();
             }
 
             if( req == null )
                 break;
 
             DatagramChannel ch = null;
             try
             {
                 ch = DatagramChannel.open();
                 ch.configureBlocking( false );
                 ch.socket().bind( req.address );
                 ch.register( selector, SelectionKey.OP_READ, req );
                 channels.put( req.address, ch );
             }
             catch( Throwable t )
             {
                 req.exception = t;
             }
             finally
             {
                 synchronized( req )
                 {
                     req.done = true;
                     req.notify();
                 }
 
                 if( ch != null && req.exception != null )
                 {
                     try
                     {
                         ch.close();
                     }
                     catch( Throwable e )
                     {
                         ExceptionMonitor.getInstance().exceptionCaught( e );
                     }
                 }
             }
         }
     }
 
     private void cancelKeys()
     {
         if( cancelQueue.isEmpty() )
             return;
 
         for( ;; )
         {
             CancellationRequest request;
             synchronized( cancelQueue )
             {
                 request = ( CancellationRequest ) cancelQueue.pop();
             }
             
             if( request == null )
             {
                 break;
             }
 
             DatagramChannel ch = ( DatagramChannel ) channels.remove( request.address );
             // close the channel
             try
             {
                 if( ch == null )
                 {
                     request.exception = new IllegalArgumentException(
                             "Address not bound: " + request.address );
                 }
                 else
                 {
                     SelectionKey key = ch.keyFor( selector );
                     key.cancel();
                     selector.wakeup(); // wake up again to trigger thread death
                     ch.close();
                 }
             }
             catch( Throwable t )
             {
                 ExceptionMonitor.getInstance().exceptionCaught( t );
             }
             finally
             {
                 synchronized( request )
                 {
                     request.done = true;
                     request.notify();
                 }
             }
         }
     }
     
     public void updateTrafficMask( DatagramSessionImpl session )
     {
         // There's no point in changing the traffic mask for sessions originating
         // from this acceptor since new sessions are created every time data is
         // received.
     }
 
     private static class RegistrationRequest
     {
         private final SocketAddress address;
         private final IoHandler handler;
         private final IoFilterChainBuilder filterChainBuilder;
 
         private Throwable exception; 
         private boolean done;
         
         private RegistrationRequest( SocketAddress address, IoHandler handler, IoFilterChainBuilder filterChainBuilder )
         {
             this.address = address;
             this.handler = handler;
             this.filterChainBuilder = filterChainBuilder;
         }
     }
 
     private static class CancellationRequest
     {
         private final SocketAddress address;
         private boolean done;
         private RuntimeException exception;
         
         private CancellationRequest( SocketAddress address )
         {
             this.address = address;
         }
     }
 }