| /* |
| * |
| * 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_10_v1_0; |
| |
| import java.nio.ByteBuffer; |
| import java.nio.charset.CharacterCodingException; |
| import java.nio.charset.CharsetDecoder; |
| import java.nio.charset.CodingErrorAction; |
| import java.nio.charset.StandardCharsets; |
| import java.util.Date; |
| import java.util.LinkedHashMap; |
| import java.util.Map; |
| |
| import org.apache.qpid.server.plugin.PluggableService; |
| import org.apache.qpid.server.protocol.converter.MessageConversionException; |
| import org.apache.qpid.server.protocol.v0_10.MessageTransferMessage; |
| import org.apache.qpid.server.protocol.v0_10.transport.DeliveryProperties; |
| import org.apache.qpid.server.protocol.v0_10.transport.MessageDeliveryMode; |
| import org.apache.qpid.server.protocol.v0_10.transport.MessageProperties; |
| import org.apache.qpid.server.protocol.v0_10.transport.ReplyTo; |
| 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.util.GZIPUtils; |
| |
| @PluggableService |
| public class MessageConverter_0_10_to_1_0 extends MessageConverter_to_1_0<MessageTransferMessage> |
| { |
| @Override |
| public Class<MessageTransferMessage> getInputClass() |
| { |
| return MessageTransferMessage.class; |
| } |
| |
| |
| @Override |
| protected MessageMetaData_1_0 convertMetaData(MessageTransferMessage serverMessage, |
| final EncodingRetainingSection<?> bodySection, |
| SectionEncoder sectionEncoder) |
| { |
| Properties props = new Properties(); |
| props.setCreationTime(new Date(serverMessage.getArrivalTime())); |
| |
| final MessageProperties msgProps = serverMessage.getHeader().getMessageProperties(); |
| final DeliveryProperties deliveryProps = serverMessage.getHeader().getDeliveryProperties(); |
| Header header = new Header(); |
| if(deliveryProps != null) |
| { |
| header.setDurable(deliveryProps.hasDeliveryMode() |
| && deliveryProps.getDeliveryMode() == MessageDeliveryMode.PERSISTENT); |
| if (deliveryProps.hasPriority()) |
| { |
| header.setPriority(UnsignedByte.valueOf((byte) deliveryProps.getPriority().getValue())); |
| } |
| if (deliveryProps.hasTtl()) |
| { |
| header.setTtl(UnsignedInteger.valueOf(deliveryProps.getTtl())); |
| } |
| else if (deliveryProps.hasExpiration()) |
| { |
| long ttl = Math.max(0, deliveryProps.getExpiration() - serverMessage.getArrivalTime()); |
| header.setTtl(UnsignedInteger.valueOf(ttl)); |
| } |
| |
| if(deliveryProps.hasTimestamp()) |
| { |
| props.setCreationTime(new Date(deliveryProps.getTimestamp())); |
| } |
| |
| String to = deliveryProps.getExchange(); |
| if (deliveryProps.getRoutingKey() != null) |
| { |
| String routingKey = deliveryProps.getRoutingKey(); |
| if (to != null && !"".equals(to)) |
| { |
| to += "/" + routingKey; |
| } |
| else |
| { |
| to = routingKey; |
| } |
| } |
| props.setTo(to); |
| |
| } |
| |
| |
| ApplicationProperties applicationProperties = null; |
| String originalContentMimeType = null; |
| if(msgProps != null) |
| { |
| if(msgProps.hasContentEncoding() |
| && !GZIPUtils.GZIP_CONTENT_ENCODING.equals(msgProps.getContentEncoding()) |
| && bodySection instanceof DataSection) |
| { |
| props.setContentEncoding(Symbol.valueOf(msgProps.getContentEncoding())); |
| } |
| |
| if(msgProps.hasCorrelationId()) |
| { |
| CharsetDecoder charsetDecoder = StandardCharsets.UTF_8.newDecoder() |
| .onMalformedInput(CodingErrorAction.REPORT) |
| .onUnmappableCharacter(CodingErrorAction.REPORT); |
| try |
| { |
| String correlationIdAsString = charsetDecoder.decode(ByteBuffer.wrap(msgProps.getCorrelationId())).toString(); |
| props.setCorrelationId(correlationIdAsString); |
| } |
| catch (CharacterCodingException e) |
| { |
| props.setCorrelationId(new Binary(msgProps.getCorrelationId())); |
| } |
| } |
| |
| if(msgProps.hasMessageId()) |
| { |
| props.setMessageId(msgProps.getMessageId()); |
| } |
| if(msgProps.hasReplyTo()) |
| { |
| ReplyTo replyTo = msgProps.getReplyTo(); |
| String to = null; |
| if (replyTo.hasExchange() && !"".equals(replyTo.getExchange())) |
| { |
| to = replyTo.getExchange(); |
| } |
| if (replyTo.hasRoutingKey()) |
| { |
| if (to != null) |
| { |
| to += "/" + replyTo.getRoutingKey(); |
| } |
| else |
| { |
| to = replyTo.getRoutingKey(); |
| } |
| } |
| props.setReplyTo(to); |
| } |
| |
| if(msgProps.hasContentType()) |
| { |
| originalContentMimeType = msgProps.getContentType(); |
| final Symbol contentType = |
| MessageConverter_to_1_0.getContentType(originalContentMimeType); |
| props.setContentType(contentType); |
| } |
| |
| if(msgProps.hasUserId()) |
| { |
| props.setUserId(new Binary(msgProps.getUserId())); |
| } |
| |
| Map<String, Object> applicationPropertiesMap = msgProps.getApplicationHeaders(); |
| if(applicationPropertiesMap != null) |
| { |
| applicationPropertiesMap = new LinkedHashMap<>(applicationPropertiesMap); |
| if (applicationPropertiesMap.containsKey("x-jms-type")) |
| { |
| props.setSubject(String.valueOf(applicationPropertiesMap.get("x-jms-type"))); |
| applicationPropertiesMap.remove("x-jms-type"); |
| } |
| |
| 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"); |
| } |
| } |
| |
| try |
| { |
| applicationProperties = new ApplicationProperties(applicationPropertiesMap); |
| } |
| catch (IllegalArgumentException e) |
| { |
| throw new MessageConversionException("Could not convert message from 0-10 to 1.0 because application headers conversion failed.", e); |
| } |
| } |
| } |
| final MessageAnnotations messageAnnotation = |
| MessageConverter_to_1_0.createMessageAnnotation(bodySection, originalContentMimeType); |
| return new MessageMetaData_1_0(header.createEncodingRetainingSection(), |
| null, |
| messageAnnotation == null ? null : messageAnnotation.createEncodingRetainingSection(), |
| props.createEncodingRetainingSection(), |
| applicationProperties == null ? null : applicationProperties.createEncodingRetainingSection(), |
| null, |
| serverMessage.getArrivalTime(), |
| bodySection.getEncodedSize()); |
| } |
| |
| @Override |
| public String getType() |
| { |
| return "v0-10 to v1-0"; |
| } |
| } |
