View Javadoc

    1   /***
    2    *
    3    * Copyright 2004 Hiram Chirino
    4    * Copyright 2004 Protique Ltd
    5    *
    6    * Licensed under the Apache License, Version 2.0 (the "License");
    7    * you may not use this file except in compliance with the License.
    8    * You may obtain a copy of the License at
    9    *
    10   * http://www.apache.org/licenses/LICENSE-2.0
    11   *
    12   * Unless required by applicable law or agreed to in writing, software
    13   * distributed under the License is distributed on an "AS IS" BASIS,
    14   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    15   * See the License for the specific language governing permissions and
    16   * limitations under the License.
    17   *
    18   **/
    19  package org.codehaus.activemq.transport.gnet;
    20
    21  import java.io.ByteArrayOutputStream;
    22  import java.io.DataInputStream;
    23  import java.io.DataOutputStream;
    24  import java.io.IOException;
    25  import java.io.InputStream;
    26  import java.net.InetAddress;
    27  import java.net.InetSocketAddress;
    28  import java.net.URI;
    29  import java.net.UnknownHostException;
    30  import java.nio.ByteBuffer;
    31  import java.util.ArrayList;
    32  import javax.jms.JMSException;
    33  import org.apache.commons.logging.Log;
    34  import org.apache.commons.logging.LogFactory;
    35  import org.apache.geronimo.network.SelectorManager;
    36  import org.apache.geronimo.network.protocol.AbstractProtocol;
    37  import org.apache.geronimo.network.protocol.DownPacket;
    38  import org.apache.geronimo.network.protocol.PlainDownPacket;
    39  import org.apache.geronimo.network.protocol.Protocol;
    40  import org.apache.geronimo.network.protocol.ProtocolException;
    41  import org.apache.geronimo.network.protocol.SocketProtocol;
    42  import org.apache.geronimo.network.protocol.UpPacket;
    43  import org.apache.geronimo.pool.ClockPool;
    44  import org.apache.geronimo.pool.ThreadPool;
    45  import org.codehaus.activemq.message.Packet;
    46  import org.codehaus.activemq.message.WireFormat;
    47  import org.codehaus.activemq.transport.AbstractTransportChannel;
    48  import EDU.oswego.cs.dl.util.concurrent.Latch;
    49  import EDU.oswego.cs.dl.util.concurrent.SynchronizedBoolean;
    50
    51  /***
    52   * An implementation of a TransportChannel which uses the Geronimo network layer
    53   * for connectivity.
    54   *
    55   * @version $Revision: 1.14 $
    56   */
    57  public class GTransportChannel extends AbstractTransportChannel {
    58      private static final Log log = LogFactory.getLog(GTransportChannel.class);
    59
    60      private SynchronizedBoolean closed;
    61      private SynchronizedBoolean started;
    62      private Protocol protocol;
    63      private Latch dispatchLatch;
    64      private ThreadPool threadPool;
    65      private WireFormat wireFormat;
    66
    67      /***
    68       * Construct basic helpers
    69       */
    70      protected GTransportChannel(WireFormat wireFormat, ThreadPool tp) {
    71          this.wireFormat = wireFormat;
    72          closed = new SynchronizedBoolean(false);
    73          started = new SynchronizedBoolean(false);
    74          dispatchLatch = new Latch();
    75          threadPool = tp;
    76      }
    77
    78      /***
    79       * @param protocol
    80       */
    81      public GTransportChannel(WireFormat wireFormat, Protocol protocol, ThreadPool tp) {
    82          this(wireFormat, tp);
    83          init(protocol);
    84      }
    85
    86      /***
    87       * @param remoteLocation
    88       * @param localLocation
    89       */
    90      public GTransportChannel(WireFormat wireFormat, URI remoteLocation, URI localLocation,
    91                               SelectorManager sm, ThreadPool tp, ClockPool cp)
    92              throws UnknownHostException, ProtocolException {
    93          this(wireFormat, tp);
    94
    95  /*
    96  		ControlClientProtocolStack clientStack = new ControlClientProtocolStack();
    97  		clientStack.setClassLoader(Thread.currentThread()
    98  				.getContextClassLoader());
    99  		clientStack.setThreadPool(tp);
    100 		clientStack.setClockPool(cp);
    101 		clientStack.setSelectorManager(sm);
    102 */
    103         SocketProtocol sp = new SocketProtocol();
    104         sp.setTimeout(1000 * 30);
    105         if (localLocation != null) {
    106             sp.setInterface(new InetSocketAddress(InetAddress
    107                     .getByName(localLocation.getHost()), localLocation
    108                     .getPort()));
    109         }
    110         sp.setAddress(new InetSocketAddress(InetAddress
    111                 .getByName(remoteLocation.getHost()), remoteLocation
    112                 .getPort()));
    113         sp.setSelectorManager(sm);
    114 /*
    115 		clientStack.push(sp);
    116 		ControlClientProtocol ccp = new ControlClientProtocol();
    117 		ccp.setTimeout(1000 * 30);
    118 		clientStack.push(ccp);
    119 		clientStack.setup();
    120 */
    121 //		init(clientStack);
    122         init(sp);
    123         sp.setup();
    124     }
    125
    126     /***
    127      * @param protocol
    128      */
    129     private void init(Protocol protocol) {
    130         this.protocol = protocol;
    131         // Hookup a new Up protocol so we can get the up stream packets.
    132         protocol.setUpProtocol(new AbstractProtocol() {
    133             public void setup() {
    134             }
    135
    136             public void drain() {
    137             }
    138
    139             public void teardown() {
    140             }
    141
    142             public void sendUp(final UpPacket p) {
    143                 try {
    144                     log.trace("AQUIRING: " + dispatchLatch);
    145                     dispatchLatch.acquire();
    146                     log.trace("AQUIRED: " + dispatchLatch);
    147
    148                     dispatch(p);
    149                 }
    150                 catch (InterruptedException e) {
    151                     log.warn("Caught exception dispatching packet: " + p + ". Reason: "
    152                             + e, e);
    153                     // TODO: notify exception listner and close the connection.
    154                 }
    155             }
    156
    157             public void sendDown(DownPacket p) throws ProtocolException {
    158                 getDownProtocol().sendDown(p);
    159             }
    160
    161             public void flush() throws ProtocolException {
    162 				getDownProtocol().flush();
    163 			}
    164         });
    165     }
    166
    167     private void dispatch(UpPacket p) {
    168         try {
    169             // Dont dispatch messages until the channel is started..
    170             Packet packet = toPacket(p);
    171             log.trace("<<<< SENDING UP <<<< " + packet);
    172             if (packet != null) {
    173                 doConsumePacket(packet);
    174             }
    175         }
    176         catch (IOException e) {
    177             log.warn("Caught exception dispatching packet: " + p + ". Reason: "
    178                     + e, e);
    179             // TODO: notify exception listner and close the connection.
    180         }
    181     }
    182
    183     /***
    184      * close the channel
    185      */
    186     public void stop() {
    187         super.stop();
    188         if (closed.commit(false, true)) {
    189             try {
    190                 protocol.drain();
    191             }
    192             catch (Exception e) {
    193                 log.trace(toString() + " now closed");
    194             }
    195         }
    196     }
    197
    198     /***
    199      * start listeneing for events
    200      *
    201      * @throws JMSException if an error occurs
    202      */
    203     public void start() throws JMSException {
    204         if (started.commit(false, true)) {
    205             // Allow messages to get dispatched.
    206             dispatchLatch.release();
    207         }
    208     }
    209
    210
    211
    212     /***
    213      * Asynchronously send a Packet
    214      *
    215      * @param packet
    216      * @throws JMSException
    217      */
    218     public void asyncSend(Packet packet) throws JMSException {
    219         try {
    220             if (log.isTraceEnabled()) {
    221                 log.trace(">>>> ASYNC SENDING DOWN >>>> " + packet);
    222             }
    223
    224             // lets sync for now to avoid multiple threads writing to the same socket
    225             synchronized (protocol) {
    226                 protocol.sendDown(toPlainDownPacket(packet));
    227             }
    228         }
    229         catch (IOException e) {
    230             System.out.println("Caught: " + e);
    231             e.printStackTrace();
    232             JMSException jmsEx = new JMSException("asyncSend failed "
    233                     + e.getMessage());
    234             jmsEx.setLinkedException(e);
    235             throw jmsEx;
    236         }
    237         catch (ProtocolException e) {
    238             System.out.println("Caught: " + e);
    239             e.printStackTrace();
    240             JMSException jmsEx = new JMSException("asyncSend failed "
    241                     + e.getMessage());
    242             jmsEx.setLinkedException(e);
    243             throw jmsEx;
    244         }
    245     }
    246
    247     public boolean isMulticast() {
    248         return false;
    249     }
    250
    251     protected PlainDownPacket toPlainDownPacket(Packet mqpacket)
    252             throws IOException, JMSException {
    253
    254         ByteArrayOutputStream baos = new ByteArrayOutputStream();
    255         DataOutputStream dos = new DataOutputStream(baos);
    256         wireFormat.writePacket(mqpacket, dos);
    257         dos.close();
    258         ArrayList list = new ArrayList(1);
    259         ByteBuffer buffer = ByteBuffer.wrap(baos.toByteArray());
    260         buffer.limit(buffer.capacity());
    261         list.add(buffer);
    262         PlainDownPacket packet = new PlainDownPacket();
    263         packet.setBuffers(list);
    264         return packet;
    265     }
    266
    267     protected Packet toPacket(UpPacket packet) throws IOException {
    268         final ByteBuffer buffer = packet.getBuffer();
    269         InputStream is = new InputStream() {
    270             public int read() {
    271                 if (!buffer.hasRemaining()) {
    272                     return -1;
    273                 }
    274                 int rc = 0xFF & buffer.get();
    275                 return rc;
    276             }
    277
    278             public synchronized int read(byte[] bytes, int off, int len) {
    279                 len = Math.min(len, buffer.remaining());
    280                 buffer.get(bytes, off, len);
    281                 return len;
    282             }
    283         };
    284         DataInputStream dis = new DataInputStream(is);
    285         return wireFormat.readPacket(dis);
    286     }
    287
    288     /***
    289      * pretty print for object
    290      *
    291      * @return String representation of this object
    292      */
    293     public String toString() {
    294         return "GTransportChannel: " + protocol;
    295     }
    296 }