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.ember;
    19
    20  import java.io.IOException;
    21  import javax.jms.JMSException;
    22  import org.apache.commons.logging.Log;
    23  import org.apache.commons.logging.LogFactory;
    24  import org.codehaus.activemq.message.Packet;
    25  import org.codehaus.activemq.message.WireFormat;
    26  import org.codehaus.activemq.transport.AbstractTransportChannel;
    27  import pyrasun.eio.EIOGlobalContext;
    28  import pyrasun.eio.services.EmberServiceController;
    29  import pyrasun.eio.services.EmberServiceException;
    30  import pyrasun.eio.services.bytearray.ByteArrayServerClient;
    31  import pyrasun.eio.services.bytearray.ByteArrayServerClientListener;
    32  import EDU.oswego.cs.dl.util.concurrent.SynchronizedBoolean;
    33
    34  /***
    35   * An EmberIO (using NIO) implementation of a TransportChannel
    36   *
    37   * @version $Revision: 1.13 $
    38   */
    39  public class EmberTransportChannel extends AbstractTransportChannel implements ByteArrayServerClientListener {
    40
    41      private static final Log log = LogFactory.getLog(EmberTransportChannel.class);
    42
    43      private WireFormat wireFormat;
    44      private EIOGlobalContext context;
    45      private EmberServiceController controller;
    46      private ByteArrayServerClient client;
    47
    48      private SynchronizedBoolean closed;
    49      private SynchronizedBoolean started;
    50
    51
    52      /***
    53       * Construct basic helpers
    54       */
    55      protected EmberTransportChannel(WireFormat wireFormat) {
    56          this.wireFormat = wireFormat;
    57
    58          closed = new SynchronizedBoolean(false);
    59          started = new SynchronizedBoolean(false);
    60      }
    61
    62      /***
    63       * Connect to a remote Node - e.g. a Broker
    64       */
    65      public EmberTransportChannel(WireFormat wireFormat, EIOGlobalContext context, EmberServiceController controller, ByteArrayServerClient client) {
    66          this(wireFormat);
    67          this.context = context;
    68          this.client = client;
    69          this.controller = controller;
    70          client.setListener(this);
    71      }
    72
    73      /***
    74       * close the channel
    75       */
    76      public void stop() {
    77          super.stop();
    78          if (closed.commit(false, true)) {
    79              try {
    80                  // on the server side don't shut down the controller, the server does that
    81                  if (controller != null) {
    82                      controller.stopAll();
    83                  }
    84                  if (context != null) {
    85                      context.stop();
    86                  }
    87              }
    88              catch (EmberServiceException e) {
    89                  log.error("Caught while closing: " + e, e);
    90              }
    91          }
    92      }
    93
    94      /***
    95       * start listeneing for events
    96       *
    97       * @throws JMSException if an error occurs
    98       */
    99      public void start() throws JMSException {
    100         if (started.commit(false, true)) {
    101
    102             try {
    103                 // when using a transport channel created from a server
    104                 // we don't need to initialise these things
    105                 if (context != null) {
    106                     context.start();
    107                 }
    108                 if (controller != null) {
    109                     controller.startAll();
    110                 }
    111             }
    112             catch (EmberServiceException e) {
    113                 JMSException jmsEx = new JMSException("Error starting NIO client: " + e.getMessage());
    114                 jmsEx.setLinkedException(e);
    115                 throw jmsEx;
    116             }
    117         }
    118     }
    119
    120
    121     /***
    122      * Asynchronously send a Packet
    123      *
    124      * @param packet
    125      * @throws JMSException
    126      */
    127     public void asyncSend(Packet packet) throws JMSException {
    128         try {
    129             byte[] bytes = wireFormat.toBytes(packet);
    130             // lets sync for now to avoid multiple threads writing to the same socket
    131             synchronized (client) {
    132                 client.write(bytes);
    133             }
    134         }
    135         catch (IOException e) {
    136             throw createJMSException("Failed to write packet: " + packet + ". ", e);
    137         }
    138     }
    139
    140
    141     public boolean isMulticast() {
    142         return false;
    143     }
    144
    145     /***
    146      * Factory method to create a JMSException which is linked to the base exception
    147      */
    148     protected JMSException createJMSException(String message, Exception ex) {
    149         JMSException jmsEx = new JMSException(message + ex.getMessage());
    150         jmsEx.setLinkedException(ex);
    151         return jmsEx;
    152     }
    153
    154     /***
    155      * pretty print for object
    156      *
    157      * @return String representation of this object
    158      */
    159     public String toString() {
    160         return "EmberTransportChannel: " + client;
    161     }
    162
    163     public void newMessage(ByteArrayServerClient client, Object msg) {
    164         byte[] bytes = (byte[]) msg;
    165         Packet packet = null;
    166         try {
    167             packet = wireFormat.fromBytes(bytes);
    168             doConsumePacket(packet);
    169         }
    170         catch (IOException e) {
    171             log.error("Could not parse byte[] of size: " + bytes.length + ". Reason: " + e, e);
    172         }
    173
    174
    175     }
    176 }