You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@servicemix.apache.org by ch...@apache.org on 2006/02/22 00:40:29 UTC
svn commit: r379627 [24/34] - in /incubator/servicemix/trunk: ./ etc/
sandbox/servicemix-wsn-1.2/src/sa/META-INF/
sandbox/servicemix-wsn-1.2/src/su/META-INF/ servicemix-assembly/
servicemix-assembly/src/main/assembly/ servicemix-assembly/src/main/relea...
Modified: incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/component/WSNEndpoint.java
URL: http://svn.apache.org/viewcvs/incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/component/WSNEndpoint.java?rev=379627&r1=379626&r2=379627&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/component/WSNEndpoint.java (original)
+++ incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/component/WSNEndpoint.java Tue Feb 21 15:40:05 2006
@@ -1,238 +1,238 @@
-/*
- * Copyright 2005-2006 The Apache Software Foundation.
- *
- * Licensed 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.servicemix.wsn.component;
-
-import java.io.StringWriter;
-import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Method;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.List;
-
-import javax.jbi.component.ComponentContext;
-import javax.jbi.messaging.DeliveryChannel;
-import javax.jbi.messaging.ExchangeStatus;
-import javax.jbi.messaging.Fault;
-import javax.jbi.messaging.InOnly;
-import javax.jbi.messaging.MessageExchange;
-import javax.jbi.messaging.NormalizedMessage;
-import javax.jbi.messaging.MessageExchange.Role;
-import javax.jbi.servicedesc.ServiceEndpoint;
-import javax.jws.Oneway;
-import javax.jws.WebMethod;
-import javax.jws.WebService;
-import javax.xml.bind.JAXBContext;
-import javax.xml.bind.JAXBException;
-import javax.xml.bind.annotation.XmlRootElement;
-import javax.xml.namespace.QName;
-import javax.xml.ws.WebFault;
-
-import org.apache.servicemix.common.Endpoint;
-import org.apache.servicemix.common.ExchangeProcessor;
-import org.apache.servicemix.jbi.jaxp.StringSource;
-import org.oasis_open.docs.wsrf.bf_2.BaseFaultType;
-
-public class WSNEndpoint extends Endpoint implements ExchangeProcessor {
-
- protected ServiceEndpoint activated;
- protected String address;
- protected Object pojo;
- protected DeliveryChannel channel;
- protected JAXBContext jaxbContext;
- protected Class endpointInterface;
-
- public WSNEndpoint(String address, Object pojo) {
- this.address = address;
- this.pojo = pojo;
- String[] parts = split(address);
- service = new QName(parts[0], parts[1]);
- endpoint = parts[2];
- }
-
- @Override
- public Role getRole() {
- return Role.PROVIDER;
- }
-
- @Override
- public void activate() throws Exception {
- logger = this.serviceUnit.getComponent().getLogger();
- ComponentContext ctx = this.serviceUnit.getComponent().getComponentContext();
- activated = ctx.activateEndpoint(service, endpoint);
- channel = ctx.getDeliveryChannel();
- jaxbContext = createJAXBContext();
- }
-
- protected JAXBContext createJAXBContext() throws Exception {
- WebService ws = getWebServiceAnnotation();
- if (ws == null) {
- throw new IllegalStateException("Unable to find WebService annotation");
- }
- endpointInterface = Class.forName(ws.endpointInterface());
- return createJAXBContext(endpointInterface);
- }
-
- public static JAXBContext createJAXBContext(Class interfaceClass) throws JAXBException {
- List<Class> classes = new ArrayList<Class>();
- classes.add(JbiFault.class);
- for (Method mth : interfaceClass.getMethods()) {
- WebMethod wm = (WebMethod) mth.getAnnotation(WebMethod.class);
- if (wm != null) {
- classes.add(mth.getReturnType());
- classes.addAll(Arrays.asList(mth.getParameterTypes()));
- }
- }
- return JAXBContext.newInstance(classes.toArray(new Class[0]));
- }
-
- @Override
- public void deactivate() throws Exception {
- ServiceEndpoint ep = activated;
- activated = null;
- ComponentContext ctx = this.serviceUnit.getComponent().getComponentContext();
- ctx.deactivateEndpoint(ep);
- }
-
- @Override
- public ExchangeProcessor getProcessor() {
- return this;
- }
-
- @SuppressWarnings("unchecked")
- public void process(MessageExchange exchange) throws Exception {
- if (exchange.getStatus() == ExchangeStatus.DONE) {
- return;
- } else if (exchange.getStatus() == ExchangeStatus.ERROR) {
- exchange.setStatus(ExchangeStatus.DONE);
- channel.send(exchange);
- return;
- }
- Object input = jaxbContext.createUnmarshaller().unmarshal(exchange.getMessage("in").getContent());
- Method webMethod = null;
- for (Method mth : endpointInterface.getMethods()) {
- Class[] params = mth.getParameterTypes();
- if (params.length == 1 && params[0].isAssignableFrom(input.getClass())) {
- webMethod = mth;
- break;
- }
- }
- if (webMethod == null) {
- throw new IllegalStateException("Could not determine invoked web method");
- }
- boolean oneWay = webMethod.getAnnotation(Oneway.class) != null;
- Object output;
- try {
- output = webMethod.invoke(pojo, new Object[] { input });
- } catch (InvocationTargetException e) {
- if (e.getCause() instanceof Exception) {
- WebFault fa = (WebFault) e.getCause().getClass().getAnnotation(WebFault.class);
- if (exchange instanceof InOnly == false && fa != null) {
- BaseFaultType info = (BaseFaultType) e.getCause().getClass().getMethod("getFaultInfo").invoke(e.getCause());
- Fault fault = exchange.createFault();
- exchange.setFault(fault);
- exchange.setError((Exception) e.getCause());
- StringWriter writer = new StringWriter();
- jaxbContext.createMarshaller().marshal(new JbiFault(info), writer);
- fault.setContent(new StringSource(writer.toString()));
- channel.send(exchange);
- return;
- } else {
- throw (Exception) e.getCause();
- }
- } else if (e.getCause() instanceof Error) {
- throw (Error) e.getCause();
- } else {
- throw new RuntimeException(e.getCause());
- }
- }
- if (oneWay) {
- exchange.setStatus(ExchangeStatus.DONE);
- channel.send(exchange);
- } else {
- NormalizedMessage msg = exchange.createMessage();
- exchange.setMessage(msg, "out");
- StringWriter writer = new StringWriter();
- jaxbContext.createMarshaller().marshal(output, writer);
- msg.setContent(new StringSource(writer.toString()));
- channel.send(exchange);
- }
- }
-
- @XmlRootElement(name = "Fault")
- public static class JbiFault {
- private BaseFaultType info;
- public JbiFault() {
- }
- public JbiFault(BaseFaultType info) {
- this.info = info;
- }
- public BaseFaultType getInfo() {
- return info;
- }
- public void setInfo(BaseFaultType info) {
- this.info = info;
- }
- }
-
- protected Method getWebServiceMethod(QName interfaceName, QName operation) throws Exception {
- WebService ws = getWebServiceAnnotation();
- if (ws == null) {
- throw new IllegalStateException("Unable to find WebService annotation");
- }
- Class itf = Class.forName(ws.endpointInterface());
- for (Method mth : itf.getMethods()) {
- WebMethod wm = (WebMethod) mth.getAnnotation(WebMethod.class);
- if (wm != null) {
-
- }
- }
- return null;
- }
-
- @SuppressWarnings("unchecked")
- protected WebService getWebServiceAnnotation() {
- for (Class cl = pojo.getClass(); cl != null; cl = cl.getSuperclass()) {
- WebService ws = (WebService) cl.getAnnotation(WebService.class);
- if (ws != null) {
- return ws;
- }
- }
- return null;
- }
-
- public void start() throws Exception {
- // Nothing to do
- }
-
- public void stop() throws Exception {
- // Nothing to do
- }
-
- protected String[] split(String uri) {
- char sep;
- if (uri.indexOf('/') > 0) {
- sep = '/';
- } else {
- sep = ':';
- }
- int idx1 = uri.lastIndexOf(sep);
- int idx2 = uri.lastIndexOf(sep, idx1 - 1);
- String epName = uri.substring(idx1 + 1);
- String svcName = uri.substring(idx2 + 1, idx1);
- String nsUri = uri.substring(0, idx2);
- return new String[] { nsUri, svcName, epName };
- }
-}
+/*
+ * Copyright 2005-2006 The Apache Software Foundation.
+ *
+ * Licensed 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.servicemix.wsn.component;
+
+import java.io.StringWriter;
+import java.lang.reflect.InvocationTargetException;
+import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+
+import javax.jbi.component.ComponentContext;
+import javax.jbi.messaging.DeliveryChannel;
+import javax.jbi.messaging.ExchangeStatus;
+import javax.jbi.messaging.Fault;
+import javax.jbi.messaging.InOnly;
+import javax.jbi.messaging.MessageExchange;
+import javax.jbi.messaging.NormalizedMessage;
+import javax.jbi.messaging.MessageExchange.Role;
+import javax.jbi.servicedesc.ServiceEndpoint;
+import javax.jws.Oneway;
+import javax.jws.WebMethod;
+import javax.jws.WebService;
+import javax.xml.bind.JAXBContext;
+import javax.xml.bind.JAXBException;
+import javax.xml.bind.annotation.XmlRootElement;
+import javax.xml.namespace.QName;
+import javax.xml.ws.WebFault;
+
+import org.apache.servicemix.common.Endpoint;
+import org.apache.servicemix.common.ExchangeProcessor;
+import org.apache.servicemix.jbi.jaxp.StringSource;
+import org.oasis_open.docs.wsrf.bf_2.BaseFaultType;
+
+public class WSNEndpoint extends Endpoint implements ExchangeProcessor {
+
+ protected ServiceEndpoint activated;
+ protected String address;
+ protected Object pojo;
+ protected DeliveryChannel channel;
+ protected JAXBContext jaxbContext;
+ protected Class endpointInterface;
+
+ public WSNEndpoint(String address, Object pojo) {
+ this.address = address;
+ this.pojo = pojo;
+ String[] parts = split(address);
+ service = new QName(parts[0], parts[1]);
+ endpoint = parts[2];
+ }
+
+ @Override
+ public Role getRole() {
+ return Role.PROVIDER;
+ }
+
+ @Override
+ public void activate() throws Exception {
+ logger = this.serviceUnit.getComponent().getLogger();
+ ComponentContext ctx = this.serviceUnit.getComponent().getComponentContext();
+ activated = ctx.activateEndpoint(service, endpoint);
+ channel = ctx.getDeliveryChannel();
+ jaxbContext = createJAXBContext();
+ }
+
+ protected JAXBContext createJAXBContext() throws Exception {
+ WebService ws = getWebServiceAnnotation();
+ if (ws == null) {
+ throw new IllegalStateException("Unable to find WebService annotation");
+ }
+ endpointInterface = Class.forName(ws.endpointInterface());
+ return createJAXBContext(endpointInterface);
+ }
+
+ public static JAXBContext createJAXBContext(Class interfaceClass) throws JAXBException {
+ List<Class> classes = new ArrayList<Class>();
+ classes.add(JbiFault.class);
+ for (Method mth : interfaceClass.getMethods()) {
+ WebMethod wm = (WebMethod) mth.getAnnotation(WebMethod.class);
+ if (wm != null) {
+ classes.add(mth.getReturnType());
+ classes.addAll(Arrays.asList(mth.getParameterTypes()));
+ }
+ }
+ return JAXBContext.newInstance(classes.toArray(new Class[0]));
+ }
+
+ @Override
+ public void deactivate() throws Exception {
+ ServiceEndpoint ep = activated;
+ activated = null;
+ ComponentContext ctx = this.serviceUnit.getComponent().getComponentContext();
+ ctx.deactivateEndpoint(ep);
+ }
+
+ @Override
+ public ExchangeProcessor getProcessor() {
+ return this;
+ }
+
+ @SuppressWarnings("unchecked")
+ public void process(MessageExchange exchange) throws Exception {
+ if (exchange.getStatus() == ExchangeStatus.DONE) {
+ return;
+ } else if (exchange.getStatus() == ExchangeStatus.ERROR) {
+ exchange.setStatus(ExchangeStatus.DONE);
+ channel.send(exchange);
+ return;
+ }
+ Object input = jaxbContext.createUnmarshaller().unmarshal(exchange.getMessage("in").getContent());
+ Method webMethod = null;
+ for (Method mth : endpointInterface.getMethods()) {
+ Class[] params = mth.getParameterTypes();
+ if (params.length == 1 && params[0].isAssignableFrom(input.getClass())) {
+ webMethod = mth;
+ break;
+ }
+ }
+ if (webMethod == null) {
+ throw new IllegalStateException("Could not determine invoked web method");
+ }
+ boolean oneWay = webMethod.getAnnotation(Oneway.class) != null;
+ Object output;
+ try {
+ output = webMethod.invoke(pojo, new Object[] { input });
+ } catch (InvocationTargetException e) {
+ if (e.getCause() instanceof Exception) {
+ WebFault fa = (WebFault) e.getCause().getClass().getAnnotation(WebFault.class);
+ if (exchange instanceof InOnly == false && fa != null) {
+ BaseFaultType info = (BaseFaultType) e.getCause().getClass().getMethod("getFaultInfo").invoke(e.getCause());
+ Fault fault = exchange.createFault();
+ exchange.setFault(fault);
+ exchange.setError((Exception) e.getCause());
+ StringWriter writer = new StringWriter();
+ jaxbContext.createMarshaller().marshal(new JbiFault(info), writer);
+ fault.setContent(new StringSource(writer.toString()));
+ channel.send(exchange);
+ return;
+ } else {
+ throw (Exception) e.getCause();
+ }
+ } else if (e.getCause() instanceof Error) {
+ throw (Error) e.getCause();
+ } else {
+ throw new RuntimeException(e.getCause());
+ }
+ }
+ if (oneWay) {
+ exchange.setStatus(ExchangeStatus.DONE);
+ channel.send(exchange);
+ } else {
+ NormalizedMessage msg = exchange.createMessage();
+ exchange.setMessage(msg, "out");
+ StringWriter writer = new StringWriter();
+ jaxbContext.createMarshaller().marshal(output, writer);
+ msg.setContent(new StringSource(writer.toString()));
+ channel.send(exchange);
+ }
+ }
+
+ @XmlRootElement(name = "Fault")
+ public static class JbiFault {
+ private BaseFaultType info;
+ public JbiFault() {
+ }
+ public JbiFault(BaseFaultType info) {
+ this.info = info;
+ }
+ public BaseFaultType getInfo() {
+ return info;
+ }
+ public void setInfo(BaseFaultType info) {
+ this.info = info;
+ }
+ }
+
+ protected Method getWebServiceMethod(QName interfaceName, QName operation) throws Exception {
+ WebService ws = getWebServiceAnnotation();
+ if (ws == null) {
+ throw new IllegalStateException("Unable to find WebService annotation");
+ }
+ Class itf = Class.forName(ws.endpointInterface());
+ for (Method mth : itf.getMethods()) {
+ WebMethod wm = (WebMethod) mth.getAnnotation(WebMethod.class);
+ if (wm != null) {
+
+ }
+ }
+ return null;
+ }
+
+ @SuppressWarnings("unchecked")
+ protected WebService getWebServiceAnnotation() {
+ for (Class cl = pojo.getClass(); cl != null; cl = cl.getSuperclass()) {
+ WebService ws = (WebService) cl.getAnnotation(WebService.class);
+ if (ws != null) {
+ return ws;
+ }
+ }
+ return null;
+ }
+
+ public void start() throws Exception {
+ // Nothing to do
+ }
+
+ public void stop() throws Exception {
+ // Nothing to do
+ }
+
+ protected String[] split(String uri) {
+ char sep;
+ if (uri.indexOf('/') > 0) {
+ sep = '/';
+ } else {
+ sep = ':';
+ }
+ int idx1 = uri.lastIndexOf(sep);
+ int idx2 = uri.lastIndexOf(sep, idx1 - 1);
+ String epName = uri.substring(idx1 + 1);
+ String svcName = uri.substring(idx2 + 1, idx1);
+ String nsUri = uri.substring(0, idx2);
+ return new String[] { nsUri, svcName, epName };
+ }
+}
Propchange: incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/component/WSNEndpoint.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/component/WSNLifeCycle.java
URL: http://svn.apache.org/viewcvs/incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/component/WSNLifeCycle.java?rev=379627&r1=379626&r2=379627&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/component/WSNLifeCycle.java (original)
+++ incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/component/WSNLifeCycle.java Tue Feb 21 15:40:05 2006
@@ -1,157 +1,157 @@
-/*
- * Copyright 2005-2006 The Apache Software Foundation.
- *
- * Licensed 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.servicemix.wsn.component;
-
-import java.util.Hashtable;
-
-import javax.jms.ConnectionFactory;
-import javax.naming.Context;
-import javax.naming.InitialContext;
-import javax.naming.NamingException;
-
-import org.apache.servicemix.common.BaseComponent;
-import org.apache.servicemix.common.BaseLifeCycle;
-import org.apache.servicemix.common.ServiceUnit;
-import org.apache.servicemix.wsn.EndpointManager;
-import org.apache.servicemix.wsn.EndpointRegistrationException;
-import org.apache.servicemix.wsn.jbi.JbiNotificationBroker;
-import org.apache.servicemix.wsn.jms.JmsCreatePullPoint;
-
-public class WSNLifeCycle extends BaseLifeCycle {
-
- private JbiNotificationBroker notificationBroker;
- private JmsCreatePullPoint createPullPoint;
- private WSNConfiguration configuration;
- private ConnectionFactory connectionFactory;
- private ServiceUnit serviceUnit;
-
- public WSNLifeCycle(BaseComponent component) {
- super(component);
- configuration = new WSNConfiguration();
- serviceUnit = new ServiceUnit();
- serviceUnit.setComponent(component);
- }
-
- protected Object getExtensionMBean() throws Exception {
- return configuration;
- }
-
- @Override
- protected void doInit() throws Exception {
- super.doInit();
- // Notification Broker
- notificationBroker = new JbiNotificationBroker(configuration.getBrokerName());
- notificationBroker.setContext(context);
- notificationBroker.setManager(new WSNEndpointManager());
- if (connectionFactory == null) {
- connectionFactory = lookupConnectionFactory();
- }
- notificationBroker.setConnectionFactory(connectionFactory);
- notificationBroker.init();
- // Create PullPoint
- createPullPoint = new JmsCreatePullPoint(configuration.getBrokerName());
- createPullPoint.setManager(new WSNEndpointManager());
- if (connectionFactory == null) {
- connectionFactory = lookupConnectionFactory();
- }
- createPullPoint.setConnectionFactory(connectionFactory);
- createPullPoint.init();
- }
-
- @Override
- protected void doShutDown() throws Exception {
- // TODO Auto-generated method stub
- super.doShutDown();
- }
-
- @Override
- protected void doStart() throws Exception {
- super.doStart();
- }
-
- @Override
- protected void doStop() throws Exception {
- // TODO Auto-generated method stub
- super.doStop();
- }
-
- public WSNConfiguration getConfiguration() {
- return configuration;
- }
-
- public void setConfiguration(WSNConfiguration configuration) {
- this.configuration = configuration;
- }
-
- public ConnectionFactory getConnectionFactory() {
- return connectionFactory;
- }
-
- public void setConnectionFactory(ConnectionFactory connectionFactory) {
- this.connectionFactory = connectionFactory;
- }
-
- protected ConnectionFactory lookupConnectionFactory() throws NamingException {
- Hashtable<String,String> props = new Hashtable<String,String>();
- if (configuration.getInitialContextFactory() != null && configuration.getJndiProviderURL() != null) {
- props.put(Context.INITIAL_CONTEXT_FACTORY, configuration.getInitialContextFactory());
- props.put(Context.PROVIDER_URL, configuration.getJndiProviderURL());
- }
- InitialContext ctx = new InitialContext(props);
- ConnectionFactory connectionFactory = (ConnectionFactory) ctx.lookup(configuration.getJndiConnectionFactoryName());
- return connectionFactory;
- }
-
- public class WSNEndpointManager implements EndpointManager {
-
- public Object register(String address, Object service) throws EndpointRegistrationException {
- component.getRegistry().unregisterServiceUnit(serviceUnit);
- try {
- WSNEndpoint endpoint = new WSNEndpoint(address, service);
- endpoint.setServiceUnit(serviceUnit);
- endpoint.activate();
- serviceUnit.addEndpoint(endpoint);
- return endpoint;
- } catch (Exception e) {
- throw new EndpointRegistrationException("Unable to activate endpoint", e);
- } finally {
- component.getRegistry().registerServiceUnit(serviceUnit);
- }
- }
-
- public void unregister(Object endpoint) throws EndpointRegistrationException {
- component.getRegistry().unregisterServiceUnit(serviceUnit);
- try {
- ((WSNEndpoint) endpoint).deactivate();
- } catch (Exception e) {
- throw new EndpointRegistrationException("Unable to activate endpoint", e);
- } finally {
- serviceUnit.getEndpoints().remove(endpoint);
- component.getRegistry().registerServiceUnit(serviceUnit);
- }
- }
-
- }
-
- public JbiNotificationBroker getNotificationBroker() {
- return notificationBroker;
- }
-
- public JmsCreatePullPoint getCreatePullPoint() {
- return createPullPoint;
- }
-
-}
+/*
+ * Copyright 2005-2006 The Apache Software Foundation.
+ *
+ * Licensed 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.servicemix.wsn.component;
+
+import java.util.Hashtable;
+
+import javax.jms.ConnectionFactory;
+import javax.naming.Context;
+import javax.naming.InitialContext;
+import javax.naming.NamingException;
+
+import org.apache.servicemix.common.BaseComponent;
+import org.apache.servicemix.common.BaseLifeCycle;
+import org.apache.servicemix.common.ServiceUnit;
+import org.apache.servicemix.wsn.EndpointManager;
+import org.apache.servicemix.wsn.EndpointRegistrationException;
+import org.apache.servicemix.wsn.jbi.JbiNotificationBroker;
+import org.apache.servicemix.wsn.jms.JmsCreatePullPoint;
+
+public class WSNLifeCycle extends BaseLifeCycle {
+
+ private JbiNotificationBroker notificationBroker;
+ private JmsCreatePullPoint createPullPoint;
+ private WSNConfiguration configuration;
+ private ConnectionFactory connectionFactory;
+ private ServiceUnit serviceUnit;
+
+ public WSNLifeCycle(BaseComponent component) {
+ super(component);
+ configuration = new WSNConfiguration();
+ serviceUnit = new ServiceUnit();
+ serviceUnit.setComponent(component);
+ }
+
+ protected Object getExtensionMBean() throws Exception {
+ return configuration;
+ }
+
+ @Override
+ protected void doInit() throws Exception {
+ super.doInit();
+ // Notification Broker
+ notificationBroker = new JbiNotificationBroker(configuration.getBrokerName());
+ notificationBroker.setContext(context);
+ notificationBroker.setManager(new WSNEndpointManager());
+ if (connectionFactory == null) {
+ connectionFactory = lookupConnectionFactory();
+ }
+ notificationBroker.setConnectionFactory(connectionFactory);
+ notificationBroker.init();
+ // Create PullPoint
+ createPullPoint = new JmsCreatePullPoint(configuration.getBrokerName());
+ createPullPoint.setManager(new WSNEndpointManager());
+ if (connectionFactory == null) {
+ connectionFactory = lookupConnectionFactory();
+ }
+ createPullPoint.setConnectionFactory(connectionFactory);
+ createPullPoint.init();
+ }
+
+ @Override
+ protected void doShutDown() throws Exception {
+ // TODO Auto-generated method stub
+ super.doShutDown();
+ }
+
+ @Override
+ protected void doStart() throws Exception {
+ super.doStart();
+ }
+
+ @Override
+ protected void doStop() throws Exception {
+ // TODO Auto-generated method stub
+ super.doStop();
+ }
+
+ public WSNConfiguration getConfiguration() {
+ return configuration;
+ }
+
+ public void setConfiguration(WSNConfiguration configuration) {
+ this.configuration = configuration;
+ }
+
+ public ConnectionFactory getConnectionFactory() {
+ return connectionFactory;
+ }
+
+ public void setConnectionFactory(ConnectionFactory connectionFactory) {
+ this.connectionFactory = connectionFactory;
+ }
+
+ protected ConnectionFactory lookupConnectionFactory() throws NamingException {
+ Hashtable<String,String> props = new Hashtable<String,String>();
+ if (configuration.getInitialContextFactory() != null && configuration.getJndiProviderURL() != null) {
+ props.put(Context.INITIAL_CONTEXT_FACTORY, configuration.getInitialContextFactory());
+ props.put(Context.PROVIDER_URL, configuration.getJndiProviderURL());
+ }
+ InitialContext ctx = new InitialContext(props);
+ ConnectionFactory connectionFactory = (ConnectionFactory) ctx.lookup(configuration.getJndiConnectionFactoryName());
+ return connectionFactory;
+ }
+
+ public class WSNEndpointManager implements EndpointManager {
+
+ public Object register(String address, Object service) throws EndpointRegistrationException {
+ component.getRegistry().unregisterServiceUnit(serviceUnit);
+ try {
+ WSNEndpoint endpoint = new WSNEndpoint(address, service);
+ endpoint.setServiceUnit(serviceUnit);
+ endpoint.activate();
+ serviceUnit.addEndpoint(endpoint);
+ return endpoint;
+ } catch (Exception e) {
+ throw new EndpointRegistrationException("Unable to activate endpoint", e);
+ } finally {
+ component.getRegistry().registerServiceUnit(serviceUnit);
+ }
+ }
+
+ public void unregister(Object endpoint) throws EndpointRegistrationException {
+ component.getRegistry().unregisterServiceUnit(serviceUnit);
+ try {
+ ((WSNEndpoint) endpoint).deactivate();
+ } catch (Exception e) {
+ throw new EndpointRegistrationException("Unable to activate endpoint", e);
+ } finally {
+ serviceUnit.getEndpoints().remove(endpoint);
+ component.getRegistry().registerServiceUnit(serviceUnit);
+ }
+ }
+
+ }
+
+ public JbiNotificationBroker getNotificationBroker() {
+ return notificationBroker;
+ }
+
+ public JmsCreatePullPoint getCreatePullPoint() {
+ return createPullPoint;
+ }
+
+}
Propchange: incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/component/WSNLifeCycle.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/component/package.html
URL: http://svn.apache.org/viewcvs/incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/component/package.html?rev=379627&r1=379626&r2=379627&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/component/package.html (original)
+++ incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/component/package.html Tue Feb 21 15:40:05 2006
@@ -1,9 +1,9 @@
-<html>
-<head>
-</head>
-<body>
-
-JBI Service Engine implementing WS-Notification 1.3.
-
-</body>
-</html>
+<html>
+<head>
+</head>
+<body>
+
+JBI Service Engine implementing WS-Notification 1.3.
+
+</body>
+</html>
Propchange: incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/component/package.html
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jbi/JbiNotificationBroker.java
URL: http://svn.apache.org/viewcvs/incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jbi/JbiNotificationBroker.java?rev=379627&r1=379626&r2=379627&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jbi/JbiNotificationBroker.java (original)
+++ incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jbi/JbiNotificationBroker.java Tue Feb 21 15:40:05 2006
@@ -1,55 +1,55 @@
-/*
- * Copyright 2005-2006 The Apache Software Foundation.
- *
- * Licensed 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.servicemix.wsn.jbi;
-
-import javax.jbi.component.ComponentContext;
-
-import org.apache.servicemix.wsn.jms.JmsNotificationBroker;
-import org.apache.servicemix.wsn.jms.JmsPublisher;
-import org.apache.servicemix.wsn.jms.JmsSubscription;
-
-public class JbiNotificationBroker extends JmsNotificationBroker {
-
- private ComponentContext context;
-
- public JbiNotificationBroker(String name) {
- super(name);
- }
-
- @Override
- protected JmsSubscription createJmsSubscription(String name) {
- JbiSubscription subscription = new JbiSubscription(name);
- subscription.setContext(context);
- return subscription;
- }
-
- @Override
- protected JmsPublisher createJmsPublisher(String name) {
- JbiPublisher publisher = new JbiPublisher(name);
- publisher.setContext(context);
- publisher.setNotificationBrokerAddress(address);
- return publisher;
- }
-
- public ComponentContext getContext() {
- return context;
- }
-
- public void setContext(ComponentContext context) {
- this.context = context;
- }
-
-}
+/*
+ * Copyright 2005-2006 The Apache Software Foundation.
+ *
+ * Licensed 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.servicemix.wsn.jbi;
+
+import javax.jbi.component.ComponentContext;
+
+import org.apache.servicemix.wsn.jms.JmsNotificationBroker;
+import org.apache.servicemix.wsn.jms.JmsPublisher;
+import org.apache.servicemix.wsn.jms.JmsSubscription;
+
+public class JbiNotificationBroker extends JmsNotificationBroker {
+
+ private ComponentContext context;
+
+ public JbiNotificationBroker(String name) {
+ super(name);
+ }
+
+ @Override
+ protected JmsSubscription createJmsSubscription(String name) {
+ JbiSubscription subscription = new JbiSubscription(name);
+ subscription.setContext(context);
+ return subscription;
+ }
+
+ @Override
+ protected JmsPublisher createJmsPublisher(String name) {
+ JbiPublisher publisher = new JbiPublisher(name);
+ publisher.setContext(context);
+ publisher.setNotificationBrokerAddress(address);
+ return publisher;
+ }
+
+ public ComponentContext getContext() {
+ return context;
+ }
+
+ public void setContext(ComponentContext context) {
+ this.context = context;
+ }
+
+}
Propchange: incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jbi/JbiNotificationBroker.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jbi/JbiPublisher.java
URL: http://svn.apache.org/viewcvs/incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jbi/JbiPublisher.java?rev=379627&r1=379626&r2=379627&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jbi/JbiPublisher.java (original)
+++ incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jbi/JbiPublisher.java Tue Feb 21 15:40:05 2006
@@ -1,112 +1,112 @@
-/*
- * Copyright 2005-2006 The Apache Software Foundation.
- *
- * Licensed 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.servicemix.wsn.jbi;
-
-import javax.jbi.JBIException;
-import javax.jbi.component.ComponentContext;
-import javax.jbi.servicedesc.ServiceEndpoint;
-import javax.xml.bind.JAXBException;
-import javax.xml.namespace.QName;
-
-import org.apache.servicemix.wsn.client.AbstractWSAClient;
-import org.apache.servicemix.wsn.client.NotificationBroker;
-import org.apache.servicemix.wsn.client.Subscription;
-import org.apache.servicemix.wsn.jaxws.InvalidTopicExpressionFault;
-import org.apache.servicemix.wsn.jaxws.PublisherRegistrationFailedFault;
-import org.apache.servicemix.wsn.jaxws.PublisherRegistrationRejectedFault;
-import org.apache.servicemix.wsn.jaxws.ResourceUnknownFault;
-import org.apache.servicemix.wsn.jaxws.TopicNotSupportedFault;
-import org.apache.servicemix.wsn.jms.JmsPublisher;
-import org.oasis_open.docs.wsn.br_2.PublisherRegistrationFailedFaultType;
-import org.oasis_open.docs.wsn.br_2.RegisterPublisher;
-
-public class JbiPublisher extends JmsPublisher {
-
- private ComponentContext context;
- private ServiceEndpoint endpoint;
- private String notificationBrokerAddress;
-
- public JbiPublisher(String name) {
- super(name);
- }
-
- public void setContext(ComponentContext context) {
- this.context = context;
- }
-
- public String getNotificationBrokerAddress() {
- return notificationBrokerAddress;
- }
-
- public void setNotificationBrokerAddress(String notificationBrokerAddress) {
- this.notificationBrokerAddress = notificationBrokerAddress;
- }
-
- @Override
- protected Object startSubscription() {
- Subscription subscription = null;
- try {
- NotificationBroker broker = new NotificationBroker(context);
- broker.setResolver(AbstractWSAClient.resolveWSA(publisherReference));
- subscription = broker.subscribe(AbstractWSAClient.createWSA(notificationBrokerAddress),
- "noTopic", null);
- } catch (JBIException e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
- } catch (JAXBException e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
- }
- return subscription;
- }
-
- @Override
- protected void destroySubscription(Object subscription) {
- try {
- ((Subscription) subscription).unsubscribe();
- } catch (JBIException e) {
- // TODO Auto-generated catch block
- e.printStackTrace();
- }
- }
-
- @Override
- protected void validatePublisher(RegisterPublisher registerPublisherRequest) throws InvalidTopicExpressionFault, PublisherRegistrationFailedFault, PublisherRegistrationRejectedFault, ResourceUnknownFault, TopicNotSupportedFault {
- super.validatePublisher(registerPublisherRequest);
- String[] parts = split(publisherReference.getAddress().getValue());
- endpoint = context.getEndpoint(new QName(parts[0], parts[1]), parts[2]);
- if (endpoint == null) {
- PublisherRegistrationFailedFaultType fault = new PublisherRegistrationFailedFaultType();
- throw new PublisherRegistrationFailedFault("Unable to resolve consumer reference endpoint", fault);
- }
- }
-
- protected String[] split(String uri) {
- char sep;
- if (uri.indexOf('/') > 0) {
- sep = '/';
- } else {
- sep = ':';
- }
- int idx1 = uri.lastIndexOf(sep);
- int idx2 = uri.lastIndexOf(sep, idx1 - 1);
- String epName = uri.substring(idx1 + 1);
- String svcName = uri.substring(idx2 + 1, idx1);
- String nsUri = uri.substring(0, idx2);
- return new String[] { nsUri, svcName, epName };
- }
-
-}
+/*
+ * Copyright 2005-2006 The Apache Software Foundation.
+ *
+ * Licensed 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.servicemix.wsn.jbi;
+
+import javax.jbi.JBIException;
+import javax.jbi.component.ComponentContext;
+import javax.jbi.servicedesc.ServiceEndpoint;
+import javax.xml.bind.JAXBException;
+import javax.xml.namespace.QName;
+
+import org.apache.servicemix.wsn.client.AbstractWSAClient;
+import org.apache.servicemix.wsn.client.NotificationBroker;
+import org.apache.servicemix.wsn.client.Subscription;
+import org.apache.servicemix.wsn.jaxws.InvalidTopicExpressionFault;
+import org.apache.servicemix.wsn.jaxws.PublisherRegistrationFailedFault;
+import org.apache.servicemix.wsn.jaxws.PublisherRegistrationRejectedFault;
+import org.apache.servicemix.wsn.jaxws.ResourceUnknownFault;
+import org.apache.servicemix.wsn.jaxws.TopicNotSupportedFault;
+import org.apache.servicemix.wsn.jms.JmsPublisher;
+import org.oasis_open.docs.wsn.br_2.PublisherRegistrationFailedFaultType;
+import org.oasis_open.docs.wsn.br_2.RegisterPublisher;
+
+public class JbiPublisher extends JmsPublisher {
+
+ private ComponentContext context;
+ private ServiceEndpoint endpoint;
+ private String notificationBrokerAddress;
+
+ public JbiPublisher(String name) {
+ super(name);
+ }
+
+ public void setContext(ComponentContext context) {
+ this.context = context;
+ }
+
+ public String getNotificationBrokerAddress() {
+ return notificationBrokerAddress;
+ }
+
+ public void setNotificationBrokerAddress(String notificationBrokerAddress) {
+ this.notificationBrokerAddress = notificationBrokerAddress;
+ }
+
+ @Override
+ protected Object startSubscription() {
+ Subscription subscription = null;
+ try {
+ NotificationBroker broker = new NotificationBroker(context);
+ broker.setResolver(AbstractWSAClient.resolveWSA(publisherReference));
+ subscription = broker.subscribe(AbstractWSAClient.createWSA(notificationBrokerAddress),
+ "noTopic", null);
+ } catch (JBIException e) {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
+ } catch (JAXBException e) {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
+ }
+ return subscription;
+ }
+
+ @Override
+ protected void destroySubscription(Object subscription) {
+ try {
+ ((Subscription) subscription).unsubscribe();
+ } catch (JBIException e) {
+ // TODO Auto-generated catch block
+ e.printStackTrace();
+ }
+ }
+
+ @Override
+ protected void validatePublisher(RegisterPublisher registerPublisherRequest) throws InvalidTopicExpressionFault, PublisherRegistrationFailedFault, PublisherRegistrationRejectedFault, ResourceUnknownFault, TopicNotSupportedFault {
+ super.validatePublisher(registerPublisherRequest);
+ String[] parts = split(publisherReference.getAddress().getValue());
+ endpoint = context.getEndpoint(new QName(parts[0], parts[1]), parts[2]);
+ if (endpoint == null) {
+ PublisherRegistrationFailedFaultType fault = new PublisherRegistrationFailedFaultType();
+ throw new PublisherRegistrationFailedFault("Unable to resolve consumer reference endpoint", fault);
+ }
+ }
+
+ protected String[] split(String uri) {
+ char sep;
+ if (uri.indexOf('/') > 0) {
+ sep = '/';
+ } else {
+ sep = ':';
+ }
+ int idx1 = uri.lastIndexOf(sep);
+ int idx2 = uri.lastIndexOf(sep, idx1 - 1);
+ String epName = uri.substring(idx1 + 1);
+ String svcName = uri.substring(idx2 + 1, idx1);
+ String nsUri = uri.substring(0, idx2);
+ return new String[] { nsUri, svcName, epName };
+ }
+
+}
Propchange: incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jbi/JbiPublisher.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jbi/JbiSubscription.java
URL: http://svn.apache.org/viewcvs/incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jbi/JbiSubscription.java?rev=379627&r1=379626&r2=379627&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jbi/JbiSubscription.java (original)
+++ incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jbi/JbiSubscription.java Tue Feb 21 15:40:05 2006
@@ -1,111 +1,111 @@
-/*
- * Copyright 2005-2006 The Apache Software Foundation.
- *
- * Licensed 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.servicemix.wsn.jbi;
-
-import javax.jbi.JBIException;
-import javax.jbi.component.ComponentContext;
-import javax.jbi.messaging.DeliveryChannel;
-import javax.jbi.messaging.InOnly;
-import javax.jbi.messaging.MessageExchangeFactory;
-import javax.jbi.messaging.NormalizedMessage;
-import javax.jbi.servicedesc.ServiceEndpoint;
-import javax.xml.namespace.QName;
-import javax.xml.transform.dom.DOMSource;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.servicemix.wsn.jaxws.InvalidFilterFault;
-import org.apache.servicemix.wsn.jaxws.InvalidMessageContentExpressionFault;
-import org.apache.servicemix.wsn.jaxws.InvalidProducerPropertiesExpressionFault;
-import org.apache.servicemix.wsn.jaxws.InvalidTopicExpressionFault;
-import org.apache.servicemix.wsn.jaxws.SubscribeCreationFailedFault;
-import org.apache.servicemix.wsn.jaxws.TopicExpressionDialectUnknownFault;
-import org.apache.servicemix.wsn.jaxws.TopicNotSupportedFault;
-import org.apache.servicemix.wsn.jaxws.UnacceptableInitialTerminationTimeFault;
-import org.apache.servicemix.wsn.jms.JmsSubscription;
-import org.oasis_open.docs.wsn.b_2.Subscribe;
-import org.oasis_open.docs.wsn.b_2.SubscribeCreationFailedFaultType;
-import org.w3c.dom.Element;
-
-public class JbiSubscription extends JmsSubscription {
-
- private static Log log = LogFactory.getLog(JbiSubscription.class);
-
- private ComponentContext context;
- private ServiceEndpoint endpoint;
-
- public JbiSubscription(String name) {
- super(name);
- }
-
- @Override
- protected void start() throws SubscribeCreationFailedFault {
- super.start();
- }
-
- @Override
- protected void validateSubscription(Subscribe subscribeRequest) throws InvalidFilterFault, InvalidMessageContentExpressionFault, InvalidProducerPropertiesExpressionFault, InvalidTopicExpressionFault, SubscribeCreationFailedFault, TopicExpressionDialectUnknownFault, TopicNotSupportedFault, UnacceptableInitialTerminationTimeFault {
- super.validateSubscription(subscribeRequest);
- String[] parts = split(consumerReference.getAddress().getValue().trim());
- endpoint = context.getEndpoint(new QName(parts[0], parts[1]), parts[2]);
- if (endpoint == null) {
- SubscribeCreationFailedFaultType fault = new SubscribeCreationFailedFaultType();
- throw new SubscribeCreationFailedFault("Unable to resolve consumer reference endpoint", fault);
- }
- }
-
- protected String[] split(String uri) {
- char sep;
- if (uri.indexOf('/') > 0) {
- sep = '/';
- } else {
- sep = ':';
- }
- int idx1 = uri.lastIndexOf(sep);
- int idx2 = uri.lastIndexOf(sep, idx1 - 1);
- String epName = uri.substring(idx1 + 1);
- String svcName = uri.substring(idx2 + 1, idx1);
- String nsUri = uri.substring(0, idx2);
- return new String[] { nsUri, svcName, epName };
- }
-
- @Override
- protected void doNotify(Element content) {
- try {
- DeliveryChannel channel = context.getDeliveryChannel();
- MessageExchangeFactory factory = channel.createExchangeFactory(endpoint);
- InOnly inonly = factory.createInOnlyExchange();
- NormalizedMessage msg = inonly.createMessage();
- inonly.setInMessage(msg);
- msg.setContent(new DOMSource(content));
- if (!channel.sendSync(inonly)) {
- log.warn("Notification was aborted");
- }
- } catch (JBIException e) {
- log.warn("Could not deliver notification", e);
- }
- }
-
- public ComponentContext getContext() {
- return context;
- }
-
- public void setContext(ComponentContext context) {
- this.context = context;
- }
-
-
-}
+/*
+ * Copyright 2005-2006 The Apache Software Foundation.
+ *
+ * Licensed 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.servicemix.wsn.jbi;
+
+import javax.jbi.JBIException;
+import javax.jbi.component.ComponentContext;
+import javax.jbi.messaging.DeliveryChannel;
+import javax.jbi.messaging.InOnly;
+import javax.jbi.messaging.MessageExchangeFactory;
+import javax.jbi.messaging.NormalizedMessage;
+import javax.jbi.servicedesc.ServiceEndpoint;
+import javax.xml.namespace.QName;
+import javax.xml.transform.dom.DOMSource;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.servicemix.wsn.jaxws.InvalidFilterFault;
+import org.apache.servicemix.wsn.jaxws.InvalidMessageContentExpressionFault;
+import org.apache.servicemix.wsn.jaxws.InvalidProducerPropertiesExpressionFault;
+import org.apache.servicemix.wsn.jaxws.InvalidTopicExpressionFault;
+import org.apache.servicemix.wsn.jaxws.SubscribeCreationFailedFault;
+import org.apache.servicemix.wsn.jaxws.TopicExpressionDialectUnknownFault;
+import org.apache.servicemix.wsn.jaxws.TopicNotSupportedFault;
+import org.apache.servicemix.wsn.jaxws.UnacceptableInitialTerminationTimeFault;
+import org.apache.servicemix.wsn.jms.JmsSubscription;
+import org.oasis_open.docs.wsn.b_2.Subscribe;
+import org.oasis_open.docs.wsn.b_2.SubscribeCreationFailedFaultType;
+import org.w3c.dom.Element;
+
+public class JbiSubscription extends JmsSubscription {
+
+ private static Log log = LogFactory.getLog(JbiSubscription.class);
+
+ private ComponentContext context;
+ private ServiceEndpoint endpoint;
+
+ public JbiSubscription(String name) {
+ super(name);
+ }
+
+ @Override
+ protected void start() throws SubscribeCreationFailedFault {
+ super.start();
+ }
+
+ @Override
+ protected void validateSubscription(Subscribe subscribeRequest) throws InvalidFilterFault, InvalidMessageContentExpressionFault, InvalidProducerPropertiesExpressionFault, InvalidTopicExpressionFault, SubscribeCreationFailedFault, TopicExpressionDialectUnknownFault, TopicNotSupportedFault, UnacceptableInitialTerminationTimeFault {
+ super.validateSubscription(subscribeRequest);
+ String[] parts = split(consumerReference.getAddress().getValue().trim());
+ endpoint = context.getEndpoint(new QName(parts[0], parts[1]), parts[2]);
+ if (endpoint == null) {
+ SubscribeCreationFailedFaultType fault = new SubscribeCreationFailedFaultType();
+ throw new SubscribeCreationFailedFault("Unable to resolve consumer reference endpoint", fault);
+ }
+ }
+
+ protected String[] split(String uri) {
+ char sep;
+ if (uri.indexOf('/') > 0) {
+ sep = '/';
+ } else {
+ sep = ':';
+ }
+ int idx1 = uri.lastIndexOf(sep);
+ int idx2 = uri.lastIndexOf(sep, idx1 - 1);
+ String epName = uri.substring(idx1 + 1);
+ String svcName = uri.substring(idx2 + 1, idx1);
+ String nsUri = uri.substring(0, idx2);
+ return new String[] { nsUri, svcName, epName };
+ }
+
+ @Override
+ protected void doNotify(Element content) {
+ try {
+ DeliveryChannel channel = context.getDeliveryChannel();
+ MessageExchangeFactory factory = channel.createExchangeFactory(endpoint);
+ InOnly inonly = factory.createInOnlyExchange();
+ NormalizedMessage msg = inonly.createMessage();
+ inonly.setInMessage(msg);
+ msg.setContent(new DOMSource(content));
+ if (!channel.sendSync(inonly)) {
+ log.warn("Notification was aborted");
+ }
+ } catch (JBIException e) {
+ log.warn("Could not deliver notification", e);
+ }
+ }
+
+ public ComponentContext getContext() {
+ return context;
+ }
+
+ public void setContext(ComponentContext context) {
+ this.context = context;
+ }
+
+
+}
Propchange: incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jbi/JbiSubscription.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jbi/package.html
URL: http://svn.apache.org/viewcvs/incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jbi/package.html?rev=379627&r1=379626&r2=379627&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jbi/package.html (original)
+++ incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jbi/package.html Tue Feb 21 15:40:05 2006
@@ -1,11 +1,11 @@
-<html>
-<head>
-</head>
-<body>
-
-JBI implementation of NotificationBroker and related services.
-This implementation extends the JMS implemetation to provide
-JBI specific features for sending notifications.
-
-</body>
-</html>
+<html>
+<head>
+</head>
+<body>
+
+JBI implementation of NotificationBroker and related services.
+This implementation extends the JMS implemetation to provide
+JBI specific features for sending notifications.
+
+</body>
+</html>
Propchange: incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jbi/package.html
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jms/InvalidTopicException.java
URL: http://svn.apache.org/viewcvs/incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jms/InvalidTopicException.java?rev=379627&r1=379626&r2=379627&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jms/InvalidTopicException.java (original)
+++ incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jms/InvalidTopicException.java Tue Feb 21 15:40:05 2006
@@ -1,38 +1,38 @@
-/*
- * Copyright 2005-2006 The Apache Software Foundation.
- *
- * Licensed 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.servicemix.wsn.jms;
-
-public class InvalidTopicException extends Exception {
-
- private static final long serialVersionUID = -3708397351142080702L;
-
- public InvalidTopicException() {
- super();
- }
-
- public InvalidTopicException(String message, Throwable cause) {
- super(message, cause);
- }
-
- public InvalidTopicException(String message) {
- super(message);
- }
-
- public InvalidTopicException(Throwable cause) {
- super(cause);
- }
-
-}
+/*
+ * Copyright 2005-2006 The Apache Software Foundation.
+ *
+ * Licensed 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.servicemix.wsn.jms;
+
+public class InvalidTopicException extends Exception {
+
+ private static final long serialVersionUID = -3708397351142080702L;
+
+ public InvalidTopicException() {
+ super();
+ }
+
+ public InvalidTopicException(String message, Throwable cause) {
+ super(message, cause);
+ }
+
+ public InvalidTopicException(String message) {
+ super(message);
+ }
+
+ public InvalidTopicException(Throwable cause) {
+ super(cause);
+ }
+
+}
Propchange: incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jms/InvalidTopicException.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jms/JmsCreatePullPoint.java
URL: http://svn.apache.org/viewcvs/incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jms/JmsCreatePullPoint.java?rev=379627&r1=379626&r2=379627&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jms/JmsCreatePullPoint.java (original)
+++ incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jms/JmsCreatePullPoint.java Tue Feb 21 15:40:05 2006
@@ -1,57 +1,57 @@
-/*
- * Copyright 2005-2006 The Apache Software Foundation.
- *
- * Licensed 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.servicemix.wsn.jms;
-
-import javax.jms.Connection;
-import javax.jms.ConnectionFactory;
-
-import org.apache.servicemix.wsn.AbstractCreatePullPoint;
-import org.apache.servicemix.wsn.AbstractPullPoint;
-
-public class JmsCreatePullPoint extends AbstractCreatePullPoint {
-
- private ConnectionFactory connectionFactory;
- private Connection connection;
-
- public JmsCreatePullPoint(String name) {
- super(name);
- }
-
- public void init() throws Exception {
- if (connection == null) {
- connection = connectionFactory.createConnection();
- connection.start();
- }
- super.init();
- }
-
- @Override
- protected AbstractPullPoint createPullPoint(String name) {
- JmsPullPoint pullPoint = new JmsPullPoint(name);
- pullPoint.setManager(getManager());
- pullPoint.setConnection(connection);
- return pullPoint;
- }
-
- public ConnectionFactory getConnectionFactory() {
- return connectionFactory;
- }
-
- public void setConnectionFactory(ConnectionFactory connectionFactory) {
- this.connectionFactory = connectionFactory;
- }
-
-}
+/*
+ * Copyright 2005-2006 The Apache Software Foundation.
+ *
+ * Licensed 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.servicemix.wsn.jms;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+
+import org.apache.servicemix.wsn.AbstractCreatePullPoint;
+import org.apache.servicemix.wsn.AbstractPullPoint;
+
+public class JmsCreatePullPoint extends AbstractCreatePullPoint {
+
+ private ConnectionFactory connectionFactory;
+ private Connection connection;
+
+ public JmsCreatePullPoint(String name) {
+ super(name);
+ }
+
+ public void init() throws Exception {
+ if (connection == null) {
+ connection = connectionFactory.createConnection();
+ connection.start();
+ }
+ super.init();
+ }
+
+ @Override
+ protected AbstractPullPoint createPullPoint(String name) {
+ JmsPullPoint pullPoint = new JmsPullPoint(name);
+ pullPoint.setManager(getManager());
+ pullPoint.setConnection(connection);
+ return pullPoint;
+ }
+
+ public ConnectionFactory getConnectionFactory() {
+ return connectionFactory;
+ }
+
+ public void setConnectionFactory(ConnectionFactory connectionFactory) {
+ this.connectionFactory = connectionFactory;
+ }
+
+}
Propchange: incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jms/JmsCreatePullPoint.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jms/JmsNotificationBroker.java
URL: http://svn.apache.org/viewcvs/incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jms/JmsNotificationBroker.java?rev=379627&r1=379626&r2=379627&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jms/JmsNotificationBroker.java (original)
+++ incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jms/JmsNotificationBroker.java Tue Feb 21 15:40:05 2006
@@ -1,70 +1,70 @@
-/*
- * Copyright 2005-2006 The Apache Software Foundation.
- *
- * Licensed 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.servicemix.wsn.jms;
-
-import javax.jms.Connection;
-import javax.jms.ConnectionFactory;
-
-import org.apache.servicemix.wsn.AbstractNotificationBroker;
-import org.apache.servicemix.wsn.AbstractPublisher;
-import org.apache.servicemix.wsn.AbstractSubscription;
-
-public abstract class JmsNotificationBroker extends AbstractNotificationBroker {
-
- private ConnectionFactory connectionFactory;
- private Connection connection;
-
- public JmsNotificationBroker(String name) {
- super(name);
- }
-
- public void init() throws Exception {
- if (connection == null) {
- connection = connectionFactory.createConnection();
- connection.start();
- }
- super.init();
- }
-
- @Override
- protected AbstractPublisher createPublisher(String name) {
- JmsPublisher publisher = createJmsPublisher(name);
- publisher.setManager(getManager());
- publisher.setConnection(connection);
- return publisher;
- }
-
- @Override
- protected AbstractSubscription createSubcription(String name) {
- JmsSubscription subscription = createJmsSubscription(name);
- subscription.setManager(getManager());
- subscription.setConnection(connection);
- return subscription;
- }
-
- protected abstract JmsSubscription createJmsSubscription(String name);
-
- protected abstract JmsPublisher createJmsPublisher(String name);
-
- public ConnectionFactory getConnectionFactory() {
- return connectionFactory;
- }
-
- public void setConnectionFactory(ConnectionFactory connectionFactory) {
- this.connectionFactory = connectionFactory;
- }
-
-}
+/*
+ * Copyright 2005-2006 The Apache Software Foundation.
+ *
+ * Licensed 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.servicemix.wsn.jms;
+
+import javax.jms.Connection;
+import javax.jms.ConnectionFactory;
+
+import org.apache.servicemix.wsn.AbstractNotificationBroker;
+import org.apache.servicemix.wsn.AbstractPublisher;
+import org.apache.servicemix.wsn.AbstractSubscription;
+
+public abstract class JmsNotificationBroker extends AbstractNotificationBroker {
+
+ private ConnectionFactory connectionFactory;
+ private Connection connection;
+
+ public JmsNotificationBroker(String name) {
+ super(name);
+ }
+
+ public void init() throws Exception {
+ if (connection == null) {
+ connection = connectionFactory.createConnection();
+ connection.start();
+ }
+ super.init();
+ }
+
+ @Override
+ protected AbstractPublisher createPublisher(String name) {
+ JmsPublisher publisher = createJmsPublisher(name);
+ publisher.setManager(getManager());
+ publisher.setConnection(connection);
+ return publisher;
+ }
+
+ @Override
+ protected AbstractSubscription createSubcription(String name) {
+ JmsSubscription subscription = createJmsSubscription(name);
+ subscription.setManager(getManager());
+ subscription.setConnection(connection);
+ return subscription;
+ }
+
+ protected abstract JmsSubscription createJmsSubscription(String name);
+
+ protected abstract JmsPublisher createJmsPublisher(String name);
+
+ public ConnectionFactory getConnectionFactory() {
+ return connectionFactory;
+ }
+
+ public void setConnectionFactory(ConnectionFactory connectionFactory) {
+ this.connectionFactory = connectionFactory;
+ }
+
+}
Propchange: incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jms/JmsNotificationBroker.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jms/JmsPublisher.java
URL: http://svn.apache.org/viewcvs/incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jms/JmsPublisher.java?rev=379627&r1=379626&r2=379627&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jms/JmsPublisher.java (original)
+++ incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jms/JmsPublisher.java Tue Feb 21 15:40:05 2006
@@ -1,166 +1,166 @@
-/*
- * Copyright 2005-2006 The Apache Software Foundation.
- *
- * Licensed 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.servicemix.wsn.jms;
-
-import java.io.StringWriter;
-
-import javax.jms.Connection;
-import javax.jms.JMSException;
-import javax.jms.Message;
-import javax.jms.MessageProducer;
-import javax.jms.Session;
-import javax.jms.Topic;
-import javax.xml.bind.JAXBContext;
-import javax.xml.bind.JAXBException;
-
-import org.apache.activemq.advisory.ConsumerEvent;
-import org.apache.activemq.advisory.ConsumerEventSource;
-import org.apache.activemq.advisory.ConsumerListener;
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.servicemix.wsn.AbstractPublisher;
-import org.apache.servicemix.wsn.jaxws.InvalidTopicExpressionFault;
-import org.apache.servicemix.wsn.jaxws.PublisherRegistrationFailedFault;
-import org.apache.servicemix.wsn.jaxws.PublisherRegistrationRejectedFault;
-import org.apache.servicemix.wsn.jaxws.ResourceNotDestroyedFault;
-import org.apache.servicemix.wsn.jaxws.ResourceUnknownFault;
-import org.apache.servicemix.wsn.jaxws.TopicNotSupportedFault;
-import org.oasis_open.docs.wsn.b_2.InvalidTopicExpressionFaultType;
-import org.oasis_open.docs.wsn.b_2.NotificationMessageHolderType;
-import org.oasis_open.docs.wsn.b_2.Notify;
-import org.oasis_open.docs.wsn.br_2.PublisherRegistrationFailedFaultType;
-import org.oasis_open.docs.wsn.br_2.RegisterPublisher;
-import org.oasis_open.docs.wsn.br_2.ResourceNotDestroyedFaultType;
-
-public abstract class JmsPublisher extends AbstractPublisher implements ConsumerListener {
-
- private static Log log = LogFactory.getLog(JmsPublisher.class);
-
- private Connection connection;
- private JmsTopicExpressionConverter topicConverter;
- private JAXBContext jaxbContext;
- private Topic jmsTopic;
- private ConsumerEventSource advisory;
- private Object subscription;
-
- public JmsPublisher(String name) {
- super(name);
- topicConverter = new JmsTopicExpressionConverter();
- try {
- jaxbContext = JAXBContext.newInstance(Notify.class);
- } catch (JAXBException e) {
- throw new RuntimeException("Unable to create JAXB context", e);
- }
- }
-
- public Connection getConnection() {
- return connection;
- }
-
- public void setConnection(Connection connection) {
- this.connection = connection;
- }
-
- @Override
- public void notify(NotificationMessageHolderType messageHolder) {
- Session session = null;
- try {
- Topic topic = topicConverter.toActiveMQTopic(messageHolder.getTopic());
- session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
- MessageProducer producer = session.createProducer(topic);
- Notify notify = new Notify();
- notify.getNotificationMessage().add(messageHolder);
- StringWriter writer = new StringWriter();
- jaxbContext.createMarshaller().marshal(notify, writer);
- Message message = session.createTextMessage(writer.toString());
- producer.send(message);
- } catch (JMSException e) {
- log.warn("Error dispatching message", e);
- } catch (JAXBException e) {
- log.warn("Error dispatching message", e);
- } catch (InvalidTopicException e) {
- log.warn("Error dispatching message", e);
- } finally {
- if (session != null) {
- try {
- session.close();
- } catch (JMSException e) {
- log.debug("Error closing session", e);
- }
- }
- }
- }
-
- @Override
- protected void validatePublisher(RegisterPublisher registerPublisherRequest) throws InvalidTopicExpressionFault, PublisherRegistrationFailedFault, PublisherRegistrationRejectedFault, ResourceUnknownFault, TopicNotSupportedFault {
- super.validatePublisher(registerPublisherRequest);
- try {
- jmsTopic = topicConverter.toActiveMQTopic(topic);
- } catch (InvalidTopicException e) {
- InvalidTopicExpressionFaultType fault = new InvalidTopicExpressionFaultType();
- throw new InvalidTopicExpressionFault(e.getMessage(), fault);
- }
- }
-
- @Override
- protected void start() throws PublisherRegistrationFailedFault {
- if (demand) {
- try {
- advisory = new ConsumerEventSource(connection, jmsTopic);
- advisory.setConsumerListener(this);
- advisory.start();
- } catch (Exception e) {
- PublisherRegistrationFailedFaultType fault = new PublisherRegistrationFailedFaultType();
- throw new PublisherRegistrationFailedFault("Error starting demand-based publisher", fault, e);
- }
- }
- }
-
- protected void destroy() throws ResourceNotDestroyedFault {
- try {
- if (advisory != null) {
- advisory.stop();
- }
- } catch (Exception e) {
- ResourceNotDestroyedFaultType fault = new ResourceNotDestroyedFaultType();
- throw new ResourceNotDestroyedFault("Error destroying publisher", fault, e);
- } finally {
- super.destroy();
- }
- }
-
- public void onConsumerEvent(ConsumerEvent event) {
- if (event.getConsumerCount() > 0) {
- if (subscription == null) {
- // start subscription
- subscription = startSubscription();
- }
- } else {
- if (subscription != null) {
- // destroy subscription
- Object sub = subscription;
- subscription = null;
- destroySubscription(sub);
- }
- }
- }
-
- protected abstract void destroySubscription(Object subscription);
-
- protected abstract Object startSubscription();
-
-
-}
+/*
+ * Copyright 2005-2006 The Apache Software Foundation.
+ *
+ * Licensed 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.servicemix.wsn.jms;
+
+import java.io.StringWriter;
+
+import javax.jms.Connection;
+import javax.jms.JMSException;
+import javax.jms.Message;
+import javax.jms.MessageProducer;
+import javax.jms.Session;
+import javax.jms.Topic;
+import javax.xml.bind.JAXBContext;
+import javax.xml.bind.JAXBException;
+
+import org.apache.activemq.advisory.ConsumerEvent;
+import org.apache.activemq.advisory.ConsumerEventSource;
+import org.apache.activemq.advisory.ConsumerListener;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.servicemix.wsn.AbstractPublisher;
+import org.apache.servicemix.wsn.jaxws.InvalidTopicExpressionFault;
+import org.apache.servicemix.wsn.jaxws.PublisherRegistrationFailedFault;
+import org.apache.servicemix.wsn.jaxws.PublisherRegistrationRejectedFault;
+import org.apache.servicemix.wsn.jaxws.ResourceNotDestroyedFault;
+import org.apache.servicemix.wsn.jaxws.ResourceUnknownFault;
+import org.apache.servicemix.wsn.jaxws.TopicNotSupportedFault;
+import org.oasis_open.docs.wsn.b_2.InvalidTopicExpressionFaultType;
+import org.oasis_open.docs.wsn.b_2.NotificationMessageHolderType;
+import org.oasis_open.docs.wsn.b_2.Notify;
+import org.oasis_open.docs.wsn.br_2.PublisherRegistrationFailedFaultType;
+import org.oasis_open.docs.wsn.br_2.RegisterPublisher;
+import org.oasis_open.docs.wsn.br_2.ResourceNotDestroyedFaultType;
+
+public abstract class JmsPublisher extends AbstractPublisher implements ConsumerListener {
+
+ private static Log log = LogFactory.getLog(JmsPublisher.class);
+
+ private Connection connection;
+ private JmsTopicExpressionConverter topicConverter;
+ private JAXBContext jaxbContext;
+ private Topic jmsTopic;
+ private ConsumerEventSource advisory;
+ private Object subscription;
+
+ public JmsPublisher(String name) {
+ super(name);
+ topicConverter = new JmsTopicExpressionConverter();
+ try {
+ jaxbContext = JAXBContext.newInstance(Notify.class);
+ } catch (JAXBException e) {
+ throw new RuntimeException("Unable to create JAXB context", e);
+ }
+ }
+
+ public Connection getConnection() {
+ return connection;
+ }
+
+ public void setConnection(Connection connection) {
+ this.connection = connection;
+ }
+
+ @Override
+ public void notify(NotificationMessageHolderType messageHolder) {
+ Session session = null;
+ try {
+ Topic topic = topicConverter.toActiveMQTopic(messageHolder.getTopic());
+ session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+ MessageProducer producer = session.createProducer(topic);
+ Notify notify = new Notify();
+ notify.getNotificationMessage().add(messageHolder);
+ StringWriter writer = new StringWriter();
+ jaxbContext.createMarshaller().marshal(notify, writer);
+ Message message = session.createTextMessage(writer.toString());
+ producer.send(message);
+ } catch (JMSException e) {
+ log.warn("Error dispatching message", e);
+ } catch (JAXBException e) {
+ log.warn("Error dispatching message", e);
+ } catch (InvalidTopicException e) {
+ log.warn("Error dispatching message", e);
+ } finally {
+ if (session != null) {
+ try {
+ session.close();
+ } catch (JMSException e) {
+ log.debug("Error closing session", e);
+ }
+ }
+ }
+ }
+
+ @Override
+ protected void validatePublisher(RegisterPublisher registerPublisherRequest) throws InvalidTopicExpressionFault, PublisherRegistrationFailedFault, PublisherRegistrationRejectedFault, ResourceUnknownFault, TopicNotSupportedFault {
+ super.validatePublisher(registerPublisherRequest);
+ try {
+ jmsTopic = topicConverter.toActiveMQTopic(topic);
+ } catch (InvalidTopicException e) {
+ InvalidTopicExpressionFaultType fault = new InvalidTopicExpressionFaultType();
+ throw new InvalidTopicExpressionFault(e.getMessage(), fault);
+ }
+ }
+
+ @Override
+ protected void start() throws PublisherRegistrationFailedFault {
+ if (demand) {
+ try {
+ advisory = new ConsumerEventSource(connection, jmsTopic);
+ advisory.setConsumerListener(this);
+ advisory.start();
+ } catch (Exception e) {
+ PublisherRegistrationFailedFaultType fault = new PublisherRegistrationFailedFaultType();
+ throw new PublisherRegistrationFailedFault("Error starting demand-based publisher", fault, e);
+ }
+ }
+ }
+
+ protected void destroy() throws ResourceNotDestroyedFault {
+ try {
+ if (advisory != null) {
+ advisory.stop();
+ }
+ } catch (Exception e) {
+ ResourceNotDestroyedFaultType fault = new ResourceNotDestroyedFaultType();
+ throw new ResourceNotDestroyedFault("Error destroying publisher", fault, e);
+ } finally {
+ super.destroy();
+ }
+ }
+
+ public void onConsumerEvent(ConsumerEvent event) {
+ if (event.getConsumerCount() > 0) {
+ if (subscription == null) {
+ // start subscription
+ subscription = startSubscription();
+ }
+ } else {
+ if (subscription != null) {
+ // destroy subscription
+ Object sub = subscription;
+ subscription = null;
+ destroySubscription(sub);
+ }
+ }
+ }
+
+ protected abstract void destroySubscription(Object subscription);
+
+ protected abstract Object startSubscription();
+
+
+}
Propchange: incubator/servicemix/trunk/servicemix-wsn2005/src/main/java/org/apache/servicemix/wsn/jms/JmsPublisher.java
------------------------------------------------------------------------------
svn:eol-style = native