View Javadoc

    1   /***
    2    *
    3    * Copyright 2004 Protique Ltd
    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  package org.codehaus.activemq.transport.tcp;
    19
    20  import EDU.oswego.cs.dl.util.concurrent.BoundedBuffer;
    21  import EDU.oswego.cs.dl.util.concurrent.BoundedChannel;
    22  import EDU.oswego.cs.dl.util.concurrent.BoundedLinkedQueue;
    23  import EDU.oswego.cs.dl.util.concurrent.Executor;
    24  import EDU.oswego.cs.dl.util.concurrent.PooledExecutor;
    25  import EDU.oswego.cs.dl.util.concurrent.SynchronizedBoolean;
    26  import org.apache.commons.logging.Log;
    27  import org.apache.commons.logging.LogFactory;
    28  import org.codehaus.activemq.message.Packet;
    29  import org.codehaus.activemq.message.WireFormat;
    30  import org.codehaus.activemq.transport.AbstractTransportChannel;
    31  import org.codehaus.activemq.util.JMSExceptionHelper;
    32
    33  import javax.jms.JMSException;
    34  import java.io.BufferedInputStream;
    35  import java.io.DataInputStream;
    36  import java.io.DataOutputStream;
    37  import java.io.IOException;
    38  import java.io.InterruptedIOException;
    39  import java.net.InetAddress;
    40  import java.net.Socket;
    41  import java.net.SocketTimeoutException;
    42  import java.net.URI;
    43  import java.net.UnknownHostException;
    44
    45  /***
    46   * A tcp implementation of a TransportChannel
    47   *
    48   * @version $Revision: 1.38 $
    49   */
    50  public class TcpTransportChannel extends AbstractTransportChannel implements Runnable {
    51      private static final int SOCKET_BUFFER_SIZE = 64 * 1024;
    52      private static final int SO_TIMEOUT = 5000;
    53      private static final Log log = LogFactory.getLog(TcpTransportChannel.class);
    54
    55      protected Socket socket;
    56      private WireFormat wireFormat;
    57      private DataOutputStream dataOut;
    58      private DataInputStream dataIn;
    59      private SynchronizedBoolean closed;
    60      private SynchronizedBoolean started;
    61      private Object outboundLock;
    62      private Executor executor;
    63      private Thread thread;
    64      private boolean useAsyncSend = false;
    65      private boolean serverSide = false;
    66      private BoundedChannel exceptionsList;
    67
    68      /***
    69       * Construct basic helpers
    70       */
    71      protected TcpTransportChannel(WireFormat wireFormat) {
    72          this.wireFormat = wireFormat;
    73          closed = new SynchronizedBoolean(false);
    74          started = new SynchronizedBoolean(false);
    75          // there's not much point logging all exceptions, lets just keep a few around
    76          exceptionsList = new BoundedLinkedQueue(10);
    77          outboundLock = new Object();
    78          if (useAsyncSend) {
    79              executor = new PooledExecutor(new BoundedBuffer(1000), 1);
    80          }
    81      }
    82
    83      /***
    84       * Connect to a remote Node - e.g. a Broker
    85       *
    86       * @param remoteLocation
    87       * @throws JMSException
    88       */
    89      public TcpTransportChannel(WireFormat wireFormat, URI remoteLocation) throws JMSException {
    90          this(wireFormat);
    91          try {
    92              this.socket = createSocket(remoteLocation);
    93              socket.setReceiveBufferSize(SOCKET_BUFFER_SIZE);
    94              socket.setSendBufferSize(SOCKET_BUFFER_SIZE);
    95              BufferedInputStream buffIn = new BufferedInputStream(socket.getInputStream());
    96              this.dataIn = new DataInputStream(buffIn);
    97              TcpBufferedOutputStream buffOut = new TcpBufferedOutputStream(socket.getOutputStream());
    98              this.dataOut = new DataOutputStream(buffOut);
    99          }
    100         catch (Exception ioe) {
    101             throw JMSExceptionHelper.newJMSException("Initialization of TcpTransportChannel failed: " + ioe, ioe);
    102         }
    103     }
    104
    105     /***
    106      * Connect to a remote Node - e.g. a Broker
    107      *
    108      * @param remoteLocation
    109      * @param localLocation  - e.g. local InetAddress and local port
    110      * @throws JMSException
    111      */
    112     public TcpTransportChannel(WireFormat wireFormat, URI remoteLocation, URI localLocation) throws JMSException {
    113         this(wireFormat);
    114         try {
    115             this.socket = createSocket(remoteLocation, localLocation);
    116             socket.setReceiveBufferSize(SOCKET_BUFFER_SIZE);
    117             socket.setSendBufferSize(SOCKET_BUFFER_SIZE);
    118             BufferedInputStream buffIn = new BufferedInputStream(socket.getInputStream());
    119             this.dataIn = new DataInputStream(buffIn);
    120             TcpBufferedOutputStream buffOut = new TcpBufferedOutputStream(socket.getOutputStream());
    121             this.dataOut = new DataOutputStream(buffOut);
    122         }
    123         catch (Exception ioe) {
    124             throw JMSExceptionHelper.newJMSException("Initialization of TcpTransportChannel failed: " + ioe, ioe);
    125         }
    126     }
    127
    128     /***
    129      * @param socket
    130      * @throws JMSException
    131      */
    132     public TcpTransportChannel(WireFormat wireFormat, Socket socket, Executor executor) throws JMSException {
    133         this(wireFormat);
    134         this.socket = socket;
    135         this.executor = executor;
    136         this.serverSide = true;
    137         try {
    138             socket.setReceiveBufferSize(SOCKET_BUFFER_SIZE);
    139             socket.setSendBufferSize(SOCKET_BUFFER_SIZE);
    140             TcpBufferedOutputStream buffOut = new TcpBufferedOutputStream(socket.getOutputStream());
    141             this.dataOut = new DataOutputStream(buffOut);
    142             BufferedInputStream buffIn = new BufferedInputStream(socket.getInputStream());
    143             this.dataIn = new DataInputStream(buffIn);
    144         }
    145         catch (IOException ioe) {
    146             throw JMSExceptionHelper.newJMSException("Initialization of TcpTransportChannel failed: " + ioe, ioe);
    147         }
    148     }
    149
    150     /***
    151      * close the channel
    152      */
    153     public void stop() {
    154         if (closed.commit(false, true)) {
    155             super.stop();
    156             try {
    157                 stopExecutor(executor);
    158                 dataOut.close();
    159                 dataIn.close();
    160                 socket.close();
    161
    162                 // lets wait for the receive thread to terminate
    163                 // TODO thread.join();
    164             }
    165             catch (Exception e) {
    166                 log.warn("Caught while closing: " + e + ". Now Closed", e);
    167             }
    168
    169         }
    170     }
    171
    172     /***
    173      * start listeneing for events
    174      *
    175      * @throws JMSException if an error occurs
    176      */
    177     public void start() throws JMSException {
    178         if (started.commit(false, true)) {
    179             thread = new Thread(this, "Thread:" + toString());
    180             thread.setDaemon(true);
    181             if (!serverSide) {
    182                 thread.setPriority(Thread.NORM_PRIORITY + 2);
    183             }
    184             thread.start();
    185         }
    186     }
    187
    188
    189     /***
    190      * Asynchronously send a Packet
    191      *
    192      * @param packet
    193      * @throws JMSException
    194      */
    195     public void asyncSend(final Packet packet) throws JMSException {
    196         if (executor != null) {
    197             try {
    198                 executor.execute(new Runnable() {
    199                     public void run() {
    200                         try {
    201                             if (!closed.get()) {
    202                                 doAsyncSend(packet);
    203                             }
    204                         }
    205                         catch (JMSException e) {
    206                             try {
    207                                 exceptionsList.put(e);
    208                             }
    209                             catch (InterruptedException e1) {
    210                                 log.warn("Failed to add element to exception list: " + e1);
    211                             }
    212                         }
    213                     }
    214
    215                 });
    216             }
    217             catch (InterruptedException e) {
    218                 log.info("Caught: " + e, e);
    219             }
    220             try {
    221                 JMSException e = (JMSException) exceptionsList.poll(0);
    222                 if (e != null) {
    223                     throw e;
    224                 }
    225             }
    226             catch (InterruptedException e1) {
    227                 log.warn("Failed to remove element to exception list: " + e1);
    228             }
    229         }
    230         else {
    231             doAsyncSend(packet);
    232         }
    233     }
    234
    235     /***
    236      * @return false
    237      */
    238     public boolean isMulticast() {
    239         return false;
    240     }
    241
    242     /***
    243      * reads packets from a Socket
    244      */
    245     public void run() {
    246         log.trace("TCP consumer thread starting");
    247         int count = 0;
    248         while (!closed.get()) {
    249             if (serverSide && ++count > 500) {
    250                 count = 0;
    251                 Thread.yield();
    252             }
    253             int type = 0;
    254             try {
    255                 socket.setSoTimeout(SO_TIMEOUT);
    256                 while ((type = dataIn.read()) == 0) {
    257                 }
    258                 if (type == -1) {
    259                     log.trace("The socket peer is now closed");
    260                     //doClose(new IOException("Socket peer is now closed"));
    261                     stop();
    262                 }
    263                 else {
    264                     socket.setSoTimeout(0);
    265
    266                     Packet packet = wireFormat.readPacket(type, dataIn);
    267                     if (packet != null) {
    268                         doConsumePacket(packet);
    269                     }
    270                 }
    271             }
    272             catch (SocketTimeoutException ste) {
    273                 // ignore timeouts, they are expected
    274             }
    275             catch (InterruptedIOException ioe) {
    276                 // TODO confirm that this really is a bug in the AS/400 JVM
    277                 // Patch for AS/400 JVM
    278
    279                 // lets ignore these exceptions
    280                 // as they typically just indicate the thread was interupted
    281                 // while waiting for input, not that the socket is in error
    282             }
    283             catch (IOException e) {
    284                 doClose(e);
    285             }
    286         }
    287     }
    288
    289     /***
    290      * pretty print for object
    291      *
    292      * @return String representation of this object
    293      */
    294     public String toString() {
    295         return "TcpTransportChannel: " + socket;
    296     }
    297
    298     /***
    299      * Actually performs the async send of a packet
    300      *
    301      * @param packet
    302      * @throws JMSException
    303      */
    304     protected void doAsyncSend(Packet packet) throws JMSException {
    305         try {
    306             synchronized (outboundLock) {
    307                 wireFormat.writePacket(packet, dataOut);
    308                 dataOut.flush();
    309             }
    310         }
    311         catch (IOException e) {
    312             throw JMSExceptionHelper.newJMSException("asyncSend failed: " + e, e);
    313         }
    314     }
    315
    316
    317     private void doClose(Exception ex) {
    318         if (!closed.get()) {
    319             onAsyncException(JMSExceptionHelper.newJMSException("Error reading socket: " + ex, ex));
    320             stop();
    321         }
    322     }
    323
    324     /***
    325      * Factory method to create a new socket
    326      *
    327      * @param remoteLocation the URI to connect to
    328      * @return the newly created socket
    329      * @throws UnknownHostException
    330      * @throws IOException
    331      */
    332     protected Socket createSocket(URI remoteLocation) throws UnknownHostException, IOException {
    333         return new Socket(remoteLocation.getHost(), remoteLocation.getPort());
    334     }
    335
    336     /***
    337      * Factory method to create a new socket
    338      *
    339      * @param remoteLocation
    340      * @param localLocation
    341      * @return @throws IOException
    342      * @throws IOException
    343      * @throws UnknownHostException
    344      */
    345     protected Socket createSocket(URI remoteLocation, URI localLocation) throws IOException, UnknownHostException {
    346         return new Socket(remoteLocation.getHost(), remoteLocation.getPort(), InetAddress.getByName(localLocation
    347                 .getHost()), localLocation.getPort());
    348     }
    349
    350 }