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.jabber;
    19
    20  import org.apache.commons.logging.Log;
    21  import org.apache.commons.logging.LogFactory;
    22  import org.codehaus.activemq.message.ActiveMQBytesMessage;
    23  import org.codehaus.activemq.message.ActiveMQMessage;
    24  import org.codehaus.activemq.message.ActiveMQObjectMessage;
    25  import org.codehaus.activemq.message.ActiveMQTextMessage;
    26  import org.codehaus.activemq.message.Packet;
    27  import org.codehaus.activemq.message.WireFormat;
    28
    29  import javax.jms.JMSException;
    30  import java.io.DataInput;
    31  import java.io.DataOutput;
    32  import java.io.IOException;
    33  import java.io.Serializable;
    34  import java.util.Hashtable;
    35  import java.util.Iterator;
    36  import java.util.Map;
    37
    38  /***
    39   * A wire format which uses XMPP format of messages
    40   *
    41   * @version $Revision: 1.2 $
    42   */
    43  public class JabberWireFormat extends WireFormat {
    44      private static final Log log = LogFactory.getLog(JabberWireFormat.class);
    45
    46      public WireFormat copy() {
    47          return new JabberWireFormat();
    48      }
    49
    50      public Packet readPacket(DataInput in) throws IOException {
    51          return null;  /*** TODO */
    52      }
    53
    54      public Packet readPacket(int firstByte, DataInput in) throws IOException {
    55          return null;  /*** TODO */
    56      }
    57
    58      public void writePacket(Packet packet, DataOutput out) throws IOException, JMSException {
    59          switch (packet.getPacketType()) {
    60              case Packet.ACTIVEMQ_MESSAGE:
    61                  writeMessage((ActiveMQMessage) packet, "", out);
    62                  break;
    63
    64              case Packet.ACTIVEMQ_TEXT_MESSAGE:
    65                  writeTextMessage((ActiveMQTextMessage) packet, out);
    66                  break;
    67
    68              case Packet.ACTIVEMQ_BYTES_MESSAGE:
    69                  writeBytesMessage((ActiveMQBytesMessage) packet, out);
    70                  break;
    71
    72              case Packet.ACTIVEMQ_OBJECT_MESSAGE:
    73                  writeObjectMessage((ActiveMQObjectMessage) packet, out);
    74                  break;
    75
    76              case Packet.ACTIVEMQ_MAP_MESSAGE:
    77              case Packet.ACTIVEMQ_STREAM_MESSAGE:
    78
    79
    80              case Packet.ACTIVEMQ_BROKER_INFO:
    81              case Packet.ACTIVEMQ_CONNECTION_INFO:
    82              case Packet.ACTIVEMQ_MSG_ACK:
    83              case Packet.CONSUMER_INFO:
    84              case Packet.DURABLE_UNSUBSCRIBE:
    85              case Packet.INT_RESPONSE_RECEIPT_INFO:
    86              case Packet.PRODUCER_INFO:
    87              case Packet.RECEIPT_INFO:
    88              case Packet.RESPONSE_RECEIPT_INFO:
    89              case Packet.SESSION_INFO:
    90              case Packet.TRANSACTION_INFO:
    91              case Packet.XA_TRANSACTION_INFO:
    92              default:
    93                  log.warn("Ignoring message type: " + packet.getPacketType() + " packet: " + packet);
    94          }
    95      }
    96
    97      // Implementation methods
    98      //-------------------------------------------------------------------------
    99      protected void writeObjectMessage(ActiveMQObjectMessage message, DataOutput out) throws JMSException, IOException {
    100         Serializable object = message.getObject();
    101         String text = (object != null) ? object.toString() : "";
    102         writeMessage(message, text, out);
    103     }
    104
    105     protected void writeTextMessage(ActiveMQTextMessage message, DataOutput out) throws JMSException, IOException {
    106         writeMessage(message, message.getText(), out);
    107     }
    108
    109     protected void writeBytesMessage(ActiveMQBytesMessage message, DataOutput out) throws IOException {
    110         byte[] data = message.getBodyAsBytes();
    111         String text = encodeBinary(data);
    112         writeMessage(message, text, out);
    113     }
    114
    115     protected void writeMessage(ActiveMQMessage message, String body, DataOutput out) throws IOException {
    116         String type = getXmppType(message);
    117
    118         StringBuffer buffer = new StringBuffer("<");
    119         buffer.append(type);
    120         buffer.append(" to='");
    121         buffer.append(message.getJMSDestination().toString());
    122         buffer.append("' from='");
    123         buffer.append(message.getJMSReplyTo().toString());
    124         String messageID = message.getJMSMessageID();
    125         if (messageID != null) {
    126             buffer.append("' id='");
    127             buffer.append(messageID);
    128         }
    129
    130         Hashtable properties = message.getProperties();
    131         if (properties != null) {
    132             for (Iterator iter = properties.entrySet().iterator(); iter.hasNext();) {
    133                 Map.Entry entry = (Map.Entry) iter.next();
    134                 Object key = entry.getKey();
    135                 Object value = entry.getValue();
    136                 if (value != null) {
    137                     buffer.append("' ");
    138                     buffer.append(key.toString());
    139                     buffer.append("='");
    140                     buffer.append(value.toString());
    141                 }
    142             }
    143         }
    144
    145         buffer.append("'>");
    146
    147         String id = message.getJMSCorrelationID();
    148         if (id != null) {
    149             buffer.append("<thread>");
    150             buffer.append(id);
    151             buffer.append("</thread>");
    152         }
    153         buffer.append(body);
    154         buffer.append("</");
    155         buffer.append(type);
    156         buffer.append(">");
    157
    158         out.write(buffer.toString().getBytes());
    159     }
    160
    161     protected String encodeBinary(byte[] data) {
    162         // TODO
    163         throw new RuntimeException("Not implemented yet!");
    164     }
    165
    166     protected String getXmppType(ActiveMQMessage message) {
    167         String type = message.getJMSType();
    168         if (type == null) {
    169             type = "message";
    170         }
    171         return type;
    172     }
    173 }