View Javadoc

    1   /***
    2    *
    3    * Copyright 2004 Hiram Chirino
    4    *
    5    * Licensed under the Apache License, Version 2.0 (the "License");
    6    * you may not use this file except in compliance with the License.
    7    * You may obtain a copy of the License at
    8    *
    9    * http://www.apache.org/licenses/LICENSE-2.0
    10   *
    11   * Unless required by applicable law or agreed to in writing, software
    12   * distributed under the License is distributed on an "AS IS" BASIS,
    13   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    14   * See the License for the specific language governing permissions and
    15   * limitations under the License.
    16   *
    17   **/
    18
    19  package org.activeio.net;
    20
    21  import java.io.IOException;
    22  import java.io.InterruptedIOException;
    23  import java.net.SocketException;
    24  import java.nio.ByteBuffer;
    25  import java.nio.channels.SelectionKey;
    26  import java.nio.channels.SocketChannel;
    27
    28  import org.activeio.AsynchChannel;
    29  import org.activeio.AsynchChannelListener;
    30  import org.activeio.Packet;
    31  import org.activeio.Packet.ByteSequence;
    32  import org.activeio.net.NIOAsynchChannelSelectorManager.SelectorManagerListener;
    33  import org.activeio.net.NIOAsynchChannelSelectorManager.SocketChannelAsynchChannelSelection;
    34  import org.activeio.packet.ByteBufferPacket;
    35  import org.activeio.packet.EOSPacket;
    36
    37  /***
    38   * @version $Revision$
    39   */
    40  final public class NIOAsynchChannel extends NIOBaseChannel implements AsynchChannel {
    41
    42      private AsynchChannelListener channelListener;
    43      private SocketChannelAsynchChannelSelection selection;
    44      private ByteBuffer inputByteBuffer;
    45      private boolean running;
    46
    47      public NIOAsynchChannel(SocketChannel socketChannel, boolean useDirect) throws IOException {
    48          super(socketChannel, useDirect);
    49
    50          socketChannel.configureBlocking(false);
    51          selection = NIOAsynchChannelSelectorManager.register(socketChannel, new SelectorManagerListener(){
    52              public void onSelect(SocketChannelAsynchChannelSelection selection) {
    53                  String origName = Thread.currentThread().getName();
    54                  if (selection.isReadable())
    55                  try {
    56                      Thread.currentThread().setName(NIOAsynchChannel.this.toString());
    57                      serviceRead();
    58                   } catch ( Throwable e ) {
    59                       System.err.println("ActiveIO unexpected error: ");
    60                       e.printStackTrace(System.err);
    61                   } finally {
    62                       Thread.currentThread().setName(origName);
    63                   }
    64              }
    65          });
    66
    67      }
    68
    69      private void serviceRead() {
    70          try {
    71
    72              while( true ) {
    73
    74  	            if( inputByteBuffer==null || !inputByteBuffer.hasRemaining() ) {
    75  	                inputByteBuffer = allocateBuffer();
    76  	            }
    77
    78  	            int size = socketChannel.read(inputByteBuffer);
    79  	            if( size == -1 ) {
    80  	                this.channelListener.onPacket( EOSPacket.EOS_PACKET );
    81  	                selection.close();
    82  	                break;
    83  	            }
    84
    85  	            if( size==0 ) {
    86  	                break;
    87  	            }
    88
    89  	            // Per Mike Spile, some plaforms read 1 byte of data on the first read, and then
    90  	            // a but load of data on the second read.  Try to load the butload here
    91  	            if( size == 1 && inputByteBuffer.hasRemaining() ) {
    92  		            int size2 = socketChannel.read(inputByteBuffer);
    93  		            if( size2 > 0 )
    94  		            		size += size2;
    95  	            }
    96
    97  	            ByteBuffer remaining = inputByteBuffer.slice();
    98  	            Packet data = new ByteBufferPacket(((ByteBuffer)inputByteBuffer.flip()).slice());
    99  	            this.channelListener.onPacket( data );
    100
    101 	            // Keep the remaining buffer around to fill with data.
    102 	            inputByteBuffer = remaining;
    103
    104 	            if( inputByteBuffer.hasRemaining() )
    105 	                break;
    106             }
    107
    108         } catch (IOException e) {
    109             this.channelListener.onPacketError(e);
    110         }
    111     }
    112
    113     synchronized public void write(Packet packet) throws IOException {
    114
    115     	ByteBuffer data;
    116         if( packet.getClass()==ByteBufferPacket.class ) {
    117             data = ((ByteBufferPacket)packet).getByteBuffer();
    118         } else {
    119         	ByteSequence sequence = packet.asByteSequence();
    120         	data = ByteBuffer.wrap(sequence.getData(), sequence.getOffset(), sequence.getLength());
    121         }
    122
    123         long delay=1;
    124         while( data.hasRemaining() ) {
    125
    126             // Since the write is non-blocking, all the data may not have been written.
    127             int r1 = data.remaining();
    128 	        socketChannel.write( data );
    129 	        int r2 = data.remaining();
    130
    131 	        // We may need to do a little bit of sleeping to avoid a busy loop.
    132             // Slow down if no data was written out..
    133 	        if( r2>0 && r1-r2==0 ) {
    134 	            try {
    135                     // Use exponential rollback to increase sleep time.
    136                     Thread.sleep(delay);
    137                     delay *= 5;
    138                     if( delay > 1000*1 ) {
    139                         delay = 1000;
    140                     }
    141                 } catch (InterruptedException e) {
    142                     throw new InterruptedIOException();
    143                 }
    144 	        } else {
    145 	            delay = 1;
    146 	        }
    147         }
    148     }
    149
    150     public void flush() throws IOException {
    151     }
    152
    153     public void setAsynchChannelListener(AsynchChannelListener channelListener) {
    154         this.channelListener = channelListener;
    155     }
    156
    157     public AsynchChannelListener getAsynchChannelListener() {
    158         return channelListener;
    159     }
    160
    161     public void dispose() {
    162         if( running && channelListener!=null ) {
    163             channelListener.onPacketError(new SocketException("Socket closed."));
    164         }
    165         selection.close();
    166         super.dispose();
    167     }
    168
    169     public void start() throws IOException {
    170         if( running )
    171             return;
    172         running=true;
    173         selection.setInterestOps(SelectionKey.OP_READ);
    174     }
    175
    176     public void stop(long timeout) throws IOException {
    177         if( !running )
    178             return;
    179         running=false;
    180         selection.setInterestOps(0);
    181     }
    182  }