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}