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:selector-filter:string"); 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 058 // Symbols used in configuration of newly opened links. 059 public static final Symbol COPY = Symbol.getSymbol("copy"); 060 061 // Lifetime policy symbols 062 public static final Symbol LIFETIME_POLICY = Symbol.valueOf("lifetime-policy"); 063 064 /** 065 * Search for a given Symbol in a given array of Symbol object. 066 * 067 * @param symbols 068 * the set of Symbols to search. 069 * @param key 070 * the value to try and find in the Symbol array. 071 * 072 * @return true if the key is found in the given Symbol array. 073 */ 074 public static boolean contains(Symbol[] symbols, Symbol key) { 075 if (symbols == null || symbols.length == 0) { 076 return false; 077 } 078 079 for (Symbol symbol : symbols) { 080 if (symbol.equals(key)) { 081 return true; 082 } 083 } 084 085 return false; 086 } 087 088 /** 089 * Search for a particular filter using a set of known indentification values 090 * in the Map of filters. 091 * 092 * @param filters 093 * The filters map that should be searched. 094 * @param filterIds 095 * The aliases for the target filter to be located. 096 * 097 * @return the filter if found in the mapping or null if not found. 098 */ 099 public static Map.Entry<Symbol, DescribedType> findFilter(Map<Symbol, Object> filters, Object[] filterIds) { 100 101 if (filterIds == null || filterIds.length == 0) { 102 throw new IllegalArgumentException("Invalid Filter Ids array passed: " + filterIds); 103 } 104 105 if (filters == null || filters.isEmpty()) { 106 return null; 107 } 108 109 for (Map.Entry<Symbol, Object> filter : filters.entrySet()) { 110 if (filter.getValue() instanceof DescribedType) { 111 DescribedType describedType = ((DescribedType) filter.getValue()); 112 Object descriptor = describedType.getDescriptor(); 113 114 for (Object filterId : filterIds) { 115 if (descriptor.equals(filterId)) { 116 return new AbstractMap.SimpleImmutableEntry<Symbol, DescribedType>(filter.getKey(), describedType); 117 } 118 } 119 } 120 } 121 122 return null; 123 } 124 125 /** 126 * Conversion from Java ByteBuffer to a HawtBuf buffer. 127 * 128 * @param data 129 * the ByteBuffer instance to convert. 130 * 131 * @return a new HawtBuf buffer converted from the given ByteBuffer. 132 */ 133 public static Buffer toBuffer(ByteBuffer data) { 134 if (data == null) { 135 return null; 136 } 137 138 Buffer rc; 139 140 if (data.isDirect()) { 141 rc = new Buffer(data.remaining()); 142 data.get(rc.data); 143 } else { 144 rc = new Buffer(data); 145 data.position(data.position() + data.remaining()); 146 } 147 148 return rc; 149 } 150 151 /** 152 * Given a long value, convert it to a byte array for marshalling. 153 * 154 * @param value 155 * the value to convert. 156 * 157 * @return a new byte array that holds the big endian value of the long. 158 */ 159 public static byte[] toBytes(long value) { 160 Buffer buffer = new Buffer(8); 161 buffer.bigEndianEditor().writeLong(value); 162 return buffer.data; 163 } 164 165 /** 166 * Converts a Binary value to a long assuming that the contained value is 167 * stored in Big Endian encoding. 168 * 169 * @param value 170 * the Binary object whose payload is converted to a long. 171 * 172 * @return a long value constructed from the bytes of the Binary instance. 173 */ 174 public static long toLong(Binary value) { 175 Buffer buffer = new Buffer(value.getArray(), value.getArrayOffset(), value.getLength()); 176 return buffer.bigEndianEditor().readLong(); 177 } 178 179 /** 180 * Given an AMQP endpoint, deduce the appropriate ActiveMQDestination type and create 181 * a new instance. By default if the endpoint address does not carry the standard prefix 182 * value then we default to a Queue type destination. If the endpoint is null or is an 183 * AMQP Coordinator type endpoint this method returns null to indicate no destination 184 * can be mapped. 185 * 186 * @param endpoint 187 * the AMQP endpoint to construct an ActiveMQDestination from. 188 * 189 * @return a new ActiveMQDestination that best matches the address of the given endpoint 190 * 191 * @throws AmqpProtocolException if an error occurs while deducing the destination type. 192 */ 193 public static ActiveMQDestination createDestination(Object endpoint) throws AmqpProtocolException { 194 if (endpoint == null) { 195 return null; 196 } else if (endpoint instanceof Coordinator) { 197 return null; 198 } else if (endpoint instanceof org.apache.qpid.proton.amqp.messaging.Terminus) { 199 org.apache.qpid.proton.amqp.messaging.Terminus terminus = (org.apache.qpid.proton.amqp.messaging.Terminus) endpoint; 200 if (terminus.getAddress() == null || terminus.getAddress().length() == 0) { 201 if (terminus instanceof org.apache.qpid.proton.amqp.messaging.Source) { 202 throw new AmqpProtocolException("amqp:invalid-field", "source address not set"); 203 } else { 204 throw new AmqpProtocolException("amqp:invalid-field", "target address not set"); 205 } 206 } 207 208 return ActiveMQDestination.createDestination(terminus.getAddress(), ActiveMQDestination.QUEUE_TYPE); 209 } else { 210 throw new RuntimeException("Unexpected terminus type: " + endpoint); 211 } 212 } 213}