| /* |
| * |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, |
| * software distributed under the License is distributed on an |
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| * KIND, either express or implied. See the License for the |
| * specific language governing permissions and limitations |
| * under the License. |
| * |
| */ |
| package org.apache.qpid.server.protocol.converter.v0_8_v1_0; |
| |
| import java.net.URISyntaxException; |
| import java.nio.charset.StandardCharsets; |
| import java.util.Arrays; |
| import java.util.Date; |
| import java.util.LinkedHashMap; |
| import java.util.Map; |
| import java.util.UUID; |
| |
| import org.apache.qpid.server.plugin.PluggableService; |
| import org.apache.qpid.server.protocol.converter.MessageConversionException; |
| import org.apache.qpid.server.protocol.v0_8.AMQMessage; |
| import org.apache.qpid.server.protocol.v0_8.AMQShortString; |
| import org.apache.qpid.server.protocol.v0_8.transport.BasicContentHeaderProperties; |
| import org.apache.qpid.server.protocol.v0_8.transport.MessagePublishInfo; |
| import org.apache.qpid.server.protocol.v1_0.MessageConverter_to_1_0; |
| import org.apache.qpid.server.protocol.v1_0.MessageMetaData_1_0; |
| import org.apache.qpid.server.protocol.v1_0.messaging.SectionEncoder; |
| import org.apache.qpid.server.protocol.v1_0.type.Binary; |
| import org.apache.qpid.server.protocol.v1_0.type.Symbol; |
| import org.apache.qpid.server.protocol.v1_0.type.UnsignedByte; |
| import org.apache.qpid.server.protocol.v1_0.type.UnsignedInteger; |
| import org.apache.qpid.server.protocol.v1_0.type.messaging.ApplicationProperties; |
| import org.apache.qpid.server.protocol.v1_0.type.messaging.DataSection; |
| import org.apache.qpid.server.protocol.v1_0.type.messaging.EncodingRetainingSection; |
| import org.apache.qpid.server.protocol.v1_0.type.messaging.Header; |
| import org.apache.qpid.server.protocol.v1_0.type.messaging.MessageAnnotations; |
| import org.apache.qpid.server.protocol.v1_0.type.messaging.Properties; |
| import org.apache.qpid.server.url.AMQBindingURL; |
| import org.apache.qpid.server.util.GZIPUtils; |
| |
| @PluggableService |
| public class MessageConverter_0_8_to_1_0 extends MessageConverter_to_1_0<AMQMessage> |
| { |
| @Override |
| public Class<AMQMessage> getInputClass() |
| { |
| return AMQMessage.class; |
| } |
| |
| @Override |
| protected MessageMetaData_1_0 convertMetaData(final AMQMessage serverMessage, |
| final EncodingRetainingSection<?> bodySection, |
| SectionEncoder sectionEncoder) |
| { |
| Header header = new Header(); |
| Properties props = new Properties(); |
| header.setDurable(serverMessage.isPersistent()); |
| |
| BasicContentHeaderProperties contentHeader = serverMessage.getContentHeaderBody().getProperties(); |
| |
| header.setPriority(UnsignedByte.valueOf(contentHeader.getPriority())); |
| |
| if(contentHeader.hasExpiration()) |
| { |
| final long expiration = serverMessage.getExpiration(); |
| final long arrivalTime = serverMessage.getArrivalTime(); |
| header.setTtl(UnsignedInteger.valueOf(Math.max(0, expiration - arrivalTime))); |
| } |
| |
| if(!GZIPUtils.GZIP_CONTENT_ENCODING.equals(contentHeader.getEncodingAsString()) && bodySection instanceof DataSection) |
| { |
| props.setContentEncoding(Symbol.valueOf(contentHeader.getEncodingAsString())); |
| } |
| |
| Symbol contentType = getContentType(contentHeader.getContentTypeAsString()); |
| props.setContentType(contentType); |
| |
| final AMQShortString correlationId = contentHeader.getCorrelationId(); |
| if(correlationId != null) |
| { |
| final byte[] correlationIdAsBytes = correlationId.getBytes(); |
| final String correlationIdAsString = contentHeader.getCorrelationIdAsString(); |
| if (Arrays.equals(correlationIdAsBytes, correlationIdAsString.getBytes(StandardCharsets.UTF_8))) |
| { |
| props.setCorrelationId(convertMessageId(correlationId)); |
| } |
| else |
| { |
| props.setCorrelationId(correlationIdAsBytes); |
| } |
| } |
| |
| final AMQShortString messageId = contentHeader.getMessageId(); |
| if(messageId != null) |
| { |
| props.setMessageId(convertMessageId(messageId)); |
| } |
| |
| if (contentHeader.getReplyTo() != null) |
| { |
| props.setReplyTo(convertReplyTo(contentHeader.getReplyTo())); |
| } |
| |
| if(contentHeader.getUserId() != null) |
| { |
| props.setUserId(new Binary(contentHeader.getUserId().getBytes())); |
| } |
| |
| if (contentHeader.hasTimestamp()) |
| { |
| props.setCreationTime(new Date(contentHeader.getTimestamp())); |
| } |
| else |
| { |
| props.setCreationTime(new Date(serverMessage.getArrivalTime())); |
| } |
| |
| if (contentHeader.getType() != null) |
| { |
| props.setSubject(contentHeader.getType().toString()); |
| } |
| |
| Map<String, Object> applicationPropertiesMap = new LinkedHashMap<>(contentHeader.getHeadersAsMap()); |
| |
| if(applicationPropertiesMap.containsKey("qpid.subject")) |
| { |
| props.setSubject(String.valueOf(applicationPropertiesMap.get("qpid.subject"))); |
| applicationPropertiesMap.remove("qpid.subject"); |
| } |
| |
| if(applicationPropertiesMap.containsKey("JMSXGroupID")) |
| { |
| props.setGroupId(String.valueOf(applicationPropertiesMap.get("JMSXGroupID"))); |
| applicationPropertiesMap.remove("JMSXGroupID"); |
| } |
| |
| if(applicationPropertiesMap.containsKey("JMSXGroupSeq")) |
| { |
| Object jmsxGroupSeq = applicationPropertiesMap.get("JMSXGroupSeq"); |
| if (jmsxGroupSeq instanceof Integer) |
| { |
| props.setGroupSequence(UnsignedInteger.valueOf((Integer)jmsxGroupSeq)); |
| applicationPropertiesMap.remove("JMSXGroupSeq"); |
| } |
| } |
| |
| MessagePublishInfo messagePublishInfo = serverMessage.getMessagePublishInfo(); |
| String to = AMQShortString.toString(messagePublishInfo.getExchange()); |
| if (messagePublishInfo.getRoutingKey() != null) |
| { |
| String routingKey = AMQShortString.toString(messagePublishInfo.getRoutingKey()); |
| if (to != null && !"".equals(to)) |
| { |
| to += "/" + routingKey; |
| } |
| else |
| { |
| to = routingKey; |
| } |
| } |
| props.setTo(to); |
| |
| |
| final ApplicationProperties applicationProperties; |
| try |
| { |
| applicationProperties = new ApplicationProperties(applicationPropertiesMap); |
| } |
| catch (IllegalArgumentException e) |
| { |
| throw new MessageConversionException("Could not convert message from 0-8 to 1.0 because headers conversion failed.", e); |
| } |
| |
| MessageAnnotations messageAnnotations = createMessageAnnotation(bodySection, |
| contentHeader.getContentTypeAsString()); |
| |
| return new MessageMetaData_1_0(header.createEncodingRetainingSection(), |
| null, |
| messageAnnotations == null ? null : messageAnnotations.createEncodingRetainingSection(), |
| props.createEncodingRetainingSection(), |
| applicationProperties.createEncodingRetainingSection(), |
| null, |
| serverMessage.getArrivalTime(), |
| bodySection.getEncodedSize()); |
| } |
| |
| private Object convertMessageId(final AMQShortString messageId) |
| { |
| try |
| { |
| String messageIdAsString = messageId.toString(); |
| if(messageIdAsString.startsWith("ID:")) |
| { |
| messageIdAsString = messageIdAsString.substring(3); |
| } |
| return UUID.fromString(messageIdAsString); |
| } |
| catch (IllegalArgumentException e) |
| { |
| return messageId.toString(); |
| } |
| } |
| |
| private String convertReplyTo(final AMQShortString replyTo) |
| { |
| String convertedReplyTo; |
| final String originalReplyTo = String.valueOf(replyTo); |
| try |
| { |
| AMQBindingURL burl = new AMQBindingURL(originalReplyTo); |
| |
| if (burl.getExchangeName() != null && !burl.getExchangeName().equals("")) |
| { |
| convertedReplyTo = burl.getExchangeName(); |
| |
| if (burl.getRoutingKey() != null) |
| { |
| convertedReplyTo += "/" + burl.getRoutingKey(); |
| } |
| } |
| else if (burl.getQueueName() != null && !burl.getQueueName().equals("")) |
| { |
| convertedReplyTo = burl.getQueueName(); |
| } |
| else if (burl.getRoutingKey() != null) |
| { |
| convertedReplyTo = burl.getRoutingKey(); |
| } |
| else |
| { |
| convertedReplyTo = originalReplyTo; |
| } |
| } |
| catch (URISyntaxException e) |
| { |
| convertedReplyTo = originalReplyTo; |
| } |
| return convertedReplyTo; |
| } |
| |
| @Override |
| public String getType() |
| { |
| return "v0-8 to v1-0"; |
| } |
| } |
