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}