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.InetAddress;
    24  import java.net.Socket;
    25  import java.net.SocketAddress;
    26  import java.net.SocketException;
    27  import java.nio.ByteBuffer;
    28
    29  import org.activeio.AsynchChannel;
    30  import org.activeio.AsynchChannelListener;
    31  import org.activeio.Packet;
    32  import org.activeio.packet.ByteBufferPacket;
    33  import org.activeio.packet.EOSPacket;
    34
    35  import EDU.oswego.cs.dl.util.concurrent.Latch;
    36  import EDU.oswego.cs.dl.util.concurrent.SynchronizedBoolean;
    37
    38  import com.ibm.io.async.AsyncSocketChannel;
    39  import com.ibm.io.async.IAbstractAsyncFuture;
    40  import com.ibm.io.async.IAsyncFuture;
    41  import com.ibm.io.async.ICompletionListener;
    42
    43  /***
    44   * @version $Revision$
    45   */
    46  final public class AIOAsynchChannel implements AsynchChannel, ICompletionListener, SocketMetadata {
    47
    48      protected static final int DEFAULT_BUFFER_SIZE = ByteBufferPacket.DEFAULT_DIRECT_BUFFER_SIZE;
    49
    50      private final AsyncSocketChannel socketChannel;
    51      private final Socket socket;
    52
    53      private AsynchChannelListener channelListener;
    54      private ByteBuffer inputByteBuffer;
    55
    56      private final SynchronizedBoolean running = new SynchronizedBoolean(false);
    57      private Latch doneLatch;
    58
    59      protected AIOAsynchChannel(AsyncSocketChannel socketChannel) throws IOException {
    60          this.socketChannel = socketChannel;
    61          this.socket = socketChannel.socket();
    62          this.socket.setSendBufferSize(DEFAULT_BUFFER_SIZE);
    63          this.socket.setReceiveBufferSize(DEFAULT_BUFFER_SIZE);
    64          this.socket.setSoTimeout(0);
    65      }
    66
    67      private ByteBuffer allocateBuffer() {
    68          return ByteBuffer.allocateDirect(DEFAULT_BUFFER_SIZE);
    69      }
    70
    71      public void setAsynchChannelListener(AsynchChannelListener channelListener) {
    72          this.channelListener = channelListener;
    73      }
    74
    75      public AsynchChannelListener getAsynchChannelListener() {
    76          return channelListener;
    77      }
    78
    79      public Object narrow(Class target) {
    80          if( target.isAssignableFrom(getClass()) ) {
    81              return this;
    82          }
    83          return null;
    84      }
    85
    86      public void dispose() {
    87          if( running.get() && channelListener!=null ) {
    88              channelListener.onPacketError(new SocketException("Socket closed."));
    89          }
    90          try {
    91              stop(NO_WAIT_TIMEOUT);
    92          } catch (IOException e) {
    93          }
    94          try {
    95              socketChannel.close();
    96          } catch (IOException e) {
    97          }
    98      }
    99
    100     public void start() throws IOException {
    101         if( running.commit(false, true) ) {
    102             doneLatch = new Latch();
    103             requestNextRead();
    104         }
    105     }
    106
    107     public void stop(long timeout) throws IOException {
    108         if( running.commit(true, false) ) {
    109             try {
    110                 if( timeout == NO_WAIT_TIMEOUT ) {
    111                     doneLatch.attempt(0);
    112                 } else if( timeout == WAIT_FOREVER_TIMEOUT ) {
    113                     doneLatch.acquire();
    114                 } else {
    115                     doneLatch.attempt(timeout);
    116                 }
    117             } catch (InterruptedException e) {
    118                 throw new InterruptedIOException();
    119             }
    120         }
    121     }
    122
    123     public void write(Packet packet) throws IOException {
    124         ByteBuffer data = ((ByteBufferPacket)packet).getByteBuffer();
    125         while( data.hasRemaining() ) {
    126 	        IAsyncFuture future = socketChannel.write(data);
    127 	        try {
    128 	            future.getByteCount();
    129 	        } catch (InterruptedException e) {
    130 	            throw new InterruptedIOException();
    131 	        }
    132         }
    133     }
    134
    135     public void flush() throws IOException {
    136     }
    137
    138     public void futureCompleted(IAbstractAsyncFuture abstractFuture, Object attribute) {
    139         IAsyncFuture future = (IAsyncFuture)abstractFuture;
    140         try {
    141
    142             if( inputByteBuffer.position()>0 ) {
    143 	            ByteBuffer remaining = inputByteBuffer.slice();
    144 	            Packet data = new ByteBufferPacket(((ByteBuffer)inputByteBuffer.flip()).slice());
    145
    146 	            channelListener.onPacket(data);
    147 	            // Keep the remaining buffer around to fill with data.
    148 	            inputByteBuffer = remaining;
    149 	            requestNextRead();
    150
    151             } else {
    152                 channelListener.onPacket(EOSPacket.EOS_PACKET);
    153             }
    154
    155         } catch (IOException e) {
    156             channelListener.onPacketError(e);
    157         }
    158     }
    159
    160     private void requestNextRead() throws InterruptedIOException {
    161
    162         // Don't do next read if we have stopped running.
    163         if( !running.get() ) {
    164             doneLatch.release();
    165             return;
    166         }
    167
    168         try {
    169
    170             if( inputByteBuffer==null || !inputByteBuffer.hasRemaining() ) {
    171                 inputByteBuffer = allocateBuffer();
    172             }
    173
    174             IAsyncFuture future = socketChannel.read(inputByteBuffer);
    175             future.addCompletionListener(this, null, false);
    176
    177         } catch (InterruptedException e) {
    178             throw new InterruptedIOException();
    179         }
    180
    181     }
    182
    183     public InetAddress getInetAddress() {
    184         return socket.getInetAddress();
    185     }
    186     public boolean getKeepAlive() throws SocketException {
    187         return socket.getKeepAlive();
    188     }
    189     public InetAddress getLocalAddress() {
    190         return socket.getLocalAddress();
    191     }
    192     public int getLocalPort() {
    193         return socket.getLocalPort();
    194     }
    195     public SocketAddress getLocalSocketAddress() {
    196         return socket.getLocalSocketAddress();
    197     }
    198     public boolean getOOBInline() throws SocketException {
    199         return socket.getOOBInline();
    200     }
    201     public int getPort() {
    202         return socket.getPort();
    203     }
    204     public int getReceiveBufferSize() throws SocketException {
    205         return socket.getReceiveBufferSize();
    206     }
    207     public SocketAddress getRemoteSocketAddress() {
    208         return socket.getRemoteSocketAddress();
    209     }
    210     public boolean getReuseAddress() throws SocketException {
    211         return socket.getReuseAddress();
    212     }
    213     public int getSendBufferSize() throws SocketException {
    214         return socket.getSendBufferSize();
    215     }
    216     public int getSoLinger() throws SocketException {
    217         return socket.getSoLinger();
    218     }
    219     public int getSoTimeout() throws SocketException {
    220         return socket.getSoTimeout();
    221     }
    222     public boolean getTcpNoDelay() throws SocketException {
    223         return socket.getTcpNoDelay();
    224     }
    225     public int getTrafficClass() throws SocketException {
    226         return socket.getTrafficClass();
    227     }
    228     public boolean isBound() {
    229         return socket.isBound();
    230     }
    231     public boolean isClosed() {
    232         return socket.isClosed();
    233     }
    234     public boolean isConnected() {
    235         return socket.isConnected();
    236     }
    237     public void setKeepAlive(boolean on) throws SocketException {
    238         socket.setKeepAlive(on);
    239     }
    240     public void setOOBInline(boolean on) throws SocketException {
    241         socket.setOOBInline(on);
    242     }
    243     public void setReceiveBufferSize(int size) throws SocketException {
    244         socket.setReceiveBufferSize(size);
    245     }
    246     public void setReuseAddress(boolean on) throws SocketException {
    247         socket.setReuseAddress(on);
    248     }
    249     public void setSendBufferSize(int size) throws SocketException {
    250         socket.setSendBufferSize(size);
    251     }
    252     public void setSoLinger(boolean on, int linger) throws SocketException {
    253         socket.setSoLinger(on, linger);
    254     }
    255     public void setTcpNoDelay(boolean on) throws SocketException {
    256         socket.setTcpNoDelay(on);
    257     }
    258     public void setTrafficClass(int tc) throws SocketException {
    259         socket.setTrafficClass(tc);
    260     }
    261     public String toString() {
    262         return "AIO Connection: "+getLocalSocketAddress()+" -> "+getRemoteSocketAddress();
    263     }
    264  }