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 if (!options.isEmpty()) { 123 throw new IllegalArgumentException("Invalid connect parameters: " + options); 124 } 125 return rc; 126 } catch (URISyntaxException e) { 127 throw IOExceptionSupport.create(e); 128 } 129 } 130 131 public Transport doCompositeConnect(URI location) throws Exception { 132 try { 133 Map<String, String> options = new HashMap<String, String>(URISupport.parseParameters(location)); 134 WireFormat wf = createWireFormat(options); 135 Transport transport = createTransport(location, wf); 136 Transport rc = compositeConfigure(transport, wf, options); 137 if (!options.isEmpty()) { 138 throw new IllegalArgumentException("Invalid connect parameters: " + options); 139 } 140 return rc; 141 } catch (URISyntaxException e) { 142 throw IOExceptionSupport.create(e); 143 } 144 } 145 146 /** 147 * Allow registration of a transport factory without wiring via META-INF classes 148 * @param scheme 149 * @param tf 150 */ 151 public static void registerTransportFactory(String scheme, TransportFactory tf) { 152 TRANSPORT_FACTORYS.put(scheme, tf); 153 } 154 155 /** 156 * Factory method to create a new transport 157 * 158 * @throws IOException 159 * @throws UnknownHostException 160 */ 161 protected Transport createTransport(URI location, WireFormat wf) throws MalformedURLException, UnknownHostException, IOException { 162 throw new IOException("createTransport() method not implemented!"); 163 } 164 165 /** 166 * @param location 167 * @return 168 * @throws IOException 169 */ 170 public static TransportFactory findTransportFactory(URI location) throws IOException { 171 String scheme = location.getScheme(); 172 if (scheme == null) { 173 throw new IOException("Transport not scheme specified: [" + location + "]"); 174 } 175 TransportFactory tf = TRANSPORT_FACTORYS.get(scheme); 176 if (tf == null) { 177 // Try to load if from a META-INF property. 178 try { 179 tf = (TransportFactory)TRANSPORT_FACTORY_FINDER.newInstance(scheme); 180 TRANSPORT_FACTORYS.put(scheme, tf); 181 } catch (Throwable e) { 182 throw IOExceptionSupport.create("Transport scheme NOT recognized: [" + scheme + "]", e); 183 } 184 } 185 return tf; 186 } 187 188 protected WireFormat createWireFormat(Map<String, String> options) throws IOException { 189 WireFormatFactory factory = createWireFormatFactory(options); 190 WireFormat format = factory.createWireFormat(); 191 return format; 192 } 193 194 protected WireFormatFactory createWireFormatFactory(Map<String, String> options) throws IOException { 195 String wireFormat = options.remove("wireFormat"); 196 if (wireFormat == null) { 197 wireFormat = getDefaultWireFormatType(); 198 } 199 200 try { 201 WireFormatFactory wff = (WireFormatFactory)WIREFORMAT_FACTORY_FINDER.newInstance(wireFormat); 202 IntrospectionSupport.setProperties(wff, options, "wireFormat."); 203 return wff; 204 } catch (Throwable e) { 205 throw IOExceptionSupport.create("Could not create wire format factory for: " + wireFormat + ", reason: " + e, e); 206 } 207 } 208 209 protected String getDefaultWireFormatType() { 210 return "default"; 211 } 212 213 /** 214 * Fully configures and adds all need transport filters so that the 215 * transport can be used by the JMS client. 216 * 217 * @param transport 218 * @param wf 219 * @param options 220 * @return 221 * @throws Exception 222 */ 223 @SuppressWarnings("rawtypes") 224 public Transport configure(Transport transport, WireFormat wf, Map options) throws Exception { 225 transport = compositeConfigure(transport, wf, options); 226 227 transport = new MutexTransport(transport); 228 transport = new ResponseCorrelator(transport); 229 230 return transport; 231 } 232 233 /** 234 * Fully configures and adds all need transport filters so that the 235 * transport can be used by the ActiveMQ message broker. The main difference 236 * between this and the configure() method is that the broker does not issue 237 * requests to the client so the ResponseCorrelator is not needed. 238 * 239 * @param transport 240 * @param format 241 * @param options 242 * @return 243 * @throws Exception 244 */ 245 @SuppressWarnings("rawtypes") 246 public Transport serverConfigure(Transport transport, WireFormat format, HashMap options) throws Exception { 247 if (options.containsKey(THREAD_NAME_FILTER)) { 248 transport = new ThreadNameFilter(transport); 249 } 250 transport = compositeConfigure(transport, format, options); 251 transport = new MutexTransport(transport); 252 return transport; 253 } 254 255 /** 256 * Similar to configure(...) but this avoid adding in the MutexTransport and 257 * ResponseCorrelator transport layers so that the resulting transport can 258 * more efficiently be used as part of a composite transport. 259 * 260 * @param transport 261 * @param format 262 * @param options 263 * @return 264 */ 265 @SuppressWarnings("rawtypes") 266 public Transport compositeConfigure(Transport transport, WireFormat format, Map options) { 267 if (options.containsKey(WRITE_TIMEOUT_FILTER)) { 268 transport = new WriteTimeoutFilter(transport); 269 String soWriteTimeout = (String)options.remove(WRITE_TIMEOUT_FILTER); 270 if (soWriteTimeout!=null) { 271 ((WriteTimeoutFilter)transport).setWriteTimeout(Long.parseLong(soWriteTimeout)); 272 } 273 } 274 IntrospectionSupport.setProperties(transport, options); 275 return transport; 276 } 277 278 @SuppressWarnings("rawtypes") 279 protected String getOption(Map options, String key, String def) { 280 String rc = (String) options.remove(key); 281 if( rc == null ) { 282 rc = def; 283 } 284 return rc; 285 } 286}