| /** |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF licenses this file to You under the Apache License, Version 2.0 |
| * (the "License"); you may not use this file except in compliance with |
| * the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| package org.apache.activemq.transport.vm; |
| |
| import java.io.IOException; |
| import java.net.URI; |
| import java.net.URISyntaxException; |
| import java.util.HashMap; |
| import java.util.Map; |
| import java.util.concurrent.ConcurrentHashMap; |
| |
| import org.apache.activemq.broker.BrokerFactory; |
| import org.apache.activemq.broker.BrokerFactoryHandler; |
| import org.apache.activemq.broker.BrokerRegistry; |
| import org.apache.activemq.broker.BrokerService; |
| import org.apache.activemq.broker.TransportConnector; |
| import org.apache.activemq.transport.MarshallingTransportFilter; |
| import org.apache.activemq.transport.Transport; |
| import org.apache.activemq.transport.TransportFactory; |
| import org.apache.activemq.transport.TransportServer; |
| import org.apache.activemq.util.IOExceptionSupport; |
| import org.apache.activemq.util.IntrospectionSupport; |
| import org.apache.activemq.util.ServiceSupport; |
| import org.apache.activemq.util.URISupport; |
| import org.apache.activemq.util.URISupport.CompositeData; |
| import org.slf4j.Logger; |
| import org.slf4j.LoggerFactory; |
| import org.slf4j.MDC; |
| |
| public class VMTransportFactory extends TransportFactory { |
| |
| public static final ConcurrentHashMap<String, BrokerService> BROKERS = new ConcurrentHashMap<String, BrokerService>(); |
| public static final ConcurrentHashMap<String, TransportConnector> CONNECTORS = new ConcurrentHashMap<String, TransportConnector>(); |
| public static final ConcurrentHashMap<String, VMTransportServer> SERVERS = new ConcurrentHashMap<String, VMTransportServer>(); |
| private static final Logger LOG = LoggerFactory.getLogger(VMTransportFactory.class); |
| |
| BrokerFactoryHandler brokerFactoryHandler; |
| |
| public Transport doConnect(URI location) throws Exception { |
| return VMTransportServer.configure(doCompositeConnect(location)); |
| } |
| |
| public Transport doCompositeConnect(URI location) throws Exception { |
| URI brokerURI; |
| String host; |
| Map<String, String> options; |
| boolean create = true; |
| int waitForStart = -1; |
| CompositeData data = URISupport.parseComposite(location); |
| if (data.getComponents().length == 1 && "broker".equals(data.getComponents()[0].getScheme())) { |
| brokerURI = data.getComponents()[0]; |
| CompositeData brokerData = URISupport.parseComposite(brokerURI); |
| host = (String)brokerData.getParameters().get("brokerName"); |
| if (host == null) { |
| host = "localhost"; |
| } |
| if (brokerData.getPath() != null) { |
| host = brokerData.getPath(); |
| } |
| options = data.getParameters(); |
| location = new URI("vm://" + host); |
| } else { |
| // If using the less complex vm://localhost?broker.persistent=true |
| // form |
| try { |
| host = extractHost(location); |
| options = URISupport.parseParameters(location); |
| String config = (String)options.remove("brokerConfig"); |
| if (config != null) { |
| brokerURI = new URI(config); |
| } else { |
| Map brokerOptions = IntrospectionSupport.extractProperties(options, "broker."); |
| brokerURI = new URI("broker://()/" + host + "?" |
| + URISupport.createQueryString(brokerOptions)); |
| } |
| if ("false".equals(options.remove("create"))) { |
| create = false; |
| } |
| String waitForStartString = options.remove("waitForStart"); |
| if (waitForStartString != null) { |
| waitForStart = Integer.parseInt(waitForStartString); |
| } |
| } catch (URISyntaxException e1) { |
| throw IOExceptionSupport.create(e1); |
| } |
| location = new URI("vm://" + host); |
| } |
| if (host == null) { |
| host = "localhost"; |
| } |
| VMTransportServer server = SERVERS.get(host); |
| // validate the broker is still active |
| if (!validateBroker(host) || server == null) { |
| BrokerService broker = null; |
| // Synchronize on the registry so that multiple concurrent threads |
| // doing this do not think that the broker has not been created and |
| // cause multiple brokers to be started. |
| synchronized (BrokerRegistry.getInstance().getRegistryMutext()) { |
| broker = lookupBroker(BrokerRegistry.getInstance(), host, waitForStart); |
| if (broker == null) { |
| if (!create) { |
| throw new IOException("Broker named '" + host + "' does not exist."); |
| } |
| try { |
| if (brokerFactoryHandler != null) { |
| broker = brokerFactoryHandler.createBroker(brokerURI); |
| } else { |
| broker = BrokerFactory.createBroker(brokerURI); |
| } |
| broker.start(); |
| MDC.put("activemq.broker", broker.getBrokerName()); |
| } catch (URISyntaxException e) { |
| throw IOExceptionSupport.create(e); |
| } |
| BROKERS.put(host, broker); |
| BrokerRegistry.getInstance().getRegistryMutext().notifyAll(); |
| } |
| |
| server = SERVERS.get(host); |
| if (server == null) { |
| server = (VMTransportServer)bind(location, true); |
| TransportConnector connector = new TransportConnector(server); |
| connector.setBrokerService(broker); |
| connector.setUri(location); |
| connector.setTaskRunnerFactory(broker.getTaskRunnerFactory()); |
| connector.start(); |
| CONNECTORS.put(host, connector); |
| } |
| |
| } |
| } |
| |
| VMTransport vmtransport = server.connect(); |
| IntrospectionSupport.setProperties(vmtransport.peer, new HashMap<String,String>(options)); |
| IntrospectionSupport.setProperties(vmtransport, options); |
| Transport transport = vmtransport; |
| if (vmtransport.isMarshal()) { |
| Map<String, String> optionsCopy = new HashMap<String, String>(options); |
| transport = new MarshallingTransportFilter(transport, createWireFormat(options), |
| createWireFormat(optionsCopy)); |
| } |
| if (!options.isEmpty()) { |
| throw new IllegalArgumentException("Invalid connect parameters: " + options); |
| } |
| return transport; |
| } |
| |
| private static String extractHost(URI location) { |
| String host = location.getHost(); |
| if (host == null || host.length() == 0) { |
| host = location.getAuthority(); |
| if (host == null || host.length() == 0) { |
| host = "localhost"; |
| } |
| } |
| return host; |
| } |
| |
| /** |
| * @param registry |
| * @param brokerName |
| * @param waitForStart - time in milliseconds to wait for a broker to appear |
| * @return |
| */ |
| private BrokerService lookupBroker(final BrokerRegistry registry, final String brokerName, int waitForStart) { |
| BrokerService broker = null; |
| synchronized(registry.getRegistryMutext()) { |
| broker = registry.lookup(brokerName); |
| if (broker == null && waitForStart > 0) { |
| final long expiry = System.currentTimeMillis() + waitForStart; |
| while (broker == null && expiry > System.currentTimeMillis()) { |
| long timeout = Math.max(0, expiry - System.currentTimeMillis()); |
| try { |
| LOG.debug("waiting for broker named: " + brokerName + " to start"); |
| registry.getRegistryMutext().wait(timeout); |
| } catch (InterruptedException ignored) { |
| } |
| broker = registry.lookup(brokerName); |
| } |
| } |
| } |
| return broker; |
| } |
| |
| public TransportServer doBind(URI location) throws IOException { |
| return bind(location, false); |
| } |
| |
| /** |
| * @param location |
| * @return the TransportServer |
| * @throws IOException |
| */ |
| private TransportServer bind(URI location, boolean dispose) throws IOException { |
| String host = extractHost(location); |
| LOG.debug("binding to broker: " + host); |
| VMTransportServer server = new VMTransportServer(location, dispose); |
| Object currentBoundValue = SERVERS.get(host); |
| if (currentBoundValue != null) { |
| throw new IOException("VMTransportServer already bound at: " + location); |
| } |
| SERVERS.put(host, server); |
| return server; |
| } |
| |
| public static void stopped(VMTransportServer server) { |
| String host = extractHost(server.getBindURI()); |
| stopped(host); |
| } |
| |
| public static void stopped(String host) { |
| SERVERS.remove(host); |
| TransportConnector connector = CONNECTORS.remove(host); |
| if (connector != null) { |
| LOG.debug("Shutting down VM connectors for broker: " + host); |
| ServiceSupport.dispose(connector); |
| BrokerService broker = BROKERS.remove(host); |
| if (broker != null) { |
| ServiceSupport.dispose(broker); |
| } |
| MDC.remove("activemq.broker"); |
| } |
| } |
| |
| public BrokerFactoryHandler getBrokerFactoryHandler() { |
| return brokerFactoryHandler; |
| } |
| |
| public void setBrokerFactoryHandler(BrokerFactoryHandler brokerFactoryHandler) { |
| this.brokerFactoryHandler = brokerFactoryHandler; |
| } |
| |
| private boolean validateBroker(String host) { |
| boolean result = true; |
| if (BROKERS.containsKey(host) || SERVERS.containsKey(host) || CONNECTORS.containsKey(host)) { |
| // check the broker is still in the BrokerRegistry |
| TransportConnector connector = CONNECTORS.get(host); |
| if (BrokerRegistry.getInstance().lookup(host) == null |
| || (connector != null && connector.getBroker().isStopped())) { |
| result = false; |
| // clean-up |
| BROKERS.remove(host); |
| SERVERS.remove(host); |
| if (connector != null) { |
| CONNECTORS.remove(host); |
| if (connector != null) { |
| ServiceSupport.dispose(connector); |
| } |
| } |
| } |
| } |
| return result; |
| } |
| } |
