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; 018 019import java.io.IOException; 020import java.net.MalformedURLException; 021import java.net.URI; 022import java.net.URISyntaxException; 023import java.net.UnknownHostException; 024import java.util.HashMap; 025import java.util.Map; 026import java.util.concurrent.ConcurrentHashMap; 027import java.util.concurrent.ConcurrentMap; 028import java.util.concurrent.Executor; 029 030import org.apache.activemq.util.FactoryFinder; 031import org.apache.activemq.util.IOExceptionSupport; 032import org.apache.activemq.util.IntrospectionSupport; 033import org.apache.activemq.util.URISupport; 034import org.apache.activemq.wireformat.WireFormat; 035import org.apache.activemq.wireformat.WireFormatFactory; 036 037public abstract class TransportFactory { 038 039 private static final FactoryFinder TRANSPORT_FACTORY_FINDER = new FactoryFinder("META-INF/services/org/apache/activemq/transport/"); 040 private static final FactoryFinder WIREFORMAT_FACTORY_FINDER = new FactoryFinder("META-INF/services/org/apache/activemq/wireformat/"); 041 private static final ConcurrentMap<String, TransportFactory> TRANSPORT_FACTORYS = new ConcurrentHashMap<String, TransportFactory>(); 042 043 private static final String WRITE_TIMEOUT_FILTER = "soWriteTimeout"; 044 private static final String THREAD_NAME_FILTER = "threadName"; 045 046 public abstract TransportServer doBind(URI location) throws IOException; 047 048 public Transport doConnect(URI location, Executor ex) throws Exception { 049 return doConnect(location); 050 } 051 052 public Transport doCompositeConnect(URI location, Executor ex) throws Exception { 053 return doCompositeConnect(location); 054 } 055 056 /** 057 * Creates a normal transport. 058 * 059 * @param location 060 * @return the transport 061 * @throws Exception 062 */ 063 public static Transport connect(URI location) throws Exception { 064 TransportFactory tf = findTransportFactory(location); 065 return tf.doConnect(location); 066 } 067 068 /** 069 * Creates a normal transport. 070 * 071 * @param location 072 * @param ex 073 * @return the transport 074 * @throws Exception 075 */ 076 public static Transport connect(URI location, Executor ex) throws Exception { 077 TransportFactory tf = findTransportFactory(location); 078 return tf.doConnect(location, ex); 079 } 080 081 /** 082 * Creates a slimmed down transport that is more efficient so that it can be 083 * used by composite transports like reliable and HA. 084 * 085 * @param location 086 * @return the Transport 087 * @throws Exception 088 */ 089 public static Transport compositeConnect(URI location) throws Exception { 090 TransportFactory tf = findTransportFactory(location); 091 return tf.doCompositeConnect(location); 092 } 093 094 /** 095 * Creates a slimmed down transport that is more efficient so that it can be 096 * used by composite transports like reliable and HA. 097 * 098 * @param location 099 * @param ex 100 * @return the Transport 101 * @throws Exception 102 */ 103 public static Transport compositeConnect(URI location, Executor ex) throws Exception { 104 TransportFactory tf = findTransportFactory(location); 105 return tf.doCompositeConnect(location, ex); 106 } 107 108 public static TransportServer bind(URI location) throws IOException { 109 TransportFactory tf = findTransportFactory(location); 110 return tf.doBind(location); 111 } 112 113 public Transport doConnect(URI location) throws Exception { 114 try { 115 Map<String, String> options = new HashMap<String, String>(URISupport.parseParameters(location)); 116 if( !options.containsKey("wireFormat.host") ) { 117 options.put("wireFormat.host", location.getHost()); 118 } 119 WireFormat wf = createWireFormat(options); 120 Transport transport = createTransport(location, wf); 121 Transport rc = configure(transport, wf, options); 122 //remove auto 123 IntrospectionSupport.extractProperties(options, "auto."); 124 125 if (!options.isEmpty()) { 126 throw new IllegalArgumentException("Invalid connect parameters: " + options); 127 } 128 return rc; 129 } catch (URISyntaxException e) { 130 throw IOExceptionSupport.create(e); 131 } 132 } 133 134 public Transport doCompositeConnect(URI location) throws Exception { 135 try { 136 Map<String, String> options = new HashMap<String, String>(URISupport.parseParameters(location)); 137 WireFormat wf = createWireFormat(options); 138 Transport transport = createTransport(location, wf); 139 Transport rc = compositeConfigure(transport, wf, options); 140 if (!options.isEmpty()) { 141 throw new IllegalArgumentException("Invalid connect parameters: " + options); 142 } 143 return rc; 144 } catch (URISyntaxException e) { 145 throw IOExceptionSupport.create(e); 146 } 147 } 148 149 /** 150 * Allow registration of a transport factory without wiring via META-INF classes 151 * @param scheme 152 * @param tf 153 */ 154 public static void registerTransportFactory(String scheme, TransportFactory tf) { 155 TRANSPORT_FACTORYS.put(scheme, tf); 156 } 157 158 /** 159 * Factory method to create a new transport 160 * 161 * @throws IOException 162 * @throws UnknownHostException 163 */ 164 protected Transport createTransport(URI location, WireFormat wf) throws MalformedURLException, UnknownHostException, IOException { 165 throw new IOException("createTransport() method not implemented!"); 166 } 167 168 /** 169 * @param location 170 * @return 171 * @throws IOException 172 */ 173 public static TransportFactory findTransportFactory(URI location) throws IOException { 174 String scheme = location.getScheme(); 175 if (scheme == null) { 176 throw new IOException("Transport not scheme specified: [" + location + "]"); 177 } 178 TransportFactory tf = TRANSPORT_FACTORYS.get(scheme); 179 if (tf == null) { 180 // Try to load if from a META-INF property. 181 try { 182 tf = (TransportFactory)TRANSPORT_FACTORY_FINDER.newInstance(scheme); 183 TRANSPORT_FACTORYS.put(scheme, tf); 184 } catch (Throwable e) { 185 throw IOExceptionSupport.create("Transport scheme NOT recognized: [" + scheme + "]", e); 186 } 187 } 188 return tf; 189 } 190 191 protected WireFormat createWireFormat(Map<String, String> options) throws IOException { 192 WireFormatFactory factory = createWireFormatFactory(options); 193 WireFormat format = factory.createWireFormat(); 194 return format; 195 } 196 197 protected WireFormatFactory createWireFormatFactory(Map<String, String> options) throws IOException { 198 String wireFormat = options.remove("wireFormat"); 199 if (wireFormat == null) { 200 wireFormat = getDefaultWireFormatType(); 201 } 202 203 try { 204 WireFormatFactory wff = (WireFormatFactory)WIREFORMAT_FACTORY_FINDER.newInstance(wireFormat); 205 IntrospectionSupport.setProperties(wff, options, "wireFormat."); 206 return wff; 207 } catch (Throwable e) { 208 throw IOExceptionSupport.create("Could not create wire format factory for: " + wireFormat + ", reason: " + e, e); 209 } 210 } 211 212 protected String getDefaultWireFormatType() { 213 return "default"; 214 } 215 216 /** 217 * Fully configures and adds all need transport filters so that the 218 * transport can be used by the JMS client. 219 * 220 * @param transport 221 * @param wf 222 * @param options 223 * @return 224 * @throws Exception 225 */ 226 @SuppressWarnings("rawtypes") 227 public Transport configure(Transport transport, WireFormat wf, Map options) throws Exception { 228 transport = compositeConfigure(transport, wf, options); 229 230 transport = new MutexTransport(transport); 231 transport = new ResponseCorrelator(transport); 232 233 return transport; 234 } 235 236 /** 237 * Fully configures and adds all need transport filters so that the 238 * transport can be used by the ActiveMQ message broker. The main difference 239 * between this and the configure() method is that the broker does not issue 240 * requests to the client so the ResponseCorrelator is not needed. 241 * 242 * @param transport 243 * @param format 244 * @param options 245 * @return 246 * @throws Exception 247 */ 248 @SuppressWarnings("rawtypes") 249 public Transport serverConfigure(Transport transport, WireFormat format, HashMap options) throws Exception { 250 if (options.containsKey(THREAD_NAME_FILTER)) { 251 transport = new ThreadNameFilter(transport); 252 } 253 transport = compositeConfigure(transport, format, options); 254 transport = new MutexTransport(transport); 255 return transport; 256 } 257 258 /** 259 * Similar to configure(...) but this avoid adding in the MutexTransport and 260 * ResponseCorrelator transport layers so that the resulting transport can 261 * more efficiently be used as part of a composite transport. 262 * 263 * @param transport 264 * @param format 265 * @param options 266 * @return 267 */ 268 @SuppressWarnings("rawtypes") 269 public Transport compositeConfigure(Transport transport, WireFormat format, Map options) { 270 if (options.containsKey(WRITE_TIMEOUT_FILTER)) { 271 transport = new WriteTimeoutFilter(transport); 272 String soWriteTimeout = (String)options.remove(WRITE_TIMEOUT_FILTER); 273 if (soWriteTimeout!=null) { 274 ((WriteTimeoutFilter)transport).setWriteTimeout(Long.parseLong(soWriteTimeout)); 275 } 276 } 277 IntrospectionSupport.setProperties(transport, options); 278 return transport; 279 } 280 281 @SuppressWarnings("rawtypes") 282 protected String getOption(Map options, String key, String def) { 283 String rc = (String) options.remove(key); 284 if( rc == null ) { 285 rc = def; 286 } 287 return rc; 288 } 289}