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