Updated Branches: refs/heads/master d75f6f0d5 -> bde7577f9
http://git-wip-us.apache.org/repos/asf/karaf/blob/bde7577f/jms/core/src/main/java/org/apache/karaf/jms/internal/JmsServiceImpl.java ---------------------------------------------------------------------- diff --git a/jms/core/src/main/java/org/apache/karaf/jms/internal/JmsServiceImpl.java b/jms/core/src/main/java/org/apache/karaf/jms/internal/JmsServiceImpl.java new file mode 100644 index 0000000..223169c --- /dev/null +++ b/jms/core/src/main/java/org/apache/karaf/jms/internal/JmsServiceImpl.java @@ -0,0 +1,418 @@ +/* + * 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.karaf.jms.internal; + +import org.apache.activemq.ActiveMQConnection; +import org.apache.activemq.advisory.DestinationSource; +import org.apache.activemq.command.ActiveMQQueue; +import org.apache.activemq.command.ActiveMQTopic; +import org.apache.activemq.pool.PooledConnection; +import org.apache.karaf.jms.JmsMessage; +import org.apache.karaf.jms.JmsService; +import org.osgi.framework.BundleContext; +import org.osgi.framework.Constants; +import org.osgi.framework.ServiceReference; + +import javax.jms.*; +import java.io.*; +import java.lang.IllegalStateException; +import java.util.*; + +/** + * Default implementation of the JMS Service. + */ +public class JmsServiceImpl implements JmsService { + + private BundleContext bundleContext; + + @Override + public void create(String name, String type, String url) throws Exception { + if (!type.equalsIgnoreCase("activemq") && !type.equalsIgnoreCase("webspheremq")) { + throw new IllegalArgumentException("JMS connection factory type not known"); + } + + File karafBase = new File(System.getProperty("karaf.base")); + File deployFolder = new File(karafBase, "deploy"); + File outFile = new File(deployFolder, "connectionfactory-" + name + ".xml"); + + if (type.equalsIgnoreCase("activemq")) { + // activemq + HashMap<String, String> properties = new HashMap<String, String>(); + properties.put("${name}", name); + properties.put("${url}", url); + copyDataSourceFile(outFile, "connectionfactory-activemq.xml", properties); + } else { + // webspheremq + String[] splitted = url.split("/"); + if (splitted.length != 4) { + throw new IllegalStateException("WebsphereMQ URI should be in the following format: host/port/queuemanager/channel"); + } + HashMap<String, String> properties = new HashMap<String, String>(); + properties.put("${name}", name); + properties.put("${host}", splitted[0]); + properties.put("${port}", splitted[1]); + properties.put("${queuemanager}", splitted[2]); + properties.put("${channel}", splitted[3]); + copyDataSourceFile(outFile, "connectionfactory-webspheremq.xml", properties); + } + } + + @Override + public void delete(String name) throws Exception { + File karafBase = new File(System.getProperty("karaf.base")); + File deployFolder = new File(karafBase, "deploy"); + File connectionFactoryFile = new File(deployFolder, "connectionfactory-" + name + ".xml"); + if (!connectionFactoryFile.exists()) { + throw new IllegalStateException("The JMS connection factory file " + connectionFactoryFile.getPath() + " doesn't exist"); + } + connectionFactoryFile.delete(); + } + + @Override + public List<String> connectionFactories() throws Exception { + List<String> connectionFactories = new ArrayList<String>(); + ServiceReference[] references = bundleContext.getServiceReferences(ConnectionFactory.class.getName(), null); + if (references != null) { + for (ServiceReference reference : references) { + if (reference.getProperty("osgi.jndi.service.name") != null) { + connectionFactories.add((String) reference.getProperty("osgi.jndi.service.name")); + } else if (reference.getProperty("name") != null) { + connectionFactories.add((String) reference.getProperty("name")); + } else { + connectionFactories.add(reference.getProperty(Constants.SERVICE_ID).toString()); + } + } + } + return connectionFactories; + } + + @Override + public Map<String, String> info(String connectionFactory) throws Exception { + Map<String, String> map = new HashMap<String, String>(); + ServiceReference reference = this.lookupConnectionFactory(connectionFactory); + Connection connection = null; + try { + ConnectionFactory cf = (ConnectionFactory) bundleContext.getService(reference); + connection = cf.createConnection(); + ConnectionMetaData metaData = connection.getMetaData(); + map.put("product", metaData.getJMSProviderName()); + map.put("version", metaData.getProviderVersion()); + } finally { + if (connection != null) { + connection.close(); + } + if (reference != null) { + bundleContext.ungetService(reference); + } + } + return map; + } + + @Override + public int count(String connectionFactory, String destination) throws Exception { + ServiceReference reference = this.lookupConnectionFactory(connectionFactory); + Connection connection = null; + Session session = null; + try { + ConnectionFactory cf = (ConnectionFactory) bundleContext.getService(reference); + connection = cf.createConnection(); + session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + QueueBrowser browser = session.createBrowser(session.createQueue(destination)); + Enumeration<Message> enumeration = browser.getEnumeration(); + int count = 0; + while (enumeration.hasMoreElements()) { + enumeration.nextElement(); + count++; + } + browser.close(); + return count; + } finally { + if (session != null) { + session.close(); + } + if (connection != null) { + connection.close(); + } + if (reference != null) { + bundleContext.ungetService(reference); + } + } + } + + @Override + public List<String> queues(String connectionFactory) throws Exception { + List<String> queues = new ArrayList<String>(); + ServiceReference reference = this.lookupConnectionFactory(connectionFactory); + Connection connection = null; + try { + ConnectionFactory cf = (ConnectionFactory) bundleContext.getService(reference); + connection = cf.createConnection(); + if (connection instanceof PooledConnection) { + connection = ((PooledConnection) connection).getConnection(); + } + if (connection instanceof ActiveMQConnection) { + DestinationSource destinationSource = ((ActiveMQConnection) connection).getDestinationSource(); + Set<ActiveMQQueue> activeMQQueues = destinationSource.getQueues(); + for (ActiveMQQueue activeMQQueue : activeMQQueues) { + queues.add(activeMQQueue.getQueueName()); + } + } + } finally { + if (connection != null) { + connection.close(); + } + if (reference != null) { + bundleContext.ungetService(reference); + } + } + return queues; + } + + @Override + public List<String> topics(String connectionFactory) throws Exception { + List<String> topics = new ArrayList<String>(); + ServiceReference reference = this.lookupConnectionFactory(connectionFactory); + Connection connection = null; + try { + ConnectionFactory cf = (ConnectionFactory) bundleContext.getService(reference); + connection = cf.createConnection(); + if (connection instanceof PooledConnection) { + connection = ((PooledConnection) connection).getConnection(); + } + if (connection instanceof ActiveMQConnection) { + DestinationSource destinationSource = ((ActiveMQConnection) connection).getDestinationSource(); + Set<ActiveMQTopic> activeMQTopics = destinationSource.getTopics(); + for (ActiveMQTopic activeMQTopic : activeMQTopics) { + topics.add(activeMQTopic.getTopicName()); + } + } + } finally { + if (connection != null) { + connection.close(); + } + if (reference != null) { + bundleContext.ungetService(reference); + } + } + return topics; + } + + @Override + public List<JmsMessage> browse(String connectionFactory, String queue, String filter) throws Exception { + List<JmsMessage> messages = new ArrayList<JmsMessage>(); + ServiceReference reference = this.lookupConnectionFactory(connectionFactory); + Connection connection = null; + Session session = null; + try { + ConnectionFactory cf = (ConnectionFactory) bundleContext.getService(reference); + connection = cf.createConnection(); + session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + QueueBrowser browser = session.createBrowser(session.createQueue(queue), filter); + Enumeration<Message> enumeration = browser.getEnumeration(); + while (enumeration.hasMoreElements()) { + Message message = enumeration.nextElement(); + messages.add(new JmsMessage(message)); + } + browser.close(); + } finally { + if (session != null) { + session.close(); + } + if (connection != null) { + connection.close(); + } + if (reference != null) { + bundleContext.ungetService(reference); + } + } + return messages; + } + + @Override + public void send(String connectionFactory, String queue, String body, String replyTo) throws Exception { + ServiceReference reference = this.lookupConnectionFactory(connectionFactory); + Connection connection = null; + Session session = null; + try { + ConnectionFactory cf = (ConnectionFactory) bundleContext.getService(reference); + connection = cf.createConnection(); + session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + Message message = session.createTextMessage(body); + if (replyTo != null) { + message.setJMSReplyTo(session.createQueue(replyTo)); + } + MessageProducer producer = session.createProducer(session.createQueue(queue)); + producer.send(message); + producer.close(); + } finally { + if (session != null) { + session.close(); + } + if (connection != null) { + connection.close(); + } + if (reference != null) { + bundleContext.ungetService(reference); + } + } + } + + @Override + public int consume(String connectionFactory, String queue, String selector) throws Exception { + int count = 0; + ServiceReference reference = this.lookupConnectionFactory(connectionFactory); + Connection connection = null; + Session session = null; + try { + ConnectionFactory cf = (ConnectionFactory) bundleContext.getService(reference); + connection = cf.createConnection(); + session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); + MessageConsumer consumer = session.createConsumer(session.createQueue(queue), selector); + Message message; + do { + message = consumer.receiveNoWait(); + if (message != null) { + count++; + } + } while (message != null); + } finally { + if (session != null) { + session.close(); + } + if (connection != null) { + connection.close(); + } + if (reference != null) { + bundleContext.ungetService(reference); + } + } + return count; + } + + @Override + public int move(String connectionFactory, String sourceQueue, String targetQueue, String selector) throws Exception { + int count = 0; + ServiceReference reference = this.lookupConnectionFactory(connectionFactory); + Connection connection = null; + Session session = null; + try { + ConnectionFactory cf = (ConnectionFactory) bundleContext.getService(reference); + connection = cf.createConnection(); + session = connection.createSession(true, Session.SESSION_TRANSACTED); + MessageConsumer consumer = session.createConsumer(session.createQueue(sourceQueue), selector); + Message message; + do { + message = consumer.receiveNoWait(); + if (message != null) { + MessageProducer producer = session.createProducer(session.createQueue(targetQueue)); + producer.send(message); + count++; + } + } while (message != null); + } finally { + if (session != null) { + session.close(); + } + if (connection != null) { + connection.close(); + } + if (reference != null) { + bundleContext.ungetService(reference); + } + } + return count; + } + + private ServiceReference lookupConnectionFactory(String name) throws Exception { + ServiceReference[] references = bundleContext.getServiceReferences(ConnectionFactory.class.getName(), "(|(osgi.jndi.service.name=" + name + ")(name=" + name + ")(service.id=" + name + "))"); + if (references == null || references.length == 0) { + throw new IllegalArgumentException("No JMS connection factory found for " + name); + } + if (references.length > 1) { + throw new IllegalArgumentException("Multiple JMS connection factories found for " + name); + } + return references[0]; + } + + private void copyDataSourceFile(File outFile, String resource, HashMap<String, String> properties) throws Exception { + if (!outFile.exists()) { + InputStream is = JmsServiceImpl.class.getResourceAsStream(resource); + if (is == null) { + throw new IllegalArgumentException("Resource " + resource + " doesn't exist"); + } + try { + // read it line at a time so that we can use the platform line ending when we write it out + PrintStream out = new PrintStream(new FileOutputStream(outFile)); + try { + Scanner scanner = new Scanner(is); + while (scanner.hasNextLine()) { + String line = scanner.nextLine(); + line = filter(line, properties); + out.println(line); + } + } finally { + safeClose(out); + } + } finally { + safeClose(is); + } + } else { + throw new IllegalArgumentException("File " + outFile.getPath() + " already exists. Remove it if you wish to recreate it."); + } + } + + private void safeClose(InputStream is) throws IOException { + if (is == null) + return; + try { + is.close(); + } catch (Throwable ignore) { + // nothing to do + } + } + + private void safeClose(OutputStream is) throws IOException { + if (is == null) + return; + try { + is.close(); + } catch (Throwable ignore) { + // nothing to do + } + } + + private String filter(String line, HashMap<String, String> props) { + for (Map.Entry<String, String> i : props.entrySet()) { + int p1 = line.indexOf(i.getKey()); + if (p1 >= 0) { + String l1 = line.substring(0, p1); + String l2 = line.substring(p1 + i.getKey().length()); + line = l1 + i.getValue() + l2; + } + } + return line; + } + + public BundleContext getBundleContext() { + return bundleContext; + } + + public void setBundleContext(BundleContext bundleContext) { + this.bundleContext = bundleContext; + } + +} http://git-wip-us.apache.org/repos/asf/karaf/blob/bde7577f/jms/core/src/main/resources/OSGI-INF/blueprint/jms-core.xml ---------------------------------------------------------------------- diff --git a/jms/core/src/main/resources/OSGI-INF/blueprint/jms-core.xml b/jms/core/src/main/resources/OSGI-INF/blueprint/jms-core.xml new file mode 100644 index 0000000..7dd8070 --- /dev/null +++ b/jms/core/src/main/resources/OSGI-INF/blueprint/jms-core.xml @@ -0,0 +1,41 @@ +<?xml version="1.0" encoding="UTF-8"?> +<!-- + + 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. + --> +<blueprint xmlns="http://www.osgi.org/xmlns/blueprint/v1.0.0"; + xmlns:ext="http://aries.apache.org/blueprint/xmlns/blueprint-ext/v1.0.0"; + default-activation="lazy"> + + <ext:property-placeholder placeholder-prefix="$[" placeholder-suffix="]"/> + + <bean id="jmsService" class="org.apache.karaf.jms.internal.JmsServiceImpl"> + <property name="bundleContext" ref="blueprintBundleContext"/> + </bean> + + <service ref="jmsService" interface="org.apache.karaf.jms.JmsService" /> + + <!-- Management --> + <bean id="jmsMBeanImpl" class="org.apache.karaf.jms.internal.JmsMBeanImpl"> + <property name="jmsService" ref="jmsService"/> + </bean> + + <service ref="jmsMBeanImpl" auto-export="interfaces"> + <service-properties> + <entry key="jmx.objectname" value="org.apache.karaf:type=jms,name=$[karaf.name]"/> + </service-properties> + </service> + +</blueprint> \ No newline at end of file http://git-wip-us.apache.org/repos/asf/karaf/blob/bde7577f/jms/core/src/main/resources/OSGI-INF/bundle.info ---------------------------------------------------------------------- diff --git a/jms/core/src/main/resources/OSGI-INF/bundle.info b/jms/core/src/main/resources/OSGI-INF/bundle.info new file mode 100644 index 0000000..1aeb646 --- /dev/null +++ b/jms/core/src/main/resources/OSGI-INF/bundle.info @@ -0,0 +1,18 @@ +h1. Synopsis + +${project.name} + +${project.description} + +Maven URL: +[mvn:${project.groupId}/${project.artifactId}/${project.version}] + +h1. Description + +This bundle is the core implementation of the JMS service support. + +The JMS service allows you to create connection factories, and send/browse/consume messages. + +h1. See also + +JMS - section of the Karaf User Guide http://git-wip-us.apache.org/repos/asf/karaf/blob/bde7577f/jms/core/src/main/resources/org.apache.karaf.jms.internal/connectionFactory-webspheremq.xml ---------------------------------------------------------------------- diff --git a/jms/core/src/main/resources/org.apache.karaf.jms.internal/connectionFactory-webspheremq.xml b/jms/core/src/main/resources/org.apache.karaf.jms.internal/connectionFactory-webspheremq.xml new file mode 100644 index 0000000..68d04d6 --- /dev/null +++ b/jms/core/src/main/resources/org.apache.karaf.jms.internal/connectionFactory-webspheremq.xml @@ -0,0 +1,20 @@ +<?xml version="1.0" encoding="UTF-8"?> +<blueprint xmlns="http://www.osgi.org/xmlns/blueprint/v1.0.0";> + + <bean id="wmqConnectionFactory" class="com.ibm.mq.jms.MQQueueConnectionFactory"> + <property name="transportType" value="1" /> + <property name="hostName" value="${hostname}" /> + <property name="port" value="${port}" /> + <property name="queueManager" value="${queuemanager}" /> + <property name="channel" value="${channel}" /> + <property name="useConnectionPooling" value="true" /> + </bean> + + <service ref="wmqConnectionFactory" interface="javax.jms.ConnectionFactory"> + <service-properties> + <entry key="name" value="${name}"/> + <entry key="osgi.jndi.service.name" value="/jms/${name}"/> + </service-properties> + </service> + +</blueprint> \ No newline at end of file http://git-wip-us.apache.org/repos/asf/karaf/blob/bde7577f/jms/core/src/main/resources/org.apache.karaf.jms.internal/connectionfactory-activemq.xml ---------------------------------------------------------------------- diff --git a/jms/core/src/main/resources/org.apache.karaf.jms.internal/connectionfactory-activemq.xml b/jms/core/src/main/resources/org.apache.karaf.jms.internal/connectionfactory-activemq.xml new file mode 100644 index 0000000..a0b170c --- /dev/null +++ b/jms/core/src/main/resources/org.apache.karaf.jms.internal/connectionfactory-activemq.xml @@ -0,0 +1,28 @@ +<?xml version="1.0" encoding="UTF-8"?> +<blueprint xmlns="http://www.osgi.org/xmlns/blueprint/v1.0.0";> + + <bean id="activemqConnectionFactory" class="org.apache.activemq.ActiveMQConnectionFactory"> + <property name="brokerURL" value="${url}" /> + </bean> + + <bean id="pooledConnectionFactory" class="org.apache.activemq.pool.PooledConnectionFactory"> + <property name="maxConnections" value="8" /> + <property name="connectionFactory" ref="activemqConnectionFactory" /> + </bean> + + <bean id="resourceManager" class="org.apache.activemq.pool.ActiveMQResourceManager" init-method="recoverResource"> + <property name="transactionManager" ref="transactionManager" /> + <property name="connectionFactory" ref="activemqConnectionFactory" /> + <property name="resourceName" value="activemq.localhost" /> + </bean> + + <reference id="transactionManager" interface="javax.transaction.TransactionManager" /> + + <service ref="pooledConnectionFactory" interface="javax.jms.ConnectionFactory"> + <service-properties> + <entry key="name" value="${name}" /> + <entry key="osgi.jndi.service.name" value="/jms/${name}" /> + </service-properties> + </service> + +</blueprint> \ No newline at end of file http://git-wip-us.apache.org/repos/asf/karaf/blob/bde7577f/jms/pom.xml ---------------------------------------------------------------------- diff --git a/jms/pom.xml b/jms/pom.xml new file mode 100644 index 0000000..70d9167 --- /dev/null +++ b/jms/pom.xml @@ -0,0 +1,41 @@ +<?xml version="1.0" encoding="UTF-8"?> +<project xmlns="http://maven.apache.org/POM/4.0.0"; xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"; xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd";> + + <!-- + + 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. + --> + + <modelVersion>4.0.0</modelVersion> + + <parent> + <groupId>org.apache.karaf</groupId> + <artifactId>karaf</artifactId> + <version>3.0.0-SNAPSHOT</version> + <relativePath>../pom.xml</relativePath> + </parent> + + <groupId>org.apache.karaf.jms</groupId> + <artifactId>jms</artifactId> + <packaging>pom</packaging> + <name>Apache Karaf :: JMS</name> + + <modules> + <module>core</module> + <module>command</module> + </modules> + +</project> \ No newline at end of file http://git-wip-us.apache.org/repos/asf/karaf/blob/bde7577f/pom.xml ---------------------------------------------------------------------- diff --git a/pom.xml b/pom.xml index b734f03..66615b0 100644 --- a/pom.xml +++ b/pom.xml @@ -64,6 +64,7 @@ <module>obr</module> <module>jndi</module> <module>jdbc</module> + <module>jms</module> <module>tooling</module> <module>assemblies</module> <module>demos</module> @@ -402,6 +403,17 @@ </dependency> <dependency> + <groupId>org.apache.karaf.jms</groupId> + <artifactId>org.apache.karaf.jms.core</artifactId> + <version>${project.version}</version> + </dependency> + <dependency> + <groupId>org.apache.karaf.jms</groupId> + <artifactId>org.apache.karaf.jms.command</artifactId> + <version>${project.version}</version> + </dependency> + + <dependency> <groupId>org.apache.karaf.jndi</groupId> <artifactId>org.apache.karaf.jndi.core</artifactId> <version>${project.version}</version>
