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; 018 019import java.nio.ByteBuffer; 020import java.util.AbstractMap; 021import java.util.Map; 022 023import org.apache.activemq.command.ActiveMQDestination; 024import org.apache.qpid.proton.amqp.Binary; 025import org.apache.qpid.proton.amqp.DescribedType; 026import org.apache.qpid.proton.amqp.Symbol; 027import org.apache.qpid.proton.amqp.UnsignedLong; 028import org.apache.qpid.proton.amqp.transaction.Coordinator; 029import org.fusesource.hawtbuf.Buffer; 030 031/** 032 * Set of useful methods and definitions used in the AMQP protocol handling 033 */ 034public class AmqpSupport { 035 036 // Identification values used to locating JMS selector types. 037 public static final UnsignedLong JMS_SELECTOR_CODE = UnsignedLong.valueOf(0x0000468C00000004L); 038 public static final Symbol JMS_SELECTOR_NAME = Symbol.valueOf("apache.org:selector-filter:string"); 039 public static final Object[] JMS_SELECTOR_FILTER_IDS = new Object[] { JMS_SELECTOR_CODE, JMS_SELECTOR_NAME }; 040 public static final UnsignedLong NO_LOCAL_CODE = UnsignedLong.valueOf(0x0000468C00000003L); 041 public static final Symbol NO_LOCAL_NAME = Symbol.valueOf("apache.org:no-local-filter:list"); 042 public static final Object[] NO_LOCAL_FILTER_IDS = new Object[] { NO_LOCAL_CODE, NO_LOCAL_NAME }; 043 044 // Capabilities used to identify destination type in some requests. 045 public static final Symbol TEMP_QUEUE_CAPABILITY = Symbol.valueOf("temporary-queue"); 046 public static final Symbol TEMP_TOPIC_CAPABILITY = Symbol.valueOf("temporary-topic"); 047 048 // Symbols used to announce connection information to remote peer. 049 public static final Symbol INVALID_FIELD = Symbol.valueOf("invalid-field"); 050 public static final Symbol CONTAINER_ID = Symbol.valueOf("container-id"); 051 052 // Symbols used to announce connection information to remote peer. 053 public static final Symbol ANONYMOUS_RELAY = Symbol.valueOf("ANONYMOUS-RELAY"); 054 public static final Symbol QUEUE_PREFIX = Symbol.valueOf("queue-prefix"); 055 public static final Symbol TOPIC_PREFIX = Symbol.valueOf("topic-prefix"); 056 public static final Symbol CONNECTION_OPEN_FAILED = Symbol.valueOf("amqp:connection-establishment-failed"); 057 public static final Symbol PRODUCT = Symbol.valueOf("product"); 058 public static final Symbol VERSION = Symbol.valueOf("version"); 059 public static final Symbol PLATFORM = Symbol.valueOf("platform"); 060 061 // Symbols used in configuration of newly opened links. 062 public static final Symbol COPY = Symbol.getSymbol("copy"); 063 064 // Lifetime policy symbols 065 public static final Symbol LIFETIME_POLICY = Symbol.valueOf("lifetime-policy"); 066 067 /** 068 * Search for a given Symbol in a given array of Symbol object. 069 * 070 * @param symbols 071 * the set of Symbols to search. 072 * @param key 073 * the value to try and find in the Symbol array. 074 * 075 * @return true if the key is found in the given Symbol array. 076 */ 077 public static boolean contains(Symbol[] symbols, Symbol key) { 078 if (symbols == null || symbols.length == 0) { 079 return false; 080 } 081 082 for (Symbol symbol : symbols) { 083 if (symbol.equals(key)) { 084 return true; 085 } 086 } 087 088 return false; 089 } 090 091 /** 092 * Search for a particular filter using a set of known indentification values 093 * in the Map of filters. 094 * 095 * @param filters 096 * The filters map that should be searched. 097 * @param filterIds 098 * The aliases for the target filter to be located. 099 * 100 * @return the filter if found in the mapping or null if not found. 101 */ 102 public static Map.Entry<Symbol, DescribedType> findFilter(Map<Symbol, Object> filters, Object[] filterIds) { 103 104 if (filterIds == null || filterIds.length == 0) { 105 throw new IllegalArgumentException("Invalid empty Filter Ids array passed: "); 106 } 107 108 if (filters == null || filters.isEmpty()) { 109 return null; 110 } 111 112 for (Map.Entry<Symbol, Object> filter : filters.entrySet()) { 113 if (filter.getValue() instanceof DescribedType) { 114 DescribedType describedType = ((DescribedType) filter.getValue()); 115 Object descriptor = describedType.getDescriptor(); 116 117 for (Object filterId : filterIds) { 118 if (descriptor.equals(filterId)) { 119 return new AbstractMap.SimpleImmutableEntry<Symbol, DescribedType>(filter.getKey(), describedType); 120 } 121 } 122 } 123 } 124 125 return null; 126 } 127 128 /** 129 * Conversion from Java ByteBuffer to a HawtBuf buffer. 130 * 131 * @param data 132 * the ByteBuffer instance to convert. 133 * 134 * @return a new HawtBuf buffer converted from the given ByteBuffer. 135 */ 136 public static Buffer toBuffer(ByteBuffer data) { 137 if (data == null) { 138 return null; 139 } 140 141 Buffer rc; 142 143 if (data.isDirect()) { 144 rc = new Buffer(data.remaining()); 145 data.get(rc.data); 146 } else { 147 rc = new Buffer(data); 148 data.position(data.position() + data.remaining()); 149 } 150 151 return rc; 152 } 153 154 /** 155 * Given a long value, convert it to a byte array for marshalling. 156 * 157 * @param value 158 * the value to convert. 159 * 160 * @return a new byte array that holds the big endian value of the long. 161 */ 162 public static byte[] toBytes(long value) { 163 Buffer buffer = new Buffer(8); 164 buffer.bigEndianEditor().writeLong(value); 165 return buffer.data; 166 } 167 168 /** 169 * Converts a Binary value to a long assuming that the contained value is 170 * stored in Big Endian encoding. 171 * 172 * @param value 173 * the Binary object whose payload is converted to a long. 174 * 175 * @return a long value constructed from the bytes of the Binary instance. 176 */ 177 public static long toLong(Binary value) { 178 Buffer buffer = new Buffer(value.getArray(), value.getArrayOffset(), value.getLength()); 179 return buffer.bigEndianEditor().readLong(); 180 } 181 182 /** 183 * Given an AMQP endpoint, deduce the appropriate ActiveMQDestination type and create 184 * a new instance. By default if the endpoint address does not carry the standard prefix 185 * value then we default to a Queue type destination. If the endpoint is null or is an 186 * AMQP Coordinator type endpoint this method returns null to indicate no destination 187 * can be mapped. 188 * 189 * @param endpoint 190 * the AMQP endpoint to construct an ActiveMQDestination from. 191 * 192 * @return a new ActiveMQDestination that best matches the address of the given endpoint 193 * 194 * @throws AmqpProtocolException if an error occurs while deducing the destination type. 195 */ 196 public static ActiveMQDestination createDestination(Object endpoint) throws AmqpProtocolException { 197 if (endpoint == null) { 198 return null; 199 } else if (endpoint instanceof Coordinator) { 200 return null; 201 } else if (endpoint instanceof org.apache.qpid.proton.amqp.messaging.Terminus) { 202 org.apache.qpid.proton.amqp.messaging.Terminus terminus = (org.apache.qpid.proton.amqp.messaging.Terminus) endpoint; 203 if (terminus.getAddress() == null || terminus.getAddress().length() == 0) { 204 if (terminus instanceof org.apache.qpid.proton.amqp.messaging.Source) { 205 throw new AmqpProtocolException("amqp:invalid-field", "source address not set"); 206 } else { 207 throw new AmqpProtocolException("amqp:invalid-field", "target address not set"); 208 } 209 } 210 211 return ActiveMQDestination.createDestination(terminus.getAddress(), ActiveMQDestination.QUEUE_TYPE); 212 } else { 213 throw new RuntimeException("Unexpected terminus type: " + endpoint); 214 } 215 } 216}