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/07/28 07:49:50 UTC
svn commit: r426392 [21/47] - in /incubator/servicemix/trunk:
apache-servicemix/src/main/assembly/ apache-servicemix/src/main/release/conf/
apache-servicemix/src/main/release/examples/basic/
apache-servicemix/src/main/release/examples/bpel-bpe/ apache-...
Modified: incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPComponent.java
URL: http://svn.apache.org/viewvc/incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPComponent.java?rev=426392&r1=426391&r2=426392&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPComponent.java (original)
+++ incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPComponent.java Thu Jul 27 22:49:32 2006
@@ -1,44 +1,44 @@
-/*
- * 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.eip;
-
-import org.apache.servicemix.common.BaseComponent;
-import org.apache.servicemix.common.BaseLifeCycle;
-import org.apache.servicemix.common.BaseServiceUnitManager;
-import org.apache.servicemix.common.Deployer;
-
-/**
- * @author gnodet
- * @version $Revision: 376451 $
- */
-public class EIPComponent extends BaseComponent {
-
- /* (non-Javadoc)
- * @see org.servicemix.common.BaseComponent#createLifeCycle()
- */
- protected BaseLifeCycle createLifeCycle() {
- return new EIPLifeCycle(this);
- }
-
- /* (non-Javadoc)
- * @see org.servicemix.common.BaseComponent#createServiceUnitManager()
- */
- public BaseServiceUnitManager createServiceUnitManager() {
- Deployer[] deployers = new Deployer[] { new EIPDeployer(this) };
- return new BaseServiceUnitManager(this, deployers);
- }
-
-}
+/*
+ * 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.eip;
+
+import org.apache.servicemix.common.BaseComponent;
+import org.apache.servicemix.common.BaseLifeCycle;
+import org.apache.servicemix.common.BaseServiceUnitManager;
+import org.apache.servicemix.common.Deployer;
+
+/**
+ * @author gnodet
+ * @version $Revision: 376451 $
+ */
+public class EIPComponent extends BaseComponent {
+
+ /* (non-Javadoc)
+ * @see org.servicemix.common.BaseComponent#createLifeCycle()
+ */
+ protected BaseLifeCycle createLifeCycle() {
+ return new EIPLifeCycle(this);
+ }
+
+ /* (non-Javadoc)
+ * @see org.servicemix.common.BaseComponent#createServiceUnitManager()
+ */
+ public BaseServiceUnitManager createServiceUnitManager() {
+ Deployer[] deployers = new Deployer[] { new EIPDeployer(this) };
+ return new BaseServiceUnitManager(this, deployers);
+ }
+
+}
Propchange: incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPComponent.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPConfiguration.java
URL: http://svn.apache.org/viewvc/incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPConfiguration.java?rev=426392&r1=426391&r2=426392&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPConfiguration.java (original)
+++ incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPConfiguration.java Thu Jul 27 22:49:32 2006
@@ -1,28 +1,28 @@
-/*
- * 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.eip;
-
-import org.apache.servicemix.common.PersistentConfiguration;
-
-/**
- * EIP component configuration
- *
- * @author gnodet
- * @version $Revision: 376451 $
- */
-public class EIPConfiguration extends PersistentConfiguration implements EIPConfigurationMBean {
-
-}
+/*
+ * 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.eip;
+
+import org.apache.servicemix.common.PersistentConfiguration;
+
+/**
+ * EIP component configuration
+ *
+ * @author gnodet
+ * @version $Revision: 376451 $
+ */
+public class EIPConfiguration extends PersistentConfiguration implements EIPConfigurationMBean {
+
+}
Propchange: incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPConfiguration.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPConfigurationMBean.java
URL: http://svn.apache.org/viewvc/incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPConfigurationMBean.java?rev=426392&r1=426391&r2=426392&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPConfigurationMBean.java (original)
+++ incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPConfigurationMBean.java Thu Jul 27 22:49:32 2006
@@ -1,26 +1,26 @@
-/*
- * 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.eip;
-
-/**
- * MBean interface for the EIP component configuration
- *
- * @author gnodet
- * @version $Revision: 376451 $
- */
-public interface EIPConfigurationMBean {
-
-}
+/*
+ * 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.eip;
+
+/**
+ * MBean interface for the EIP component configuration
+ *
+ * @author gnodet
+ * @version $Revision: 376451 $
+ */
+public interface EIPConfigurationMBean {
+
+}
Propchange: incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPConfigurationMBean.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPDeployer.java
URL: http://svn.apache.org/viewvc/incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPDeployer.java?rev=426392&r1=426391&r2=426392&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPDeployer.java (original)
+++ incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPDeployer.java Thu Jul 27 22:49:32 2006
@@ -1,41 +1,41 @@
-/*
- * 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.eip;
-
-import javax.jbi.management.DeploymentException;
-
-import org.apache.servicemix.common.BaseComponent;
-import org.apache.servicemix.common.Endpoint;
-import org.apache.servicemix.common.xbean.AbstractXBeanDeployer;
-
-/**
- * @author gnodet
- * @version $Revision: 376451 $
- */
-public class EIPDeployer extends AbstractXBeanDeployer {
-
- public EIPDeployer(BaseComponent component) {
- super(component);
- }
-
- protected boolean validate(Endpoint endpoint) throws DeploymentException {
- if (endpoint instanceof EIPEndpoint == false) {
- throw new DeploymentException("Endpoint should be an EIPEndpoint");
- }
- ((EIPEndpoint) endpoint).validate();
- return true;
- }
-}
+/*
+ * 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.eip;
+
+import javax.jbi.management.DeploymentException;
+
+import org.apache.servicemix.common.BaseComponent;
+import org.apache.servicemix.common.Endpoint;
+import org.apache.servicemix.common.xbean.AbstractXBeanDeployer;
+
+/**
+ * @author gnodet
+ * @version $Revision: 376451 $
+ */
+public class EIPDeployer extends AbstractXBeanDeployer {
+
+ public EIPDeployer(BaseComponent component) {
+ super(component);
+ }
+
+ protected boolean validate(Endpoint endpoint) throws DeploymentException {
+ if (endpoint instanceof EIPEndpoint == false) {
+ throw new DeploymentException("Endpoint should be an EIPEndpoint");
+ }
+ ((EIPEndpoint) endpoint).validate();
+ return true;
+ }
+}
Propchange: incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPDeployer.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPEndpoint.java
URL: http://svn.apache.org/viewvc/incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPEndpoint.java?rev=426392&r1=426391&r2=426392&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPEndpoint.java (original)
+++ incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPEndpoint.java Thu Jul 27 22:49:32 2006
@@ -1,379 +1,379 @@
-/*
- * 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.eip;
-
-import java.net.URL;
-
-import javax.jbi.JBIException;
-import javax.jbi.component.ComponentContext;
-import javax.jbi.management.DeploymentException;
-import javax.jbi.messaging.DeliveryChannel;
-import javax.jbi.messaging.ExchangeStatus;
-import javax.jbi.messaging.MessageExchange;
-import javax.jbi.messaging.MessageExchangeFactory;
-import javax.jbi.messaging.MessagingException;
-import javax.jbi.messaging.MessageExchange.Role;
-import javax.jbi.servicedesc.ServiceEndpoint;
-import javax.wsdl.Definition;
-import javax.wsdl.WSDLException;
-import javax.wsdl.factory.WSDLFactory;
-import javax.wsdl.xml.WSDLReader;
-
-import org.apache.servicemix.JbiConstants;
-import org.apache.servicemix.common.BaseLifeCycle;
-import org.apache.servicemix.common.Endpoint;
-import org.apache.servicemix.common.ExchangeProcessor;
-import org.apache.servicemix.eip.support.ExchangeTarget;
-import org.apache.servicemix.locks.LockManager;
-import org.apache.servicemix.locks.impl.SimpleLockManager;
-import org.apache.servicemix.store.Store;
-import org.apache.servicemix.store.StoreFactory;
-import org.apache.servicemix.store.memory.MemoryStoreFactory;
-import org.apache.servicemix.timers.TimerManager;
-import org.apache.servicemix.timers.impl.TimerManagerImpl;
-import org.springframework.core.io.Resource;
-import org.w3c.dom.Document;
-
-/**
- * @author gnodet
- * @version $Revision: 376451 $
- */
-public abstract class EIPEndpoint extends Endpoint implements ExchangeProcessor {
-
- private ServiceEndpoint activated;
- private DeliveryChannel channel;
- private Resource wsdlResource;
-
- /**
- * The store to keep pending exchanges
- */
- protected Store store;
- /**
- * The store factory.
- */
- protected StoreFactory storeFactory;
- /**
- * The lock manager.
- */
- protected LockManager lockManager;
- /**
- * The timer manager.
- */
- protected TimerManager timerManager;
- /**
- * The exchange factory
- */
- protected MessageExchangeFactory exchangeFactory;
-
- /**
- * The ExchangeTarget to use to get the WSDL
- */
- protected ExchangeTarget wsdlExchangeTarget;
-
- /**
- * @return Returns the exchangeFactory.
- */
- public MessageExchangeFactory getExchangeFactory() {
- return exchangeFactory;
- }
- /**
- * @param exchangeFactory The exchangeFactory to set.
- */
- public void setExchangeFactory(MessageExchangeFactory exchangeFactory) {
- this.exchangeFactory = exchangeFactory;
- }
- /**
- * @return Returns the store.
- */
- public Store getStore() {
- return store;
- }
- /**
- * @param store The store to set.
- */
- public void setStore(Store store) {
- this.store = store;
- }
- /**
- * @return Returns the storeFactory.
- */
- public StoreFactory getStoreFactory() {
- return storeFactory;
- }
- /**
- * @param storeFactory The storeFactory to set.
- */
- public void setStoreFactory(StoreFactory storeFactory) {
- this.storeFactory = storeFactory;
- }
- /**
- * @return the lockManager
- */
- public LockManager getLockManager() {
- return lockManager;
- }
- /**
- * @param lockManager the lockManager to set
- */
- public void setLockManager(LockManager lockManager) {
- this.lockManager = lockManager;
- }
- /**
- * @return the timerManager
- */
- public TimerManager getTimerManager() {
- return timerManager;
- }
- /**
- * @param timerManager the timerManager to set
- */
- public void setTimerManager(TimerManager timerManager) {
- this.timerManager = timerManager;
- }
-
- /* (non-Javadoc)
- * @see org.apache.servicemix.common.Endpoint#getRole()
- */
- public Role getRole() {
- return Role.PROVIDER;
- }
-
- public void activate() throws Exception {
- logger = this.serviceUnit.getComponent().getLogger();
- ComponentContext ctx = getContext();
- channel = ctx.getDeliveryChannel();
- exchangeFactory = channel.createExchangeFactory();
- activated = ctx.activateEndpoint(service, endpoint);
- if (store == null) {
- if (storeFactory == null) {
- storeFactory = new MemoryStoreFactory();
- }
- store = storeFactory.open(getService().toString() + getEndpoint());
- }
- if (lockManager == null) {
- lockManager = new SimpleLockManager();
- }
- if (timerManager == null) {
- timerManager = new TimerManagerImpl();
- }
- timerManager.start();
- start();
- }
-
- public void deactivate() throws Exception {
- if (timerManager != null) {
- timerManager.stop();
- }
- stop();
- ServiceEndpoint ep = activated;
- activated = null;
- ComponentContext ctx = getServiceUnit().getComponent().getComponentContext();
- ctx.deactivateEndpoint(ep);
- }
-
- public ExchangeProcessor getProcessor() {
- return this;
- }
-
- public void validate() throws DeploymentException {
- }
-
- protected ComponentContext getContext() {
- return getServiceUnit().getComponent().getComponentContext();
- }
-
- protected void send(MessageExchange me) throws MessagingException {
- if (me.getRole() == MessageExchange.Role.CONSUMER &&
- me.getStatus() == ExchangeStatus.ACTIVE) {
- BaseLifeCycle lf = (BaseLifeCycle) getServiceUnit().getComponent().getLifeCycle();
- lf.sendConsumerExchange(me, (Endpoint) this);
- } else {
- channel.send(me);
- }
- }
-
- protected void sendSync(MessageExchange me) throws MessagingException {
- if (!channel.sendSync(me)) {
- throw new MessagingException("SendSync failed");
- }
- }
-
- protected void done(MessageExchange me) throws MessagingException {
- me.setStatus(ExchangeStatus.DONE);
- send(me);
- }
-
- protected void fail(MessageExchange me, Exception error) throws MessagingException {
- me.setError(error);
- send(me);
- }
-
- public void start() throws Exception {
- }
-
- public void stop() {
- }
-
- /* (non-Javadoc)
- * @see org.apache.servicemix.common.ExchangeProcessor#process(javax.jbi.messaging.MessageExchange)
- */
- public void process(MessageExchange exchange) throws Exception {
- boolean txSync = exchange.isTransacted() && Boolean.TRUE.equals(exchange.getProperty(JbiConstants.SEND_SYNC));
- if (txSync && exchange.getRole() == Role.PROVIDER && exchange.getStatus() == ExchangeStatus.ACTIVE) {
- processSync(exchange);
- } else {
- processAsync(exchange);
- }
- }
-
- /**
- * @return Returns the description.
- */
- public Document getDescription() {
- if( description == null ) {
- definition = getDefinition();
- if( definition!=null ) {
- try {
- description = WSDLFactory.newInstance().newWSDLWriter().getDocument(definition);
- } catch (WSDLException e) {
- }
- }
- }
- return description;
- }
-
- /**
- * If the definition is not currently set, it tries to set it using
- * the following sources in the order:
- * description, wsdlResource, wsdlExchangeTarget
- *
- * @return Returns the definition.
- */
- public Definition getDefinition() {
- if( definition == null ) {
- definition = getDefinitionFromDescription();
- if( definition == null ) {
- definition = getDefinitionFromWsdlResource();
- if( definition == null ) {
- definition = getDefinitionFromWsdlExchangeTarget();
- }
- }
- }
- return definition;
- }
-
- protected Definition getDefinitionFromDescription() {
- if( description!=null ) {
- try {
- return WSDLFactory.newInstance().newWSDLReader().readWSDL(null, description);
- } catch (WSDLException ignore) {
- }
- }
- return null;
- }
-
- protected Definition getDefinitionFromWsdlResource() {
- if( wsdlResource!=null ) {
- try {
- URL resource = wsdlResource.getURL();
- WSDLReader reader = WSDLFactory.newInstance().newWSDLReader();
- return reader.readWSDL(null, resource.toString());
- } catch (Throwable ignore) {
- }
- }
- return null;
- }
-
- protected Definition getDefinitionFromWsdlExchangeTarget() {
- if( wsdlExchangeTarget != null ) {
- try {
- Document description = getDescriptionForExchangeTarget(wsdlExchangeTarget);
- return WSDLFactory.newInstance().newWSDLReader().readWSDL(null, description);
- } catch (Throwable ignore) {
- }
- }
- return null;
- }
-
- /**
- * @return Returns the wsdl's Resource.
- */
- public Resource getWsdlResource() {
- return wsdlResource;
- }
- public void setWsdlResource(Resource wsdlResource) {
- this.wsdlResource = wsdlResource;
- }
-
- protected Document getDescriptionForExchangeTarget(ExchangeTarget match) throws JBIException {
- ServiceEndpoint[] endpoints = getEndpointsForExchangeTarget(match);
- if (endpoints == null || endpoints.length == 0) {
- return null;
- }
- ServiceEndpoint endpoint = chooseFirstEndpointWithDescriptor(endpoints);
- if (endpoint == null) {
- return null;
- }
- return getContext().getEndpointDescriptor(endpoint);
- }
-
- /**
- *
- * @param match
- * @return an ServiceEndpoint[] of all the endpoints that matched.
- * @throws JBIException
- */
- protected ServiceEndpoint[] getEndpointsForExchangeTarget(ExchangeTarget match) throws JBIException {
- ServiceEndpoint[] endpoints;
- if (match.getEndpoint() != null && match.getService() != null) {
- ServiceEndpoint endpoint = getContext().getEndpoint(match.getService(), match.getEndpoint());
- if (endpoint == null) {
- endpoints = new ServiceEndpoint[0];
- } else {
- endpoints = new ServiceEndpoint[] { endpoint };
- }
- } else if (match.getService() != null) {
- endpoints = getContext().getEndpointsForService(match.getService());
- } else if (interfaceName != null) {
- endpoints = getContext().getEndpoints(interfaceName);
- } else {
- throw new IllegalStateException("One of interfaceName or serviceName should be provided");
- }
- return endpoints;
- }
-
- protected ServiceEndpoint chooseFirstEndpointWithDescriptor(ServiceEndpoint[] endpoints) throws JBIException {
- for (int i = 0; i < endpoints.length; i++) {
- if (getContext().getEndpointDescriptor(endpoints[i]) != null) {
- return endpoints[i];
- }
- }
- return null;
- }
-
-
- protected abstract void processAsync(MessageExchange exchange) throws Exception;
-
- protected abstract void processSync(MessageExchange exchange) throws Exception;
-
- public ExchangeTarget getWsdlExchangeTarget() {
- return wsdlExchangeTarget;
- }
- public void setWsdlExchangeTarget(ExchangeTarget wsdlExchangeTarget) {
- this.wsdlExchangeTarget = wsdlExchangeTarget;
- }
-
-}
+/*
+ * 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.eip;
+
+import java.net.URL;
+
+import javax.jbi.JBIException;
+import javax.jbi.component.ComponentContext;
+import javax.jbi.management.DeploymentException;
+import javax.jbi.messaging.DeliveryChannel;
+import javax.jbi.messaging.ExchangeStatus;
+import javax.jbi.messaging.MessageExchange;
+import javax.jbi.messaging.MessageExchangeFactory;
+import javax.jbi.messaging.MessagingException;
+import javax.jbi.messaging.MessageExchange.Role;
+import javax.jbi.servicedesc.ServiceEndpoint;
+import javax.wsdl.Definition;
+import javax.wsdl.WSDLException;
+import javax.wsdl.factory.WSDLFactory;
+import javax.wsdl.xml.WSDLReader;
+
+import org.apache.servicemix.JbiConstants;
+import org.apache.servicemix.common.BaseLifeCycle;
+import org.apache.servicemix.common.Endpoint;
+import org.apache.servicemix.common.ExchangeProcessor;
+import org.apache.servicemix.eip.support.ExchangeTarget;
+import org.apache.servicemix.locks.LockManager;
+import org.apache.servicemix.locks.impl.SimpleLockManager;
+import org.apache.servicemix.store.Store;
+import org.apache.servicemix.store.StoreFactory;
+import org.apache.servicemix.store.memory.MemoryStoreFactory;
+import org.apache.servicemix.timers.TimerManager;
+import org.apache.servicemix.timers.impl.TimerManagerImpl;
+import org.springframework.core.io.Resource;
+import org.w3c.dom.Document;
+
+/**
+ * @author gnodet
+ * @version $Revision: 376451 $
+ */
+public abstract class EIPEndpoint extends Endpoint implements ExchangeProcessor {
+
+ private ServiceEndpoint activated;
+ private DeliveryChannel channel;
+ private Resource wsdlResource;
+
+ /**
+ * The store to keep pending exchanges
+ */
+ protected Store store;
+ /**
+ * The store factory.
+ */
+ protected StoreFactory storeFactory;
+ /**
+ * The lock manager.
+ */
+ protected LockManager lockManager;
+ /**
+ * The timer manager.
+ */
+ protected TimerManager timerManager;
+ /**
+ * The exchange factory
+ */
+ protected MessageExchangeFactory exchangeFactory;
+
+ /**
+ * The ExchangeTarget to use to get the WSDL
+ */
+ protected ExchangeTarget wsdlExchangeTarget;
+
+ /**
+ * @return Returns the exchangeFactory.
+ */
+ public MessageExchangeFactory getExchangeFactory() {
+ return exchangeFactory;
+ }
+ /**
+ * @param exchangeFactory The exchangeFactory to set.
+ */
+ public void setExchangeFactory(MessageExchangeFactory exchangeFactory) {
+ this.exchangeFactory = exchangeFactory;
+ }
+ /**
+ * @return Returns the store.
+ */
+ public Store getStore() {
+ return store;
+ }
+ /**
+ * @param store The store to set.
+ */
+ public void setStore(Store store) {
+ this.store = store;
+ }
+ /**
+ * @return Returns the storeFactory.
+ */
+ public StoreFactory getStoreFactory() {
+ return storeFactory;
+ }
+ /**
+ * @param storeFactory The storeFactory to set.
+ */
+ public void setStoreFactory(StoreFactory storeFactory) {
+ this.storeFactory = storeFactory;
+ }
+ /**
+ * @return the lockManager
+ */
+ public LockManager getLockManager() {
+ return lockManager;
+ }
+ /**
+ * @param lockManager the lockManager to set
+ */
+ public void setLockManager(LockManager lockManager) {
+ this.lockManager = lockManager;
+ }
+ /**
+ * @return the timerManager
+ */
+ public TimerManager getTimerManager() {
+ return timerManager;
+ }
+ /**
+ * @param timerManager the timerManager to set
+ */
+ public void setTimerManager(TimerManager timerManager) {
+ this.timerManager = timerManager;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.servicemix.common.Endpoint#getRole()
+ */
+ public Role getRole() {
+ return Role.PROVIDER;
+ }
+
+ public void activate() throws Exception {
+ logger = this.serviceUnit.getComponent().getLogger();
+ ComponentContext ctx = getContext();
+ channel = ctx.getDeliveryChannel();
+ exchangeFactory = channel.createExchangeFactory();
+ activated = ctx.activateEndpoint(service, endpoint);
+ if (store == null) {
+ if (storeFactory == null) {
+ storeFactory = new MemoryStoreFactory();
+ }
+ store = storeFactory.open(getService().toString() + getEndpoint());
+ }
+ if (lockManager == null) {
+ lockManager = new SimpleLockManager();
+ }
+ if (timerManager == null) {
+ timerManager = new TimerManagerImpl();
+ }
+ timerManager.start();
+ start();
+ }
+
+ public void deactivate() throws Exception {
+ if (timerManager != null) {
+ timerManager.stop();
+ }
+ stop();
+ ServiceEndpoint ep = activated;
+ activated = null;
+ ComponentContext ctx = getServiceUnit().getComponent().getComponentContext();
+ ctx.deactivateEndpoint(ep);
+ }
+
+ public ExchangeProcessor getProcessor() {
+ return this;
+ }
+
+ public void validate() throws DeploymentException {
+ }
+
+ protected ComponentContext getContext() {
+ return getServiceUnit().getComponent().getComponentContext();
+ }
+
+ protected void send(MessageExchange me) throws MessagingException {
+ if (me.getRole() == MessageExchange.Role.CONSUMER &&
+ me.getStatus() == ExchangeStatus.ACTIVE) {
+ BaseLifeCycle lf = (BaseLifeCycle) getServiceUnit().getComponent().getLifeCycle();
+ lf.sendConsumerExchange(me, (Endpoint) this);
+ } else {
+ channel.send(me);
+ }
+ }
+
+ protected void sendSync(MessageExchange me) throws MessagingException {
+ if (!channel.sendSync(me)) {
+ throw new MessagingException("SendSync failed");
+ }
+ }
+
+ protected void done(MessageExchange me) throws MessagingException {
+ me.setStatus(ExchangeStatus.DONE);
+ send(me);
+ }
+
+ protected void fail(MessageExchange me, Exception error) throws MessagingException {
+ me.setError(error);
+ send(me);
+ }
+
+ public void start() throws Exception {
+ }
+
+ public void stop() {
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.servicemix.common.ExchangeProcessor#process(javax.jbi.messaging.MessageExchange)
+ */
+ public void process(MessageExchange exchange) throws Exception {
+ boolean txSync = exchange.isTransacted() && Boolean.TRUE.equals(exchange.getProperty(JbiConstants.SEND_SYNC));
+ if (txSync && exchange.getRole() == Role.PROVIDER && exchange.getStatus() == ExchangeStatus.ACTIVE) {
+ processSync(exchange);
+ } else {
+ processAsync(exchange);
+ }
+ }
+
+ /**
+ * @return Returns the description.
+ */
+ public Document getDescription() {
+ if( description == null ) {
+ definition = getDefinition();
+ if( definition!=null ) {
+ try {
+ description = WSDLFactory.newInstance().newWSDLWriter().getDocument(definition);
+ } catch (WSDLException e) {
+ }
+ }
+ }
+ return description;
+ }
+
+ /**
+ * If the definition is not currently set, it tries to set it using
+ * the following sources in the order:
+ * description, wsdlResource, wsdlExchangeTarget
+ *
+ * @return Returns the definition.
+ */
+ public Definition getDefinition() {
+ if( definition == null ) {
+ definition = getDefinitionFromDescription();
+ if( definition == null ) {
+ definition = getDefinitionFromWsdlResource();
+ if( definition == null ) {
+ definition = getDefinitionFromWsdlExchangeTarget();
+ }
+ }
+ }
+ return definition;
+ }
+
+ protected Definition getDefinitionFromDescription() {
+ if( description!=null ) {
+ try {
+ return WSDLFactory.newInstance().newWSDLReader().readWSDL(null, description);
+ } catch (WSDLException ignore) {
+ }
+ }
+ return null;
+ }
+
+ protected Definition getDefinitionFromWsdlResource() {
+ if( wsdlResource!=null ) {
+ try {
+ URL resource = wsdlResource.getURL();
+ WSDLReader reader = WSDLFactory.newInstance().newWSDLReader();
+ return reader.readWSDL(null, resource.toString());
+ } catch (Throwable ignore) {
+ }
+ }
+ return null;
+ }
+
+ protected Definition getDefinitionFromWsdlExchangeTarget() {
+ if( wsdlExchangeTarget != null ) {
+ try {
+ Document description = getDescriptionForExchangeTarget(wsdlExchangeTarget);
+ return WSDLFactory.newInstance().newWSDLReader().readWSDL(null, description);
+ } catch (Throwable ignore) {
+ }
+ }
+ return null;
+ }
+
+ /**
+ * @return Returns the wsdl's Resource.
+ */
+ public Resource getWsdlResource() {
+ return wsdlResource;
+ }
+ public void setWsdlResource(Resource wsdlResource) {
+ this.wsdlResource = wsdlResource;
+ }
+
+ protected Document getDescriptionForExchangeTarget(ExchangeTarget match) throws JBIException {
+ ServiceEndpoint[] endpoints = getEndpointsForExchangeTarget(match);
+ if (endpoints == null || endpoints.length == 0) {
+ return null;
+ }
+ ServiceEndpoint endpoint = chooseFirstEndpointWithDescriptor(endpoints);
+ if (endpoint == null) {
+ return null;
+ }
+ return getContext().getEndpointDescriptor(endpoint);
+ }
+
+ /**
+ *
+ * @param match
+ * @return an ServiceEndpoint[] of all the endpoints that matched.
+ * @throws JBIException
+ */
+ protected ServiceEndpoint[] getEndpointsForExchangeTarget(ExchangeTarget match) throws JBIException {
+ ServiceEndpoint[] endpoints;
+ if (match.getEndpoint() != null && match.getService() != null) {
+ ServiceEndpoint endpoint = getContext().getEndpoint(match.getService(), match.getEndpoint());
+ if (endpoint == null) {
+ endpoints = new ServiceEndpoint[0];
+ } else {
+ endpoints = new ServiceEndpoint[] { endpoint };
+ }
+ } else if (match.getService() != null) {
+ endpoints = getContext().getEndpointsForService(match.getService());
+ } else if (interfaceName != null) {
+ endpoints = getContext().getEndpoints(interfaceName);
+ } else {
+ throw new IllegalStateException("One of interfaceName or serviceName should be provided");
+ }
+ return endpoints;
+ }
+
+ protected ServiceEndpoint chooseFirstEndpointWithDescriptor(ServiceEndpoint[] endpoints) throws JBIException {
+ for (int i = 0; i < endpoints.length; i++) {
+ if (getContext().getEndpointDescriptor(endpoints[i]) != null) {
+ return endpoints[i];
+ }
+ }
+ return null;
+ }
+
+
+ protected abstract void processAsync(MessageExchange exchange) throws Exception;
+
+ protected abstract void processSync(MessageExchange exchange) throws Exception;
+
+ public ExchangeTarget getWsdlExchangeTarget() {
+ return wsdlExchangeTarget;
+ }
+ public void setWsdlExchangeTarget(ExchangeTarget wsdlExchangeTarget) {
+ this.wsdlExchangeTarget = wsdlExchangeTarget;
+ }
+
+}
Propchange: incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPEndpoint.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPLifeCycle.java
URL: http://svn.apache.org/viewvc/incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPLifeCycle.java?rev=426392&r1=426391&r2=426392&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPLifeCycle.java (original)
+++ incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPLifeCycle.java Thu Jul 27 22:49:32 2006
@@ -1,61 +1,61 @@
-/*
- * 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.eip;
-
-import org.apache.servicemix.common.BaseComponent;
-import org.apache.servicemix.common.BaseLifeCycle;
-
-/**
- * @author gnodet
- * @version $Revision: 376451 $
- */
-public class EIPLifeCycle extends BaseLifeCycle {
-
- protected EIPConfiguration configuration;
-
- public EIPLifeCycle(BaseComponent component) {
- super(component);
- configuration = new EIPConfiguration();
- }
-
- /**
- * @return Returns the configuration.
- */
- public EIPConfiguration getConfiguration() {
- return configuration;
- }
-
- /**
- * @param configuration The configuration to set.
- */
- public void setConfiguration(EIPConfiguration configuration) {
- this.configuration = configuration;
- }
-
- /* (non-Javadoc)
- * @see org.servicemix.common.BaseComponentLifeCycle#getExtensionMBean()
- */
- protected Object getExtensionMBean() throws Exception {
- return configuration;
- }
-
- protected void doInit() throws Exception {
- super.doInit();
- configuration.setRootDir(context.getWorkspaceRoot());
- configuration.load();
- }
-
-}
+/*
+ * 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.eip;
+
+import org.apache.servicemix.common.BaseComponent;
+import org.apache.servicemix.common.BaseLifeCycle;
+
+/**
+ * @author gnodet
+ * @version $Revision: 376451 $
+ */
+public class EIPLifeCycle extends BaseLifeCycle {
+
+ protected EIPConfiguration configuration;
+
+ public EIPLifeCycle(BaseComponent component) {
+ super(component);
+ configuration = new EIPConfiguration();
+ }
+
+ /**
+ * @return Returns the configuration.
+ */
+ public EIPConfiguration getConfiguration() {
+ return configuration;
+ }
+
+ /**
+ * @param configuration The configuration to set.
+ */
+ public void setConfiguration(EIPConfiguration configuration) {
+ this.configuration = configuration;
+ }
+
+ /* (non-Javadoc)
+ * @see org.servicemix.common.BaseComponentLifeCycle#getExtensionMBean()
+ */
+ protected Object getExtensionMBean() throws Exception {
+ return configuration;
+ }
+
+ protected void doInit() throws Exception {
+ super.doInit();
+ configuration.setRootDir(context.getWorkspaceRoot());
+ configuration.load();
+ }
+
+}
Propchange: incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPLifeCycle.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPSpringComponent.java
URL: http://svn.apache.org/viewvc/incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPSpringComponent.java?rev=426392&r1=426391&r2=426392&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPSpringComponent.java (original)
+++ incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPSpringComponent.java Thu Jul 27 22:49:32 2006
@@ -1,106 +1,106 @@
-/*
- * 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.eip;
-
-import org.apache.servicemix.common.BaseComponent;
-import org.apache.servicemix.common.BaseLifeCycle;
-import org.apache.servicemix.common.ServiceUnit;
-import org.apache.servicemix.common.xbean.XBeanServiceUnit;
-
-/**
- *
- * @author gnodet
- * @version $Revision: 376451 $
- * @org.apache.xbean.XBean element="component"
- * description="An EIP component"
- */
-public class EIPSpringComponent extends BaseComponent {
-
- private EIPEndpoint[] endpoints;
-
- /**
- * @return Returns the endpoints.
- */
- public EIPEndpoint[] getEndpoints() {
- return endpoints;
- }
-
- /**
- * @param endpoints The endpoints to set.
- */
- public void setEndpoints(EIPEndpoint[] endpoints) {
- this.endpoints = endpoints;
- }
-
- /* (non-Javadoc)
- * @see org.servicemix.common.BaseComponent#createLifeCycle()
- */
- protected BaseLifeCycle createLifeCycle() {
- return new LifeCycle();
- }
-
- /**
- * @author gnodet
- */
- public class LifeCycle extends EIPLifeCycle {
-
- protected ServiceUnit su;
-
- public LifeCycle() {
- super(EIPSpringComponent.this);
- }
-
- /* (non-Javadoc)
- * @see org.servicemix.common.BaseLifeCycle#doInit()
- */
- protected void doInit() throws Exception {
- super.doInit();
- su = new XBeanServiceUnit();
- su.setComponent(EIPSpringComponent.this);
- for (int i = 0; i < endpoints.length; i++) {
- endpoints[i].setServiceUnit(su);
- endpoints[i].validate();
- su.addEndpoint(endpoints[i]);
- }
- getRegistry().registerServiceUnit(su);
- }
-
- /* (non-Javadoc)
- * @see org.servicemix.common.BaseLifeCycle#doStart()
- */
- protected void doStart() throws Exception {
- super.doStart();
- su.start();
- }
-
- /* (non-Javadoc)
- * @see org.servicemix.common.BaseLifeCycle#doStop()
- */
- protected void doStop() throws Exception {
- su.stop();
- super.doStop();
- }
-
- /* (non-Javadoc)
- * @see org.servicemix.common.BaseLifeCycle#doShutDown()
- */
- protected void doShutDown() throws Exception {
- su.shutDown();
- super.doShutDown();
- }
- }
-
-}
+/*
+ * 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.eip;
+
+import org.apache.servicemix.common.BaseComponent;
+import org.apache.servicemix.common.BaseLifeCycle;
+import org.apache.servicemix.common.ServiceUnit;
+import org.apache.servicemix.common.xbean.XBeanServiceUnit;
+
+/**
+ *
+ * @author gnodet
+ * @version $Revision: 376451 $
+ * @org.apache.xbean.XBean element="component"
+ * description="An EIP component"
+ */
+public class EIPSpringComponent extends BaseComponent {
+
+ private EIPEndpoint[] endpoints;
+
+ /**
+ * @return Returns the endpoints.
+ */
+ public EIPEndpoint[] getEndpoints() {
+ return endpoints;
+ }
+
+ /**
+ * @param endpoints The endpoints to set.
+ */
+ public void setEndpoints(EIPEndpoint[] endpoints) {
+ this.endpoints = endpoints;
+ }
+
+ /* (non-Javadoc)
+ * @see org.servicemix.common.BaseComponent#createLifeCycle()
+ */
+ protected BaseLifeCycle createLifeCycle() {
+ return new LifeCycle();
+ }
+
+ /**
+ * @author gnodet
+ */
+ public class LifeCycle extends EIPLifeCycle {
+
+ protected ServiceUnit su;
+
+ public LifeCycle() {
+ super(EIPSpringComponent.this);
+ }
+
+ /* (non-Javadoc)
+ * @see org.servicemix.common.BaseLifeCycle#doInit()
+ */
+ protected void doInit() throws Exception {
+ super.doInit();
+ su = new XBeanServiceUnit();
+ su.setComponent(EIPSpringComponent.this);
+ for (int i = 0; i < endpoints.length; i++) {
+ endpoints[i].setServiceUnit(su);
+ endpoints[i].validate();
+ su.addEndpoint(endpoints[i]);
+ }
+ getRegistry().registerServiceUnit(su);
+ }
+
+ /* (non-Javadoc)
+ * @see org.servicemix.common.BaseLifeCycle#doStart()
+ */
+ protected void doStart() throws Exception {
+ super.doStart();
+ su.start();
+ }
+
+ /* (non-Javadoc)
+ * @see org.servicemix.common.BaseLifeCycle#doStop()
+ */
+ protected void doStop() throws Exception {
+ su.stop();
+ super.doStop();
+ }
+
+ /* (non-Javadoc)
+ * @see org.servicemix.common.BaseLifeCycle#doShutDown()
+ */
+ protected void doShutDown() throws Exception {
+ su.shutDown();
+ super.doShutDown();
+ }
+ }
+
+}
Propchange: incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/EIPSpringComponent.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/patterns/ContentBasedRouter.java
URL: http://svn.apache.org/viewvc/incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/patterns/ContentBasedRouter.java?rev=426392&r1=426391&r2=426392&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/patterns/ContentBasedRouter.java (original)
+++ incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/patterns/ContentBasedRouter.java Thu Jul 27 22:49:32 2006
@@ -1,193 +1,193 @@
-/*
- * 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.eip.patterns;
-
-import javax.jbi.management.DeploymentException;
-import javax.jbi.messaging.ExchangeStatus;
-import javax.jbi.messaging.Fault;
-import javax.jbi.messaging.MessageExchange;
-import javax.jbi.messaging.MessagingException;
-import javax.jbi.messaging.NormalizedMessage;
-
-import org.apache.servicemix.JbiConstants;
-import org.apache.servicemix.eip.EIPEndpoint;
-import org.apache.servicemix.eip.support.ExchangeTarget;
-import org.apache.servicemix.eip.support.MessageUtil;
-import org.apache.servicemix.eip.support.RoutingRule;
-import org.apache.servicemix.store.Store;
-
-/**
- * ContentBasedRouter can be used for all kind of content-based routing.
- * This component implements the
- * <a href="http://www.enterpriseintegrationpatterns.com/ContentBasedRouter.html">Content-Based Router</a>
- * pattern.
- *
- * @author gnodet
- * @version $Revision: 376451 $
- * @org.apache.xbean.XBean element="content-based-router"
- * description="A Content-Based Router"
- */
-public class ContentBasedRouter extends EIPEndpoint {
-
- /**
- * Routing rules that are evaluated to find the target destination
- */
- private RoutingRule[] rules;
- /**
- * The correlation property used by this component
- */
- private String correlation;
-
- /**
- * @return Returns the rules.
- */
- public RoutingRule[] getRules() {
- return rules;
- }
-
- /**
- * @param rules The rules to set.
- */
- public void setRules(RoutingRule[] rules) {
- this.rules = rules;
- }
-
- /* (non-Javadoc)
- * @see org.apache.servicemix.eip.EIPEndpoint#validate()
- */
- public void validate() throws DeploymentException {
- super.validate();
- // Check rules
- if (rules == null || rules.length == 0) {
- throw new IllegalArgumentException("rules should contain at least one RoutingRule");
- }
- // Create correlation property
- correlation = "AbstractContentBasedRouter.Correlation." + getService() + "." + getEndpoint();
- }
-
- /* (non-Javadoc)
- * @see org.apache.servicemix.eip.EIPEndpoint#processSync(javax.jbi.messaging.MessageExchange)
- */
- protected void processSync(MessageExchange exchange) throws Exception {
- // Create exchange for target
- MessageExchange tme = exchangeFactory.createExchange(exchange.getPattern());
- // Now copy input to new exchange
- // We need to read the message once for finding routing target
- // so ensure we have a re-readable source
- NormalizedMessage in = MessageUtil.copyIn(exchange);
- MessageUtil.transferToIn(in, tme);
- // Retrieve target
- ExchangeTarget target = getDestination(tme);
- target.configureTarget(tme, getContext());
- // Send in to target
- sendSync(tme);
- // Send back the result
- if (tme.getStatus() == ExchangeStatus.DONE) {
- done(exchange);
- } else if (tme.getStatus() == ExchangeStatus.ERROR) {
- fail(exchange, tme.getError());
- } else if (tme.getFault() != null) {
- Fault fault = MessageUtil.copyFault(tme);
- done(tme);
- MessageUtil.transferToFault(fault, exchange);
- sendSync(exchange);
- } else if (tme.getMessage("out") != null) {
- NormalizedMessage out = MessageUtil.copyOut(tme);
- done(tme);
- MessageUtil.transferToOut(out, exchange);
- sendSync(exchange);
- } else {
- done(tme);
- throw new IllegalStateException("Exchange status is " + ExchangeStatus.ACTIVE + " but has no Out nor Fault message");
- }
- }
-
- /* (non-Javadoc)
- * @see org.apache.servicemix.eip.EIPEndpoint#processAsync(javax.jbi.messaging.MessageExchange)
- */
- protected void processAsync(MessageExchange exchange) throws Exception {
- if (exchange.getRole() == MessageExchange.Role.PROVIDER &&
- exchange.getProperty(correlation) == null) {
- // Create exchange for target
- MessageExchange tme = exchangeFactory.createExchange(exchange.getPattern());
- if (store.hasFeature(Store.CLUSTERED)) {
- exchange.setProperty(JbiConstants.STATELESS_PROVIDER, Boolean.TRUE);
- tme.setProperty(JbiConstants.STATELESS_CONSUMER, Boolean.TRUE);
- }
- // Set correlations
- tme.setProperty(correlation, exchange.getExchangeId());
- exchange.setProperty(correlation, tme.getExchangeId());
- // Put exchange to store
- store.store(exchange.getExchangeId(), exchange);
- // Now copy input to new exchange
- // We need to read the message once for finding routing target
- // so ensure we have a re-readable source
- NormalizedMessage in = MessageUtil.copyIn(exchange);
- MessageUtil.transferToIn(in, tme);
- // Retrieve target
- ExchangeTarget target = getDestination(tme);
- target.configureTarget(tme, getContext());
- // Send in to target
- send(tme);
- // Mimic the exchange on the other side and send to needed listener
- } else {
- String id = (String) exchange.getProperty(correlation);
- if (id == null) {
- throw new IllegalStateException(correlation + " property not found");
- }
- MessageExchange org = (MessageExchange) store.load(id);
- if (org == null) {
- throw new IllegalStateException("Could not load original exchange with id " + id);
- }
- // Reproduce DONE status to the other side
- if (exchange.getStatus() == ExchangeStatus.DONE) {
- done(org);
- // Reproduce ERROR status to the other side
- } else if (exchange.getStatus() == ExchangeStatus.ERROR) {
- fail(org, exchange.getError());
- // Reproduce faults to the other side and listeners
- } else if (exchange.getFault() != null) {
- store.store(exchange.getExchangeId(), exchange);
- MessageUtil.transferTo(exchange, org, "fault");
- send(org);
- // Reproduce answers to the other side
- } else if (exchange.getMessage("out") != null) {
- store.store(exchange.getExchangeId(), exchange);
- MessageUtil.transferTo(exchange, org, "out");
- send(org);
- } else {
- throw new IllegalStateException("Exchange status is " + ExchangeStatus.ACTIVE + " but has no Out nor Fault message");
- }
- }
- }
-
- /**
- * Find the target destination for the given JBI exchange
- * @param exchange
- * @return the target for the given exchange
- * @throws Exception
- */
- protected ExchangeTarget getDestination(MessageExchange exchange) throws Exception {
- for (int i = 0; i < rules.length; i++) {
- if (rules[i].getPredicate() == null ||
- rules[i].getPredicate().matches(exchange)) {
- return rules[i].getTarget();
- }
- }
- throw new MessagingException("No matching rule found for exchange");
- }
-
-}
+/*
+ * 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.eip.patterns;
+
+import javax.jbi.management.DeploymentException;
+import javax.jbi.messaging.ExchangeStatus;
+import javax.jbi.messaging.Fault;
+import javax.jbi.messaging.MessageExchange;
+import javax.jbi.messaging.MessagingException;
+import javax.jbi.messaging.NormalizedMessage;
+
+import org.apache.servicemix.JbiConstants;
+import org.apache.servicemix.eip.EIPEndpoint;
+import org.apache.servicemix.eip.support.ExchangeTarget;
+import org.apache.servicemix.eip.support.MessageUtil;
+import org.apache.servicemix.eip.support.RoutingRule;
+import org.apache.servicemix.store.Store;
+
+/**
+ * ContentBasedRouter can be used for all kind of content-based routing.
+ * This component implements the
+ * <a href="http://www.enterpriseintegrationpatterns.com/ContentBasedRouter.html">Content-Based Router</a>
+ * pattern.
+ *
+ * @author gnodet
+ * @version $Revision: 376451 $
+ * @org.apache.xbean.XBean element="content-based-router"
+ * description="A Content-Based Router"
+ */
+public class ContentBasedRouter extends EIPEndpoint {
+
+ /**
+ * Routing rules that are evaluated to find the target destination
+ */
+ private RoutingRule[] rules;
+ /**
+ * The correlation property used by this component
+ */
+ private String correlation;
+
+ /**
+ * @return Returns the rules.
+ */
+ public RoutingRule[] getRules() {
+ return rules;
+ }
+
+ /**
+ * @param rules The rules to set.
+ */
+ public void setRules(RoutingRule[] rules) {
+ this.rules = rules;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.servicemix.eip.EIPEndpoint#validate()
+ */
+ public void validate() throws DeploymentException {
+ super.validate();
+ // Check rules
+ if (rules == null || rules.length == 0) {
+ throw new IllegalArgumentException("rules should contain at least one RoutingRule");
+ }
+ // Create correlation property
+ correlation = "AbstractContentBasedRouter.Correlation." + getService() + "." + getEndpoint();
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.servicemix.eip.EIPEndpoint#processSync(javax.jbi.messaging.MessageExchange)
+ */
+ protected void processSync(MessageExchange exchange) throws Exception {
+ // Create exchange for target
+ MessageExchange tme = exchangeFactory.createExchange(exchange.getPattern());
+ // Now copy input to new exchange
+ // We need to read the message once for finding routing target
+ // so ensure we have a re-readable source
+ NormalizedMessage in = MessageUtil.copyIn(exchange);
+ MessageUtil.transferToIn(in, tme);
+ // Retrieve target
+ ExchangeTarget target = getDestination(tme);
+ target.configureTarget(tme, getContext());
+ // Send in to target
+ sendSync(tme);
+ // Send back the result
+ if (tme.getStatus() == ExchangeStatus.DONE) {
+ done(exchange);
+ } else if (tme.getStatus() == ExchangeStatus.ERROR) {
+ fail(exchange, tme.getError());
+ } else if (tme.getFault() != null) {
+ Fault fault = MessageUtil.copyFault(tme);
+ done(tme);
+ MessageUtil.transferToFault(fault, exchange);
+ sendSync(exchange);
+ } else if (tme.getMessage("out") != null) {
+ NormalizedMessage out = MessageUtil.copyOut(tme);
+ done(tme);
+ MessageUtil.transferToOut(out, exchange);
+ sendSync(exchange);
+ } else {
+ done(tme);
+ throw new IllegalStateException("Exchange status is " + ExchangeStatus.ACTIVE + " but has no Out nor Fault message");
+ }
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.servicemix.eip.EIPEndpoint#processAsync(javax.jbi.messaging.MessageExchange)
+ */
+ protected void processAsync(MessageExchange exchange) throws Exception {
+ if (exchange.getRole() == MessageExchange.Role.PROVIDER &&
+ exchange.getProperty(correlation) == null) {
+ // Create exchange for target
+ MessageExchange tme = exchangeFactory.createExchange(exchange.getPattern());
+ if (store.hasFeature(Store.CLUSTERED)) {
+ exchange.setProperty(JbiConstants.STATELESS_PROVIDER, Boolean.TRUE);
+ tme.setProperty(JbiConstants.STATELESS_CONSUMER, Boolean.TRUE);
+ }
+ // Set correlations
+ tme.setProperty(correlation, exchange.getExchangeId());
+ exchange.setProperty(correlation, tme.getExchangeId());
+ // Put exchange to store
+ store.store(exchange.getExchangeId(), exchange);
+ // Now copy input to new exchange
+ // We need to read the message once for finding routing target
+ // so ensure we have a re-readable source
+ NormalizedMessage in = MessageUtil.copyIn(exchange);
+ MessageUtil.transferToIn(in, tme);
+ // Retrieve target
+ ExchangeTarget target = getDestination(tme);
+ target.configureTarget(tme, getContext());
+ // Send in to target
+ send(tme);
+ // Mimic the exchange on the other side and send to needed listener
+ } else {
+ String id = (String) exchange.getProperty(correlation);
+ if (id == null) {
+ throw new IllegalStateException(correlation + " property not found");
+ }
+ MessageExchange org = (MessageExchange) store.load(id);
+ if (org == null) {
+ throw new IllegalStateException("Could not load original exchange with id " + id);
+ }
+ // Reproduce DONE status to the other side
+ if (exchange.getStatus() == ExchangeStatus.DONE) {
+ done(org);
+ // Reproduce ERROR status to the other side
+ } else if (exchange.getStatus() == ExchangeStatus.ERROR) {
+ fail(org, exchange.getError());
+ // Reproduce faults to the other side and listeners
+ } else if (exchange.getFault() != null) {
+ store.store(exchange.getExchangeId(), exchange);
+ MessageUtil.transferTo(exchange, org, "fault");
+ send(org);
+ // Reproduce answers to the other side
+ } else if (exchange.getMessage("out") != null) {
+ store.store(exchange.getExchangeId(), exchange);
+ MessageUtil.transferTo(exchange, org, "out");
+ send(org);
+ } else {
+ throw new IllegalStateException("Exchange status is " + ExchangeStatus.ACTIVE + " but has no Out nor Fault message");
+ }
+ }
+ }
+
+ /**
+ * Find the target destination for the given JBI exchange
+ * @param exchange
+ * @return the target for the given exchange
+ * @throws Exception
+ */
+ protected ExchangeTarget getDestination(MessageExchange exchange) throws Exception {
+ for (int i = 0; i < rules.length; i++) {
+ if (rules[i].getPredicate() == null ||
+ rules[i].getPredicate().matches(exchange)) {
+ return rules[i].getTarget();
+ }
+ }
+ throw new MessagingException("No matching rule found for exchange");
+ }
+
+}
Propchange: incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/patterns/ContentBasedRouter.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/patterns/MessageFilter.java
URL: http://svn.apache.org/viewvc/incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/patterns/MessageFilter.java?rev=426392&r1=426391&r2=426392&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/patterns/MessageFilter.java (original)
+++ incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/patterns/MessageFilter.java Thu Jul 27 22:49:32 2006
@@ -1,192 +1,192 @@
-/*
- * 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.eip.patterns;
-
-import javax.jbi.management.DeploymentException;
-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.RobustInOnly;
-
-import org.apache.servicemix.eip.EIPEndpoint;
-import org.apache.servicemix.eip.support.ExchangeTarget;
-import org.apache.servicemix.eip.support.MessageUtil;
-import org.apache.servicemix.eip.support.Predicate;
-
-/**
- * MessageFilter allows filtering incoming JBI exchanges.
- * This component implements the
- * <a href="http://www.enterpriseintegrationpatterns.com/Filter.html">Message Filter</a>
- * pattern.
- *
- * @author gnodet
- * @version $Revision: 376451 $
- * @org.apache.xbean.XBean element="message-filter"
- * description="A Message Filter"
- */
-public class MessageFilter extends EIPEndpoint {
-
- /**
- * The main target destination which will receive the exchange
- */
- private ExchangeTarget target;
- /**
- * The filter to use on incoming messages
- */
- private Predicate filter;
- /**
- * The correlation property used by this component
- */
- private String correlation;
- /**
- * Indicates if faults and errors from recipients should be sent
- * back to the consumer. In such a case, only the first fault or
- * error received will be reported.
- * Note that if the consumer is synchronous, it will be blocked
- * until all recipients successfully acked the exchange, or
- * a fault or error is reported, and the exchange will be kept in the
- * store for recovery.
- */
- private boolean reportErrors;
-
- /**
- * @return Returns the target.
- */
- public ExchangeTarget getTarget() {
- return target;
- }
-
- /**
- * @param target The target to set.
- */
- public void setTarget(ExchangeTarget target) {
- this.target = target;
- }
-
- /**
- * @return Returns the filter.
- */
- public Predicate getFilter() {
- return filter;
- }
-
- /**
- * @param filter The filter to set.
- */
- public void setFilter(Predicate filter) {
- this.filter = filter;
- }
-
- /**
- * @return Returns the reportErrors.
- */
- public boolean isReportErrors() {
- return reportErrors;
- }
-
- /**
- * @param reportErrors The reportErrors to set.
- */
- public void setReportErrors(boolean reportErrors) {
- this.reportErrors = reportErrors;
- }
-
- /* (non-Javadoc)
- * @see org.apache.servicemix.eip.EIPEndpoint#validate()
- */
- public void validate() throws DeploymentException {
- super.validate();
- // Check target
- if (target == null) {
- throw new IllegalArgumentException("target should be set to a valid ExchangeTarget");
- }
- // Check filter
- if (filter == null) {
- throw new IllegalArgumentException("filter property should be set");
- }
- // Create correlation property
- correlation = "MessageFilter.Correlation." + getService() + "." + getEndpoint();
- }
-
- /* (non-Javadoc)
- * @see org.apache.servicemix.eip.EIPEndpoint#processSync(javax.jbi.messaging.MessageExchange)
- */
- protected void processSync(MessageExchange exchange) throws Exception {
- if (exchange instanceof InOnly == false &&
- exchange instanceof RobustInOnly == false) {
- fail(exchange, new UnsupportedOperationException("Use an InOnly or RobustInOnly MEP"));
- } else {
- NormalizedMessage in = MessageUtil.copyIn(exchange);
- MessageExchange me = exchangeFactory.createExchange(exchange.getPattern());
- target.configureTarget(me, getContext());
- MessageUtil.transferToIn(in, me);
- if (filter.matches(me)) {
- sendSync(me);
- if (me.getStatus() == ExchangeStatus.ERROR && reportErrors) {
- fail(exchange, me.getError());
- } else if (me.getStatus() == ExchangeStatus.DONE) {
- done(exchange);
- } else if (me.getFault() != null && reportErrors) {
- Fault fault = MessageUtil.copyFault(me);
- done(me);
- MessageUtil.transferToFault(fault, exchange);
- sendSync(exchange);
- }
- } else {
- done(exchange);
- }
- }
- }
-
- /* (non-Javadoc)
- * @see org.apache.servicemix.eip.EIPEndpoint#processAsync(javax.jbi.messaging.MessageExchange)
- */
- protected void processAsync(MessageExchange exchange) throws Exception {
- // If we need to report errors, the behavior is really different,
- // as we need to keep the incoming exchange in the store until
- // all acks have been received
- if (reportErrors) {
- // TODO: implement this
- throw new UnsupportedOperationException("Not implemented");
- // We are in a simple fire-and-forget behaviour.
- // This implementation is really efficient as we do not use
- // the store at all.
- } else {
- if (exchange.getStatus() == ExchangeStatus.DONE) {
- return;
- } else if (exchange.getStatus() == ExchangeStatus.ERROR) {
- return;
- } else if (exchange instanceof InOnly == false &&
- exchange instanceof RobustInOnly == false) {
- fail(exchange, new UnsupportedOperationException("Use an InOnly or RobustInOnly MEP"));
- } else if (exchange.getFault() != null) {
- done(exchange);
- } else {
- NormalizedMessage in = MessageUtil.copyIn(exchange);
- MessageExchange me = exchangeFactory.createExchange(exchange.getPattern());
- target.configureTarget(me, getContext());
- MessageUtil.transferToIn(in, me);
- if (filter.matches(me)) {
- send(me);
- }
- done(exchange);
- }
- }
- }
-
-}
+/*
+ * 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.eip.patterns;
+
+import javax.jbi.management.DeploymentException;
+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.RobustInOnly;
+
+import org.apache.servicemix.eip.EIPEndpoint;
+import org.apache.servicemix.eip.support.ExchangeTarget;
+import org.apache.servicemix.eip.support.MessageUtil;
+import org.apache.servicemix.eip.support.Predicate;
+
+/**
+ * MessageFilter allows filtering incoming JBI exchanges.
+ * This component implements the
+ * <a href="http://www.enterpriseintegrationpatterns.com/Filter.html">Message Filter</a>
+ * pattern.
+ *
+ * @author gnodet
+ * @version $Revision: 376451 $
+ * @org.apache.xbean.XBean element="message-filter"
+ * description="A Message Filter"
+ */
+public class MessageFilter extends EIPEndpoint {
+
+ /**
+ * The main target destination which will receive the exchange
+ */
+ private ExchangeTarget target;
+ /**
+ * The filter to use on incoming messages
+ */
+ private Predicate filter;
+ /**
+ * The correlation property used by this component
+ */
+ private String correlation;
+ /**
+ * Indicates if faults and errors from recipients should be sent
+ * back to the consumer. In such a case, only the first fault or
+ * error received will be reported.
+ * Note that if the consumer is synchronous, it will be blocked
+ * until all recipients successfully acked the exchange, or
+ * a fault or error is reported, and the exchange will be kept in the
+ * store for recovery.
+ */
+ private boolean reportErrors;
+
+ /**
+ * @return Returns the target.
+ */
+ public ExchangeTarget getTarget() {
+ return target;
+ }
+
+ /**
+ * @param target The target to set.
+ */
+ public void setTarget(ExchangeTarget target) {
+ this.target = target;
+ }
+
+ /**
+ * @return Returns the filter.
+ */
+ public Predicate getFilter() {
+ return filter;
+ }
+
+ /**
+ * @param filter The filter to set.
+ */
+ public void setFilter(Predicate filter) {
+ this.filter = filter;
+ }
+
+ /**
+ * @return Returns the reportErrors.
+ */
+ public boolean isReportErrors() {
+ return reportErrors;
+ }
+
+ /**
+ * @param reportErrors The reportErrors to set.
+ */
+ public void setReportErrors(boolean reportErrors) {
+ this.reportErrors = reportErrors;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.servicemix.eip.EIPEndpoint#validate()
+ */
+ public void validate() throws DeploymentException {
+ super.validate();
+ // Check target
+ if (target == null) {
+ throw new IllegalArgumentException("target should be set to a valid ExchangeTarget");
+ }
+ // Check filter
+ if (filter == null) {
+ throw new IllegalArgumentException("filter property should be set");
+ }
+ // Create correlation property
+ correlation = "MessageFilter.Correlation." + getService() + "." + getEndpoint();
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.servicemix.eip.EIPEndpoint#processSync(javax.jbi.messaging.MessageExchange)
+ */
+ protected void processSync(MessageExchange exchange) throws Exception {
+ if (exchange instanceof InOnly == false &&
+ exchange instanceof RobustInOnly == false) {
+ fail(exchange, new UnsupportedOperationException("Use an InOnly or RobustInOnly MEP"));
+ } else {
+ NormalizedMessage in = MessageUtil.copyIn(exchange);
+ MessageExchange me = exchangeFactory.createExchange(exchange.getPattern());
+ target.configureTarget(me, getContext());
+ MessageUtil.transferToIn(in, me);
+ if (filter.matches(me)) {
+ sendSync(me);
+ if (me.getStatus() == ExchangeStatus.ERROR && reportErrors) {
+ fail(exchange, me.getError());
+ } else if (me.getStatus() == ExchangeStatus.DONE) {
+ done(exchange);
+ } else if (me.getFault() != null && reportErrors) {
+ Fault fault = MessageUtil.copyFault(me);
+ done(me);
+ MessageUtil.transferToFault(fault, exchange);
+ sendSync(exchange);
+ }
+ } else {
+ done(exchange);
+ }
+ }
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.servicemix.eip.EIPEndpoint#processAsync(javax.jbi.messaging.MessageExchange)
+ */
+ protected void processAsync(MessageExchange exchange) throws Exception {
+ // If we need to report errors, the behavior is really different,
+ // as we need to keep the incoming exchange in the store until
+ // all acks have been received
+ if (reportErrors) {
+ // TODO: implement this
+ throw new UnsupportedOperationException("Not implemented");
+ // We are in a simple fire-and-forget behaviour.
+ // This implementation is really efficient as we do not use
+ // the store at all.
+ } else {
+ if (exchange.getStatus() == ExchangeStatus.DONE) {
+ return;
+ } else if (exchange.getStatus() == ExchangeStatus.ERROR) {
+ return;
+ } else if (exchange instanceof InOnly == false &&
+ exchange instanceof RobustInOnly == false) {
+ fail(exchange, new UnsupportedOperationException("Use an InOnly or RobustInOnly MEP"));
+ } else if (exchange.getFault() != null) {
+ done(exchange);
+ } else {
+ NormalizedMessage in = MessageUtil.copyIn(exchange);
+ MessageExchange me = exchangeFactory.createExchange(exchange.getPattern());
+ target.configureTarget(me, getContext());
+ MessageUtil.transferToIn(in, me);
+ if (filter.matches(me)) {
+ send(me);
+ }
+ done(exchange);
+ }
+ }
+ }
+
+}
Propchange: incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/patterns/MessageFilter.java
------------------------------------------------------------------------------
svn:eol-style = native