@Test public void systemUsage() throws Exception { final URI brokerURI = new URI("amq5factory:broker:(tcp: "systemUsage=true&systemUsage.memory.limit=1024"); final BrokerService bs = new ActiveMQ5Factory().createBroker(brokerURI); bs.stop(); ActiveMQ5Factory.brokers.remove(brokerURI); assertEquals(1024, bs.getSystemUsage().getMemoryUsage().getLimit()); }
@Override public synchronized BrokerService createBroker(final URI brokerURI) throws Exception { Logger.getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class).getChildLogger("service").info("ActiveMQ5Factory creating broker"); BrokerService broker = brokers.get(brokerURI); if (null == broker || !broker.isStarted()) { final Properties properties = getLowerCaseProperties(); final URISupport.CompositeData compositeData = URISupport.parseComposite(new URI(brokerURI.getRawSchemeSpecificPart())); final Map<String, String> params = new HashMap<>(compositeData.getParameters()); final PersistenceAdapter persistenceAdapter; if ("true".equals(params.remove("usekahadb"))) { persistenceAdapter = createPersistenceAdapter("org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter", "kahadb", params); } else if ("true".equals(params.remove("useleveldb"))) { persistenceAdapter = createPersistenceAdapter("org.apache.activemq.store.leveldb.LevelDBPersistenceAdapter", "leveldb", params); } else if (params.get("persistenceadapter") != null) { final String adapter = params.remove("persistenceadapter"); persistenceAdapter = createPersistenceAdapter(adapter, "persistence", params); } else { persistenceAdapter = null; } final String systemUsage = params.remove("systemUsage"); final SystemUsage systemUsageInstance; if ("true".equalsIgnoreCase(systemUsage)) { systemUsageInstance = newSystemUsage(params); } else { systemUsageInstance = null; } final BrokerPlugin[] plugins = createPlugins(params); final URI uri = new URI(cleanUpUri(brokerURI.getRawSchemeSpecificPart(), compositeData.getParameters(), params)); broker = "broker".equals(uri.getScheme()) ? newDefaultBroker(uri) : BrokerFactory.createBroker(uri); if (plugins != null) { broker.setPlugins(plugins); } if (systemUsageInstance != null) { broker.setSystemUsage(systemUsageInstance); } brokers.put(brokerURI, broker); if (persistenceAdapter != null) { broker.setPersistenceAdapter(persistenceAdapter); broker.setPersistenceFactory(new ProvidedPersistenceAdapterPersistenceAdapterFactory(persistenceAdapter)); broker.setPersistent(true); tomeeConfig(broker); } else { final boolean notXbean = !uri.getScheme().toLowerCase(Locale.ENGLISH).startsWith("xbean"); if (notXbean) { Object value = properties.get("datasource"); if (String.class.isInstance(value) && value.toString().length() == 0) { value = null; } final DataSource dataSource; if (value != null) { if (DataSource.class.isInstance(value)) { dataSource = DataSource.class.cast(value); } else if (String.class.isInstance(value)) { final String resouceId = String.class.cast(value); try { final ContainerSystem containerSystem = SystemInstance.get().getComponent(ContainerSystem.class); final Context context = containerSystem.getJNDIContext(); final Object obj = context.lookup("openejb/Resource/" + resouceId); if (!(obj instanceof DataSource)) { throw new IllegalArgumentException("Resource with id " + resouceId + " is not a DataSource, but is " + obj.getClass().getName()); } dataSource = (DataSource) obj; } catch (final NamingException e) { throw new IllegalArgumentException("Unknown datasource " + resouceId); } } else { throw new IllegalArgumentException("Unexpected datasource definition: " + value); } } else { dataSource = null; } if (null != dataSource) { final JDBCPersistenceAdapter adapter = new JDBCPersistenceAdapter(); if (properties.containsKey("usedatabaselock")) { adapter.setUseLock(Boolean.parseBoolean(properties.getProperty("usedatabaselock", "true"))); } adapter.setDataSource(dataSource); broker.setPersistent(true); broker.setPersistenceAdapter(adapter); } else { broker.setPersistenceAdapter(new MemoryPersistenceAdapter()); } tomeeConfig(broker); } } broker.setUseShutdownHook(false); broker.setSystemExitOnShutdown(false); broker.setStartAsync(false); final ManagementContext managementContext = new ManagementContext(); managementContext.setCreateConnector(false); broker.setManagementContext(managementContext); final BrokerService bs = broker; final Thread start = new Thread("ActiveMQFactory start and checkpoint") { @Override public void run() { Thread.currentThread().setContextClassLoader(ActiveMQResourceAdapter.class.getClassLoader()); try { if (!bs.isStarted()) { Logger .getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class) .getChildLogger("service") .info("Starting ActiveMQ BrokerService"); bs.start(); } bs.waitUntilStarted(); Logger .getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class) .getChildLogger("service") .info("Starting ActiveMQ checkpoint"); bs.getPersistenceAdapter().checkpoint(true); started.set(true); } catch (final Throwable t) { throwable = t; } } }; int timeout = 30000; try { timeout = Integer.parseInt(properties.getProperty("startuptimeout", "30000")); Logger .getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class) .getChildLogger("service") .info("Using ActiveMQ startup timeout of " + timeout + "ms"); } catch (final Throwable e) { } start.setDaemon(true); start.start(); try { start.join(timeout); } catch (final InterruptedException e) { } if (null != throwable) { Logger.getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class).getChildLogger("service").error("ActiveMQ failed to start broker", throwable); } else if (started.get()) { Logger.getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class).getChildLogger("service").info("ActiveMQ broker started"); } else { Logger .getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class) .getChildLogger("service") .warning("ActiveMQ failed to start broker within " + timeout + " seconds - It may be unusable"); } } return broker; }
ActiveMQ5Factory implements BrokerFactoryHandler { @Override public synchronized BrokerService createBroker(final URI brokerURI) throws Exception { Logger.getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class).getChildLogger("service").info("ActiveMQ5Factory creating broker"); BrokerService broker = brokers.get(brokerURI); if (null == broker || !broker.isStarted()) { final Properties properties = getLowerCaseProperties(); final URISupport.CompositeData compositeData = URISupport.parseComposite(new URI(brokerURI.getRawSchemeSpecificPart())); final Map<String, String> params = new HashMap<>(compositeData.getParameters()); final PersistenceAdapter persistenceAdapter; if ("true".equals(params.remove("usekahadb"))) { persistenceAdapter = createPersistenceAdapter("org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter", "kahadb", params); } else if ("true".equals(params.remove("useleveldb"))) { persistenceAdapter = createPersistenceAdapter("org.apache.activemq.store.leveldb.LevelDBPersistenceAdapter", "leveldb", params); } else if (params.get("persistenceadapter") != null) { final String adapter = params.remove("persistenceadapter"); persistenceAdapter = createPersistenceAdapter(adapter, "persistence", params); } else { persistenceAdapter = null; } final String systemUsage = params.remove("systemUsage"); final SystemUsage systemUsageInstance; if ("true".equalsIgnoreCase(systemUsage)) { systemUsageInstance = newSystemUsage(params); } else { systemUsageInstance = null; } final BrokerPlugin[] plugins = createPlugins(params); final URI uri = new URI(cleanUpUri(brokerURI.getRawSchemeSpecificPart(), compositeData.getParameters(), params)); broker = "broker".equals(uri.getScheme()) ? newDefaultBroker(uri) : BrokerFactory.createBroker(uri); if (plugins != null) { broker.setPlugins(plugins); } if (systemUsageInstance != null) { broker.setSystemUsage(systemUsageInstance); } brokers.put(brokerURI, broker); if (persistenceAdapter != null) { broker.setPersistenceAdapter(persistenceAdapter); broker.setPersistenceFactory(new ProvidedPersistenceAdapterPersistenceAdapterFactory(persistenceAdapter)); broker.setPersistent(true); tomeeConfig(broker); } else { final boolean notXbean = !uri.getScheme().toLowerCase(Locale.ENGLISH).startsWith("xbean"); if (notXbean) { Object value = properties.get("datasource"); if (String.class.isInstance(value) && value.toString().length() == 0) { value = null; } final DataSource dataSource; if (value != null) { if (DataSource.class.isInstance(value)) { dataSource = DataSource.class.cast(value); } else if (String.class.isInstance(value)) { final String resouceId = String.class.cast(value); try { final ContainerSystem containerSystem = SystemInstance.get().getComponent(ContainerSystem.class); final Context context = containerSystem.getJNDIContext(); final Object obj = context.lookup("openejb/Resource/" + resouceId); if (!(obj instanceof DataSource)) { throw new IllegalArgumentException("Resource with id " + resouceId + " is not a DataSource, but is " + obj.getClass().getName()); } dataSource = (DataSource) obj; } catch (final NamingException e) { throw new IllegalArgumentException("Unknown datasource " + resouceId); } } else { throw new IllegalArgumentException("Unexpected datasource definition: " + value); } } else { dataSource = null; } if (null != dataSource) { final JDBCPersistenceAdapter adapter = new JDBCPersistenceAdapter(); if (properties.containsKey("usedatabaselock")) { adapter.setUseLock(Boolean.parseBoolean(properties.getProperty("usedatabaselock", "true"))); } adapter.setDataSource(dataSource); broker.setPersistent(true); broker.setPersistenceAdapter(adapter); } else { broker.setPersistenceAdapter(new MemoryPersistenceAdapter()); } tomeeConfig(broker); } } broker.setUseShutdownHook(false); broker.setSystemExitOnShutdown(false); broker.setStartAsync(false); final ManagementContext managementContext = new ManagementContext(); managementContext.setCreateConnector(false); broker.setManagementContext(managementContext); final BrokerService bs = broker; final Thread start = new Thread("ActiveMQFactory start and checkpoint") { @Override public void run() { Thread.currentThread().setContextClassLoader(ActiveMQResourceAdapter.class.getClassLoader()); try { if (!bs.isStarted()) { Logger .getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class) .getChildLogger("service") .info("Starting ActiveMQ BrokerService"); bs.start(); } bs.waitUntilStarted(); Logger .getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class) .getChildLogger("service") .info("Starting ActiveMQ checkpoint"); bs.getPersistenceAdapter().checkpoint(true); started.set(true); } catch (final Throwable t) { throwable = t; } } }; int timeout = 30000; try { timeout = Integer.parseInt(properties.getProperty("startuptimeout", "30000")); Logger .getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class) .getChildLogger("service") .info("Using ActiveMQ startup timeout of " + timeout + "ms"); } catch (final Throwable e) { } start.setDaemon(true); start.start(); try { start.join(timeout); } catch (final InterruptedException e) { } if (null != throwable) { Logger.getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class).getChildLogger("service").error("ActiveMQ failed to start broker", throwable); } else if (started.get()) { Logger.getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class).getChildLogger("service").info("ActiveMQ broker started"); } else { Logger .getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class) .getChildLogger("service") .warning("ActiveMQ failed to start broker within " + timeout + " seconds - It may be unusable"); } } return broker; } }
ActiveMQ5Factory implements BrokerFactoryHandler { @Override public synchronized BrokerService createBroker(final URI brokerURI) throws Exception { Logger.getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class).getChildLogger("service").info("ActiveMQ5Factory creating broker"); BrokerService broker = brokers.get(brokerURI); if (null == broker || !broker.isStarted()) { final Properties properties = getLowerCaseProperties(); final URISupport.CompositeData compositeData = URISupport.parseComposite(new URI(brokerURI.getRawSchemeSpecificPart())); final Map<String, String> params = new HashMap<>(compositeData.getParameters()); final PersistenceAdapter persistenceAdapter; if ("true".equals(params.remove("usekahadb"))) { persistenceAdapter = createPersistenceAdapter("org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter", "kahadb", params); } else if ("true".equals(params.remove("useleveldb"))) { persistenceAdapter = createPersistenceAdapter("org.apache.activemq.store.leveldb.LevelDBPersistenceAdapter", "leveldb", params); } else if (params.get("persistenceadapter") != null) { final String adapter = params.remove("persistenceadapter"); persistenceAdapter = createPersistenceAdapter(adapter, "persistence", params); } else { persistenceAdapter = null; } final String systemUsage = params.remove("systemUsage"); final SystemUsage systemUsageInstance; if ("true".equalsIgnoreCase(systemUsage)) { systemUsageInstance = newSystemUsage(params); } else { systemUsageInstance = null; } final BrokerPlugin[] plugins = createPlugins(params); final URI uri = new URI(cleanUpUri(brokerURI.getRawSchemeSpecificPart(), compositeData.getParameters(), params)); broker = "broker".equals(uri.getScheme()) ? newDefaultBroker(uri) : BrokerFactory.createBroker(uri); if (plugins != null) { broker.setPlugins(plugins); } if (systemUsageInstance != null) { broker.setSystemUsage(systemUsageInstance); } brokers.put(brokerURI, broker); if (persistenceAdapter != null) { broker.setPersistenceAdapter(persistenceAdapter); broker.setPersistenceFactory(new ProvidedPersistenceAdapterPersistenceAdapterFactory(persistenceAdapter)); broker.setPersistent(true); tomeeConfig(broker); } else { final boolean notXbean = !uri.getScheme().toLowerCase(Locale.ENGLISH).startsWith("xbean"); if (notXbean) { Object value = properties.get("datasource"); if (String.class.isInstance(value) && value.toString().length() == 0) { value = null; } final DataSource dataSource; if (value != null) { if (DataSource.class.isInstance(value)) { dataSource = DataSource.class.cast(value); } else if (String.class.isInstance(value)) { final String resouceId = String.class.cast(value); try { final ContainerSystem containerSystem = SystemInstance.get().getComponent(ContainerSystem.class); final Context context = containerSystem.getJNDIContext(); final Object obj = context.lookup("openejb/Resource/" + resouceId); if (!(obj instanceof DataSource)) { throw new IllegalArgumentException("Resource with id " + resouceId + " is not a DataSource, but is " + obj.getClass().getName()); } dataSource = (DataSource) obj; } catch (final NamingException e) { throw new IllegalArgumentException("Unknown datasource " + resouceId); } } else { throw new IllegalArgumentException("Unexpected datasource definition: " + value); } } else { dataSource = null; } if (null != dataSource) { final JDBCPersistenceAdapter adapter = new JDBCPersistenceAdapter(); if (properties.containsKey("usedatabaselock")) { adapter.setUseLock(Boolean.parseBoolean(properties.getProperty("usedatabaselock", "true"))); } adapter.setDataSource(dataSource); broker.setPersistent(true); broker.setPersistenceAdapter(adapter); } else { broker.setPersistenceAdapter(new MemoryPersistenceAdapter()); } tomeeConfig(broker); } } broker.setUseShutdownHook(false); broker.setSystemExitOnShutdown(false); broker.setStartAsync(false); final ManagementContext managementContext = new ManagementContext(); managementContext.setCreateConnector(false); broker.setManagementContext(managementContext); final BrokerService bs = broker; final Thread start = new Thread("ActiveMQFactory start and checkpoint") { @Override public void run() { Thread.currentThread().setContextClassLoader(ActiveMQResourceAdapter.class.getClassLoader()); try { if (!bs.isStarted()) { Logger .getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class) .getChildLogger("service") .info("Starting ActiveMQ BrokerService"); bs.start(); } bs.waitUntilStarted(); Logger .getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class) .getChildLogger("service") .info("Starting ActiveMQ checkpoint"); bs.getPersistenceAdapter().checkpoint(true); started.set(true); } catch (final Throwable t) { throwable = t; } } }; int timeout = 30000; try { timeout = Integer.parseInt(properties.getProperty("startuptimeout", "30000")); Logger .getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class) .getChildLogger("service") .info("Using ActiveMQ startup timeout of " + timeout + "ms"); } catch (final Throwable e) { } start.setDaemon(true); start.start(); try { start.join(timeout); } catch (final InterruptedException e) { } if (null != throwable) { Logger.getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class).getChildLogger("service").error("ActiveMQ failed to start broker", throwable); } else if (started.get()) { Logger.getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class).getChildLogger("service").info("ActiveMQ broker started"); } else { Logger .getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class) .getChildLogger("service") .warning("ActiveMQ failed to start broker within " + timeout + " seconds - It may be unusable"); } } return broker; }  }
ActiveMQ5Factory implements BrokerFactoryHandler { @Override public synchronized BrokerService createBroker(final URI brokerURI) throws Exception { Logger.getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class).getChildLogger("service").info("ActiveMQ5Factory creating broker"); BrokerService broker = brokers.get(brokerURI); if (null == broker || !broker.isStarted()) { final Properties properties = getLowerCaseProperties(); final URISupport.CompositeData compositeData = URISupport.parseComposite(new URI(brokerURI.getRawSchemeSpecificPart())); final Map<String, String> params = new HashMap<>(compositeData.getParameters()); final PersistenceAdapter persistenceAdapter; if ("true".equals(params.remove("usekahadb"))) { persistenceAdapter = createPersistenceAdapter("org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter", "kahadb", params); } else if ("true".equals(params.remove("useleveldb"))) { persistenceAdapter = createPersistenceAdapter("org.apache.activemq.store.leveldb.LevelDBPersistenceAdapter", "leveldb", params); } else if (params.get("persistenceadapter") != null) { final String adapter = params.remove("persistenceadapter"); persistenceAdapter = createPersistenceAdapter(adapter, "persistence", params); } else { persistenceAdapter = null; } final String systemUsage = params.remove("systemUsage"); final SystemUsage systemUsageInstance; if ("true".equalsIgnoreCase(systemUsage)) { systemUsageInstance = newSystemUsage(params); } else { systemUsageInstance = null; } final BrokerPlugin[] plugins = createPlugins(params); final URI uri = new URI(cleanUpUri(brokerURI.getRawSchemeSpecificPart(), compositeData.getParameters(), params)); broker = "broker".equals(uri.getScheme()) ? newDefaultBroker(uri) : BrokerFactory.createBroker(uri); if (plugins != null) { broker.setPlugins(plugins); } if (systemUsageInstance != null) { broker.setSystemUsage(systemUsageInstance); } brokers.put(brokerURI, broker); if (persistenceAdapter != null) { broker.setPersistenceAdapter(persistenceAdapter); broker.setPersistenceFactory(new ProvidedPersistenceAdapterPersistenceAdapterFactory(persistenceAdapter)); broker.setPersistent(true); tomeeConfig(broker); } else { final boolean notXbean = !uri.getScheme().toLowerCase(Locale.ENGLISH).startsWith("xbean"); if (notXbean) { Object value = properties.get("datasource"); if (String.class.isInstance(value) && value.toString().length() == 0) { value = null; } final DataSource dataSource; if (value != null) { if (DataSource.class.isInstance(value)) { dataSource = DataSource.class.cast(value); } else if (String.class.isInstance(value)) { final String resouceId = String.class.cast(value); try { final ContainerSystem containerSystem = SystemInstance.get().getComponent(ContainerSystem.class); final Context context = containerSystem.getJNDIContext(); final Object obj = context.lookup("openejb/Resource/" + resouceId); if (!(obj instanceof DataSource)) { throw new IllegalArgumentException("Resource with id " + resouceId + " is not a DataSource, but is " + obj.getClass().getName()); } dataSource = (DataSource) obj; } catch (final NamingException e) { throw new IllegalArgumentException("Unknown datasource " + resouceId); } } else { throw new IllegalArgumentException("Unexpected datasource definition: " + value); } } else { dataSource = null; } if (null != dataSource) { final JDBCPersistenceAdapter adapter = new JDBCPersistenceAdapter(); if (properties.containsKey("usedatabaselock")) { adapter.setUseLock(Boolean.parseBoolean(properties.getProperty("usedatabaselock", "true"))); } adapter.setDataSource(dataSource); broker.setPersistent(true); broker.setPersistenceAdapter(adapter); } else { broker.setPersistenceAdapter(new MemoryPersistenceAdapter()); } tomeeConfig(broker); } } broker.setUseShutdownHook(false); broker.setSystemExitOnShutdown(false); broker.setStartAsync(false); final ManagementContext managementContext = new ManagementContext(); managementContext.setCreateConnector(false); broker.setManagementContext(managementContext); final BrokerService bs = broker; final Thread start = new Thread("ActiveMQFactory start and checkpoint") { @Override public void run() { Thread.currentThread().setContextClassLoader(ActiveMQResourceAdapter.class.getClassLoader()); try { if (!bs.isStarted()) { Logger .getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class) .getChildLogger("service") .info("Starting ActiveMQ BrokerService"); bs.start(); } bs.waitUntilStarted(); Logger .getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class) .getChildLogger("service") .info("Starting ActiveMQ checkpoint"); bs.getPersistenceAdapter().checkpoint(true); started.set(true); } catch (final Throwable t) { throwable = t; } } }; int timeout = 30000; try { timeout = Integer.parseInt(properties.getProperty("startuptimeout", "30000")); Logger .getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class) .getChildLogger("service") .info("Using ActiveMQ startup timeout of " + timeout + "ms"); } catch (final Throwable e) { } start.setDaemon(true); start.start(); try { start.join(timeout); } catch (final InterruptedException e) { } if (null != throwable) { Logger.getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class).getChildLogger("service").error("ActiveMQ failed to start broker", throwable); } else if (started.get()) { Logger.getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class).getChildLogger("service").info("ActiveMQ broker started"); } else { Logger .getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class) .getChildLogger("service") .warning("ActiveMQ failed to start broker within " + timeout + " seconds - It may be unusable"); } } return broker; }  static void setThreadProperties(final Properties p); @Override synchronized BrokerService createBroker(final URI brokerURI); Collection<BrokerService> getBrokers(); }
ActiveMQ5Factory implements BrokerFactoryHandler { @Override public synchronized BrokerService createBroker(final URI brokerURI) throws Exception { Logger.getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class).getChildLogger("service").info("ActiveMQ5Factory creating broker"); BrokerService broker = brokers.get(brokerURI); if (null == broker || !broker.isStarted()) { final Properties properties = getLowerCaseProperties(); final URISupport.CompositeData compositeData = URISupport.parseComposite(new URI(brokerURI.getRawSchemeSpecificPart())); final Map<String, String> params = new HashMap<>(compositeData.getParameters()); final PersistenceAdapter persistenceAdapter; if ("true".equals(params.remove("usekahadb"))) { persistenceAdapter = createPersistenceAdapter("org.apache.activemq.store.kahadb.KahaDBPersistenceAdapter", "kahadb", params); } else if ("true".equals(params.remove("useleveldb"))) { persistenceAdapter = createPersistenceAdapter("org.apache.activemq.store.leveldb.LevelDBPersistenceAdapter", "leveldb", params); } else if (params.get("persistenceadapter") != null) { final String adapter = params.remove("persistenceadapter"); persistenceAdapter = createPersistenceAdapter(adapter, "persistence", params); } else { persistenceAdapter = null; } final String systemUsage = params.remove("systemUsage"); final SystemUsage systemUsageInstance; if ("true".equalsIgnoreCase(systemUsage)) { systemUsageInstance = newSystemUsage(params); } else { systemUsageInstance = null; } final BrokerPlugin[] plugins = createPlugins(params); final URI uri = new URI(cleanUpUri(brokerURI.getRawSchemeSpecificPart(), compositeData.getParameters(), params)); broker = "broker".equals(uri.getScheme()) ? newDefaultBroker(uri) : BrokerFactory.createBroker(uri); if (plugins != null) { broker.setPlugins(plugins); } if (systemUsageInstance != null) { broker.setSystemUsage(systemUsageInstance); } brokers.put(brokerURI, broker); if (persistenceAdapter != null) { broker.setPersistenceAdapter(persistenceAdapter); broker.setPersistenceFactory(new ProvidedPersistenceAdapterPersistenceAdapterFactory(persistenceAdapter)); broker.setPersistent(true); tomeeConfig(broker); } else { final boolean notXbean = !uri.getScheme().toLowerCase(Locale.ENGLISH).startsWith("xbean"); if (notXbean) { Object value = properties.get("datasource"); if (String.class.isInstance(value) && value.toString().length() == 0) { value = null; } final DataSource dataSource; if (value != null) { if (DataSource.class.isInstance(value)) { dataSource = DataSource.class.cast(value); } else if (String.class.isInstance(value)) { final String resouceId = String.class.cast(value); try { final ContainerSystem containerSystem = SystemInstance.get().getComponent(ContainerSystem.class); final Context context = containerSystem.getJNDIContext(); final Object obj = context.lookup("openejb/Resource/" + resouceId); if (!(obj instanceof DataSource)) { throw new IllegalArgumentException("Resource with id " + resouceId + " is not a DataSource, but is " + obj.getClass().getName()); } dataSource = (DataSource) obj; } catch (final NamingException e) { throw new IllegalArgumentException("Unknown datasource " + resouceId); } } else { throw new IllegalArgumentException("Unexpected datasource definition: " + value); } } else { dataSource = null; } if (null != dataSource) { final JDBCPersistenceAdapter adapter = new JDBCPersistenceAdapter(); if (properties.containsKey("usedatabaselock")) { adapter.setUseLock(Boolean.parseBoolean(properties.getProperty("usedatabaselock", "true"))); } adapter.setDataSource(dataSource); broker.setPersistent(true); broker.setPersistenceAdapter(adapter); } else { broker.setPersistenceAdapter(new MemoryPersistenceAdapter()); } tomeeConfig(broker); } } broker.setUseShutdownHook(false); broker.setSystemExitOnShutdown(false); broker.setStartAsync(false); final ManagementContext managementContext = new ManagementContext(); managementContext.setCreateConnector(false); broker.setManagementContext(managementContext); final BrokerService bs = broker; final Thread start = new Thread("ActiveMQFactory start and checkpoint") { @Override public void run() { Thread.currentThread().setContextClassLoader(ActiveMQResourceAdapter.class.getClassLoader()); try { if (!bs.isStarted()) { Logger .getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class) .getChildLogger("service") .info("Starting ActiveMQ BrokerService"); bs.start(); } bs.waitUntilStarted(); Logger .getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class) .getChildLogger("service") .info("Starting ActiveMQ checkpoint"); bs.getPersistenceAdapter().checkpoint(true); started.set(true); } catch (final Throwable t) { throwable = t; } } }; int timeout = 30000; try { timeout = Integer.parseInt(properties.getProperty("startuptimeout", "30000")); Logger .getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class) .getChildLogger("service") .info("Using ActiveMQ startup timeout of " + timeout + "ms"); } catch (final Throwable e) { } start.setDaemon(true); start.start(); try { start.join(timeout); } catch (final InterruptedException e) { } if (null != throwable) { Logger.getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class).getChildLogger("service").error("ActiveMQ failed to start broker", throwable); } else if (started.get()) { Logger.getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class).getChildLogger("service").info("ActiveMQ broker started"); } else { Logger .getInstance(LogCategory.OPENEJB_STARTUP, ActiveMQ5Factory.class) .getChildLogger("service") .warning("ActiveMQ failed to start broker within " + timeout + " seconds - It may be unusable"); } } return broker; }  static void setThreadProperties(final Properties p); @Override synchronized BrokerService createBroker(final URI brokerURI); Collection<BrokerService> getBrokers();  }