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.InputStream;
    23  import java.io.OutputStream;
    24  import java.net.InetAddress;
    25  import java.net.Socket;
    26  import java.net.SocketAddress;
    27  import java.net.SocketException;
    28  import java.net.SocketTimeoutException;
    29
    30  import org.activeio.Packet;
    31  import org.activeio.SynchChannel;
    32  import org.activeio.SynchChannelServer;
    33  import org.activeio.Packet.ByteSequence;
    34  import org.activeio.packet.ByteArrayPacket;
    35  import org.activeio.packet.EOSPacket;
    36  import org.activeio.packet.EmptyPacket;
    37
    38  /***
    39   * A {@see org.activeio.SynchChannel} implementation that uses a {@see java.net.Socket}
    40   *  to talk to the network.
    41   *
    42   * @version $Revision$
    43   */
    44  public class SocketSynchChannel implements SynchChannel, SocketMetadata {
    45
    46      protected static final int DEFAULT_BUFFER_SIZE = 64 * 1024;
    47
    48      private final Socket socket;
    49      private final OutputStream out;
    50      private final InputStream in;
    51      private boolean disposed;
    52      private int curentSoTimeout;
    53
    54      private Packet inputPacket;
    55
    56      protected SocketSynchChannel(Socket socket) throws IOException {
    57          this.socket = socket;
    58          socket.setReceiveBufferSize(DEFAULT_BUFFER_SIZE);
    59          socket.setSendBufferSize(DEFAULT_BUFFER_SIZE);
    60          in = socket.getInputStream();
    61          out = socket.getOutputStream();
    62      }
    63
    64      /***
    65       * @see org.activeio.SynchChannel#read(long)
    66       */
    67      public org.activeio.Packet read(long timeout) throws IOException {
    68          try {
    69
    70              if( timeout==SynchChannelServer.WAIT_FOREVER_TIMEOUT )
    71                  setSoTimeout( 0 );
    72              else if( timeout==SynchChannelServer.NO_WAIT_TIMEOUT )
    73                  setSoTimeout( 1 );
    74              else
    75                  setSoTimeout( (int)timeout );
    76
    77              if( inputPacket==null || !inputPacket.hasRemaining() ) {
    78                  inputPacket = allocatePacket();
    79              }
    80
    81              ByteSequence sequence = inputPacket.asByteSequence();
    82              int size = in.read(sequence.getData(), sequence.getOffset(), sequence.getLength());
    83              if( size == -1 )
    84                  return EOSPacket.EOS_PACKET;
    85              if( size == 0 )
    86                  return EmptyPacket.EMPTY_PACKET;
    87              inputPacket.position(size);
    88
    89              Packet remaining = inputPacket.slice();
    90              inputPacket.flip();
    91              Packet data = inputPacket.slice();
    92
    93              // Keep the remaining buffer around to fill with data.
    94              inputPacket = remaining;
    95              return data;
    96
    97          } catch (SocketTimeoutException e) {
    98              return null;
    99          }
    100     }
    101
    102     private Packet allocatePacket() {
    103         return new ByteArrayPacket(new byte[DEFAULT_BUFFER_SIZE]);
    104     }
    105
    106     protected void setSoTimeout(int i) throws SocketException {
    107         if( curentSoTimeout != i ) {
    108             socket.setSoTimeout(i);
    109             curentSoTimeout = i;
    110         }
    111     }
    112
    113     /***
    114      * @see org.activeio.Channel#write(org.activeio.channel.Packet)
    115      */
    116     public void write(Packet packet) throws IOException {
    117         packet.writeTo(out);
    118     }
    119
    120     /***
    121      * @see org.activeio.Channel#flush()
    122      */
    123     public void flush() throws IOException {
    124         out.flush();
    125     }
    126
    127     /***
    128      * @see org.activeio.Disposable#dispose()
    129      */
    130     public void dispose() {
    131         if (disposed)
    132             return;
    133
    134         try {
    135             out.close();
    136         } catch (IOException ignore) {
    137         }
    138         try {
    139             in.close();
    140         } catch (IOException ignore) {
    141         }
    142         try {
    143             socket.close();
    144         } catch (IOException ignore) {
    145         }
    146         disposed = true;
    147     }
    148
    149     public void start() throws IOException {
    150     }
    151     public void stop(long timeout) throws IOException {
    152     }
    153
    154     public InetAddress getInetAddress() {
    155         return socket.getInetAddress();
    156     }
    157     public boolean getKeepAlive() throws SocketException {
    158         return socket.getKeepAlive();
    159     }
    160     public InetAddress getLocalAddress() {
    161         return socket.getLocalAddress();
    162     }
    163     public int getLocalPort() {
    164         return socket.getLocalPort();
    165     }
    166     public SocketAddress getLocalSocketAddress() {
    167         return socket.getLocalSocketAddress();
    168     }
    169     public boolean getOOBInline() throws SocketException {
    170         return socket.getOOBInline();
    171     }
    172     public int getPort() {
    173         return socket.getPort();
    174     }
    175     public int getReceiveBufferSize() throws SocketException {
    176         return socket.getReceiveBufferSize();
    177     }
    178     public SocketAddress getRemoteSocketAddress() {
    179         return socket.getRemoteSocketAddress();
    180     }
    181     public boolean getReuseAddress() throws SocketException {
    182         return socket.getReuseAddress();
    183     }
    184     public int getSendBufferSize() throws SocketException {
    185         return socket.getSendBufferSize();
    186     }
    187     public int getSoLinger() throws SocketException {
    188         return socket.getSoLinger();
    189     }
    190     public int getSoTimeout() throws SocketException {
    191         return socket.getSoTimeout();
    192     }
    193     public boolean getTcpNoDelay() throws SocketException {
    194         return socket.getTcpNoDelay();
    195     }
    196     public int getTrafficClass() throws SocketException {
    197         return socket.getTrafficClass();
    198     }
    199     public boolean isBound() {
    200         return socket.isBound();
    201     }
    202     public boolean isClosed() {
    203         return socket.isClosed();
    204     }
    205     public boolean isConnected() {
    206         return socket.isConnected();
    207     }
    208     public void setKeepAlive(boolean on) throws SocketException {
    209         socket.setKeepAlive(on);
    210     }
    211     public void setOOBInline(boolean on) throws SocketException {
    212         socket.setOOBInline(on);
    213     }
    214     public void setReceiveBufferSize(int size) throws SocketException {
    215         socket.setReceiveBufferSize(size);
    216     }
    217     public void setReuseAddress(boolean on) throws SocketException {
    218         socket.setReuseAddress(on);
    219     }
    220     public void setSendBufferSize(int size) throws SocketException {
    221         socket.setSendBufferSize(size);
    222     }
    223     public void setSoLinger(boolean on, int linger) throws SocketException {
    224         socket.setSoLinger(on, linger);
    225     }
    226     public void setTcpNoDelay(boolean on) throws SocketException {
    227         socket.setTcpNoDelay(on);
    228     }
    229     public void setTrafficClass(int tc) throws SocketException {
    230         socket.setTrafficClass(tc);
    231     }
    232
    233     public Object narrow(Class target) {
    234         if( target.isAssignableFrom(getClass()) ) {
    235             return this;
    236         }
    237         return null;
    238     }
    239
    240     public String toString() {
    241         return "Socket Connection: "+getLocalSocketAddress()+" -> "+getRemoteSocketAddress();
    242     }
    243 }