001/*
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements.  See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License.  You may obtain a copy of the License at
008 *
009 *      http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017package org.apache.activemq.transport.amqp.message;
018
019import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.AMQP_DATA;
020import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.AMQP_NULL;
021import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.AMQP_SEQUENCE;
022import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.AMQP_UNKNOWN;
023import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.AMQP_VALUE_BINARY;
024import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.AMQP_VALUE_LIST;
025import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.AMQP_VALUE_STRING;
026import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.CONTENT_ENCODING;
027import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.CONTENT_TYPE;
028import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.DELIVERY_ANNOTATION_PREFIX;
029import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.EMPTY_BINARY;
030import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.FIRST_ACQUIRER;
031import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.FOOTER_PREFIX;
032import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.HEADER;
033import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.JMS_AMQP_CONTENT_TYPE;
034import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.JMS_AMQP_DELIVERY_ANNOTATION_PREFIX;
035import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.JMS_AMQP_FOOTER_PREFIX;
036import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.JMS_AMQP_MESSAGE_ANNOTATION_PREFIX;
037import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.JMS_AMQP_ORIGINAL_ENCODING;
038import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.JMS_AMQP_PREFIX;
039import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.JMS_AMQP_PREFIX_LENGTH;
040import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.MESSAGE_ANNOTATION_PREFIX;
041import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.MESSAGE_FORMAT;
042import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.NATIVE;
043import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.ORIGINAL_ENCODING;
044import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.PROPERTIES;
045import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.REPLYTO_GROUP_ID;
046import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.SERIALIZED_JAVA_OBJECT_CONTENT_TYPE;
047import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.getBinaryFromMessageBody;
048import static org.apache.activemq.transport.amqp.message.AmqpMessageSupport.getMapFromMessageBody;
049
050import java.io.IOException;
051import java.nio.charset.StandardCharsets;
052import java.util.ArrayList;
053import java.util.Date;
054import java.util.HashMap;
055import java.util.Map;
056
057import javax.jms.JMSException;
058import javax.jms.Message;
059import javax.jms.MessageEOFException;
060import javax.jms.TextMessage;
061
062import org.apache.activemq.command.ActiveMQBytesMessage;
063import org.apache.activemq.command.ActiveMQDestination;
064import org.apache.activemq.command.ActiveMQMapMessage;
065import org.apache.activemq.command.ActiveMQMessage;
066import org.apache.activemq.command.ActiveMQObjectMessage;
067import org.apache.activemq.command.ActiveMQStreamMessage;
068import org.apache.activemq.command.ActiveMQTextMessage;
069import org.apache.activemq.command.CommandTypes;
070import org.apache.activemq.command.MessageId;
071import org.apache.activemq.transport.amqp.AmqpProtocolException;
072import org.apache.activemq.util.JMSExceptionSupport;
073import org.apache.activemq.util.TypeConversionSupport;
074import org.apache.qpid.proton.amqp.Binary;
075import org.apache.qpid.proton.amqp.Symbol;
076import org.apache.qpid.proton.amqp.UnsignedByte;
077import org.apache.qpid.proton.amqp.UnsignedInteger;
078import org.apache.qpid.proton.amqp.messaging.AmqpSequence;
079import org.apache.qpid.proton.amqp.messaging.AmqpValue;
080import org.apache.qpid.proton.amqp.messaging.ApplicationProperties;
081import org.apache.qpid.proton.amqp.messaging.Data;
082import org.apache.qpid.proton.amqp.messaging.DeliveryAnnotations;
083import org.apache.qpid.proton.amqp.messaging.Footer;
084import org.apache.qpid.proton.amqp.messaging.Header;
085import org.apache.qpid.proton.amqp.messaging.MessageAnnotations;
086import org.apache.qpid.proton.amqp.messaging.Properties;
087import org.apache.qpid.proton.amqp.messaging.Section;
088import org.apache.qpid.proton.codec.AMQPDefinedTypes;
089import org.apache.qpid.proton.codec.DecoderImpl;
090import org.apache.qpid.proton.codec.EncoderImpl;
091
092public class JMSMappingOutboundTransformer implements OutboundTransformer {
093
094    public static final Symbol JMS_DEST_TYPE_MSG_ANNOTATION = Symbol.valueOf("x-opt-jms-dest");
095    public static final Symbol JMS_REPLY_TO_TYPE_MSG_ANNOTATION = Symbol.valueOf("x-opt-jms-reply-to");
096
097    private static final String AMQ_SCHEDULED_MESSAGE_PREFIX = "AMQ_SCHEDULED_";
098
099    public static final byte QUEUE_TYPE = 0x00;
100    public static final byte TOPIC_TYPE = 0x01;
101    public static final byte TEMP_QUEUE_TYPE = 0x02;
102    public static final byte TEMP_TOPIC_TYPE = 0x03;
103
104    private final UTF8BufferType utf8BufferEncoding;
105
106    // For now Proton requires that we create a decoder to create an encoder
107    private final DecoderImpl decoder = new DecoderImpl();
108    private final EncoderImpl encoder = new EncoderImpl(decoder);
109    {
110        AMQPDefinedTypes.registerAllTypes(decoder, encoder);
111
112        utf8BufferEncoding = new UTF8BufferType(encoder, decoder);
113
114        encoder.register(utf8BufferEncoding);
115    }
116
117    @Override
118    public EncodedMessage transform(ActiveMQMessage message) throws Exception {
119        if (message == null) {
120            return null;
121        }
122
123        long messageFormat = 0;
124        Header header = null;
125        Properties properties = null;
126        Map<Symbol, Object> daMap = null;
127        Map<Symbol, Object> maMap = null;
128        Map<String,Object> apMap = null;
129        Map<Object, Object> footerMap = null;
130
131        Section body = convertBody(message);
132
133        if (message.isPersistent()) {
134            if (header == null) {
135                header = new Header();
136            }
137            header.setDurable(true);
138        }
139        byte priority = message.getPriority();
140        if (priority != Message.DEFAULT_PRIORITY) {
141            if (header == null) {
142                header = new Header();
143            }
144            header.setPriority(UnsignedByte.valueOf(priority));
145        }
146        String type = message.getType();
147        if (type != null) {
148            if (properties == null) {
149                properties = new Properties();
150            }
151            properties.setSubject(type);
152        }
153        MessageId messageId = message.getMessageId();
154        if (messageId != null) {
155            if (properties == null) {
156                properties = new Properties();
157            }
158            properties.setMessageId(getOriginalMessageId(message));
159        }
160        ActiveMQDestination destination = message.getDestination();
161        if (destination != null) {
162            if (properties == null) {
163                properties = new Properties();
164            }
165            properties.setTo(destination.getQualifiedName());
166            if (maMap == null) {
167                maMap = new HashMap<>();
168            }
169            maMap.put(JMS_DEST_TYPE_MSG_ANNOTATION, destinationType(destination));
170        }
171        ActiveMQDestination replyTo = message.getReplyTo();
172        if (replyTo != null) {
173            if (properties == null) {
174                properties = new Properties();
175            }
176            properties.setReplyTo(replyTo.getQualifiedName());
177            if (maMap == null) {
178                maMap = new HashMap<>();
179            }
180            maMap.put(JMS_REPLY_TO_TYPE_MSG_ANNOTATION, destinationType(replyTo));
181        }
182        String correlationId = message.getCorrelationId();
183        if (correlationId != null) {
184            if (properties == null) {
185                properties = new Properties();
186            }
187            try {
188                properties.setCorrelationId(AMQPMessageIdHelper.INSTANCE.toIdObject(correlationId));
189            } catch (AmqpProtocolException e) {
190                properties.setCorrelationId(correlationId);
191            }
192        }
193        long expiration = message.getExpiration();
194        if (expiration != 0) {
195            long ttl = expiration - System.currentTimeMillis();
196            if (ttl < 0) {
197                ttl = 1;
198            }
199
200            if (header == null) {
201                header = new Header();
202            }
203            header.setTtl(new UnsignedInteger((int) ttl));
204
205            if (properties == null) {
206                properties = new Properties();
207            }
208            properties.setAbsoluteExpiryTime(new Date(expiration));
209        }
210        long timeStamp = message.getTimestamp();
211        if (timeStamp != 0) {
212            if (properties == null) {
213                properties = new Properties();
214            }
215            properties.setCreationTime(new Date(timeStamp));
216        }
217
218        // JMSX Message Properties
219        int deliveryCount = message.getRedeliveryCounter();
220        if (deliveryCount > 0) {
221            if (header == null) {
222                header = new Header();
223            }
224            header.setDeliveryCount(UnsignedInteger.valueOf(deliveryCount));
225        }
226        String userId = message.getUserID();
227        if (userId != null) {
228            if (properties == null) {
229                properties = new Properties();
230            }
231            properties.setUserId(new Binary(userId.getBytes(StandardCharsets.UTF_8)));
232        }
233        String groupId = message.getGroupID();
234        if (groupId != null) {
235            if (properties == null) {
236                properties = new Properties();
237            }
238            properties.setGroupId(groupId);
239        }
240        int groupSequence = message.getGroupSequence();
241        if (groupSequence > 0) {
242            if (properties == null) {
243                properties = new Properties();
244            }
245            properties.setGroupSequence(UnsignedInteger.valueOf(groupSequence));
246        }
247
248        final Map<String, Object> entries;
249        try {
250            entries = message.getProperties();
251        } catch (IOException e) {
252            throw JMSExceptionSupport.create(e);
253        }
254
255        for (Map.Entry<String, Object> entry : entries.entrySet()) {
256            String key = entry.getKey();
257            Object value = entry.getValue();
258
259            if (key.startsWith(JMS_AMQP_PREFIX)) {
260                if (key.startsWith(NATIVE, JMS_AMQP_PREFIX_LENGTH)) {
261                    // skip transformer appended properties
262                    continue;
263                } else if (key.startsWith(ORIGINAL_ENCODING, JMS_AMQP_PREFIX_LENGTH)) {
264                    // skip transformer appended properties
265                    continue;
266                } else if (key.startsWith(MESSAGE_FORMAT, JMS_AMQP_PREFIX_LENGTH)) {
267                    messageFormat = (long) TypeConversionSupport.convert(entry.getValue(), Long.class);
268                    continue;
269                } else if (key.startsWith(HEADER, JMS_AMQP_PREFIX_LENGTH)) {
270                    if (header == null) {
271                        header = new Header();
272                    }
273                    continue;
274                } else if (key.startsWith(PROPERTIES, JMS_AMQP_PREFIX_LENGTH)) {
275                    if (properties == null) {
276                        properties = new Properties();
277                    }
278                    continue;
279                } else if (key.startsWith(MESSAGE_ANNOTATION_PREFIX, JMS_AMQP_PREFIX_LENGTH)) {
280                    if (maMap == null) {
281                        maMap = new HashMap<>();
282                    }
283                    String name = key.substring(JMS_AMQP_MESSAGE_ANNOTATION_PREFIX.length());
284                    maMap.put(Symbol.valueOf(name), value);
285                    continue;
286                } else if (key.startsWith(FIRST_ACQUIRER, JMS_AMQP_PREFIX_LENGTH)) {
287                    if (header == null) {
288                        header = new Header();
289                    }
290                    header.setFirstAcquirer((boolean) TypeConversionSupport.convert(value, Boolean.class));
291                    continue;
292                } else if (key.startsWith(CONTENT_TYPE, JMS_AMQP_PREFIX_LENGTH)) {
293                    if (properties == null) {
294                        properties = new Properties();
295                    }
296                    properties.setContentType(Symbol.getSymbol((String) TypeConversionSupport.convert(value, String.class)));
297                    continue;
298                } else if (key.startsWith(CONTENT_ENCODING, JMS_AMQP_PREFIX_LENGTH)) {
299                    if (properties == null) {
300                        properties = new Properties();
301                    }
302                    properties.setContentEncoding(Symbol.getSymbol((String) TypeConversionSupport.convert(value, String.class)));
303                    continue;
304                } else if (key.startsWith(REPLYTO_GROUP_ID, JMS_AMQP_PREFIX_LENGTH)) {
305                    if (properties == null) {
306                        properties = new Properties();
307                    }
308                    properties.setReplyToGroupId((String) TypeConversionSupport.convert(value, String.class));
309                    continue;
310                } else if (key.startsWith(DELIVERY_ANNOTATION_PREFIX, JMS_AMQP_PREFIX_LENGTH)) {
311                    if (daMap == null) {
312                        daMap = new HashMap<>();
313                    }
314                    String name = key.substring(JMS_AMQP_DELIVERY_ANNOTATION_PREFIX.length());
315                    daMap.put(Symbol.valueOf(name), value);
316                    continue;
317                } else if (key.startsWith(FOOTER_PREFIX, JMS_AMQP_PREFIX_LENGTH)) {
318                    if (footerMap == null) {
319                        footerMap = new HashMap<>();
320                    }
321                    String name = key.substring(JMS_AMQP_FOOTER_PREFIX.length());
322                    footerMap.put(name, value);
323                    continue;
324                }
325            } else if (key.startsWith(AMQ_SCHEDULED_MESSAGE_PREFIX )) {
326                // strip off the scheduled message properties
327                continue;
328            }
329
330            // The property didn't map into any other slot so we store it in the
331            // Application Properties section of the message.
332            if (apMap == null) {
333                apMap = new HashMap<>();
334            }
335            apMap.put(key, value);
336        }
337
338        final AmqpWritableBuffer buffer = new AmqpWritableBuffer();
339        encoder.setByteBuffer(buffer);
340
341        if (header != null) {
342            encoder.writeObject(header);
343        }
344        if (daMap != null) {
345            encoder.writeObject(new DeliveryAnnotations(daMap));
346        }
347        if (maMap != null) {
348            encoder.writeObject(new MessageAnnotations(maMap));
349        }
350        if (properties != null) {
351            encoder.writeObject(properties);
352        }
353        if (apMap != null) {
354            encoder.writeObject(new ApplicationProperties(apMap));
355        }
356        if (body != null) {
357            encoder.writeObject(body);
358        }
359        if (footerMap != null) {
360            encoder.writeObject(new Footer(footerMap));
361        }
362
363        return new EncodedMessage(messageFormat, buffer.getArray(), 0, buffer.getArrayLength());
364    }
365
366    private Section convertBody(ActiveMQMessage message) throws JMSException {
367
368        Section body = null;
369        short orignalEncoding = AMQP_UNKNOWN;
370
371        try {
372            orignalEncoding = message.getShortProperty(JMS_AMQP_ORIGINAL_ENCODING);
373        } catch (Exception ex) {
374            // Ignore and stick with UNKNOWN
375        }
376
377        int messageType = message.getDataStructureType();
378
379        if (messageType == CommandTypes.ACTIVEMQ_BYTES_MESSAGE) {
380            Binary payload = getBinaryFromMessageBody((ActiveMQBytesMessage) message);
381
382            if (payload == null) {
383                payload = EMPTY_BINARY;
384            }
385
386            switch (orignalEncoding) {
387                case AMQP_NULL:
388                    break;
389                case AMQP_VALUE_BINARY:
390                    body = new AmqpValue(payload);
391                    break;
392                case AMQP_DATA:
393                case AMQP_UNKNOWN:
394                default:
395                    body = new Data(payload);
396                    break;
397            }
398        } else if (messageType == CommandTypes.ACTIVEMQ_TEXT_MESSAGE) {
399            switch (orignalEncoding) {
400                case AMQP_NULL:
401                    break;
402                case AMQP_DATA:
403                    body = new Data(getBinaryFromMessageBody((ActiveMQTextMessage) message));
404                    break;
405                case AMQP_VALUE_STRING:
406                case AMQP_UNKNOWN:
407                default:
408                    body = new AmqpValue(((TextMessage) message).getText());
409                    break;
410            }
411        } else if (messageType == CommandTypes.ACTIVEMQ_MAP_MESSAGE) {
412            body = new AmqpValue(getMapFromMessageBody((ActiveMQMapMessage) message));
413        } else if (messageType == CommandTypes.ACTIVEMQ_STREAM_MESSAGE) {
414            ArrayList<Object> list = new ArrayList<>();
415            final ActiveMQStreamMessage m = (ActiveMQStreamMessage) message;
416            try {
417                while (true) {
418                    list.add(m.readObject());
419                }
420            } catch (MessageEOFException e) {
421            }
422
423            switch (orignalEncoding) {
424                case AMQP_SEQUENCE:
425                    body = new AmqpSequence(list);
426                    break;
427                case AMQP_VALUE_LIST:
428                case AMQP_UNKNOWN:
429                default:
430                    body = new AmqpValue(list);
431                    break;
432            }
433        } else if (messageType == CommandTypes.ACTIVEMQ_OBJECT_MESSAGE) {
434            Binary payload = getBinaryFromMessageBody((ActiveMQObjectMessage) message);
435
436            if (payload == null) {
437                payload = EMPTY_BINARY;
438            }
439
440            switch (orignalEncoding) {
441                case AMQP_VALUE_BINARY:
442                    body = new AmqpValue(payload);
443                    break;
444                case AMQP_DATA:
445                case AMQP_UNKNOWN:
446                default:
447                    body = new Data(payload);
448                    break;
449            }
450
451            // For a non-AMQP message we tag the outbound content type as containing
452            // a serialized Java object so that an AMQP client has a hint as to what
453            // we are sending it.
454            if (!message.propertyExists(JMS_AMQP_CONTENT_TYPE)) {
455                message.setReadOnlyProperties(false);
456                message.setStringProperty(JMS_AMQP_CONTENT_TYPE, SERIALIZED_JAVA_OBJECT_CONTENT_TYPE);
457                message.setReadOnlyProperties(true);
458            }
459        }
460
461        return body;
462    }
463
464    private static byte destinationType(ActiveMQDestination destination) {
465        if (destination.isQueue()) {
466            if (destination.isTemporary()) {
467                return TEMP_QUEUE_TYPE;
468            } else {
469                return QUEUE_TYPE;
470            }
471        } else if (destination.isTopic()) {
472            if (destination.isTemporary()) {
473                return TEMP_TOPIC_TYPE;
474            } else {
475                return TOPIC_TYPE;
476            }
477        }
478
479        throw new IllegalArgumentException("Unknown Destination Type passed to JMS Transformer.");
480    }
481
482    private static Object getOriginalMessageId(ActiveMQMessage message) {
483        Object result;
484        MessageId messageId = message.getMessageId();
485        if (messageId.getTextView() != null) {
486            try {
487                result = AMQPMessageIdHelper.INSTANCE.toIdObject(messageId.getTextView());
488            } catch (AmqpProtocolException e) {
489                result = messageId.getTextView();
490            }
491        } else {
492            result = messageId.toString();
493        }
494
495        return result;
496    }
497}