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 [23/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/patterns/StaticRoutingSlip.java
URL: http://svn.apache.org/viewvc/incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/patterns/StaticRoutingSlip.java?rev=426392&r1=426391&r2=426392&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/patterns/StaticRoutingSlip.java (original)
+++ incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/patterns/StaticRoutingSlip.java Thu Jul 27 22:49:32 2006
@@ -1,237 +1,237 @@
-/*
- * 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.InOut;
-import javax.jbi.messaging.MessageExchange;
-
-import org.apache.servicemix.eip.EIPEndpoint;
-import org.apache.servicemix.eip.support.ExchangeTarget;
-import org.apache.servicemix.eip.support.MessageUtil;
-
-/**
- * A RoutingSlip component can be used to route an incoming In-Out exchange
- * through a series of target services.
- * This component implements the
- * <a href="http://www.enterpriseintegrationpatterns.com/RoutingTable.html">Routing Slip</a>
- * pattern, with the limitation that the routing table is static.
- * This component only uses In-Out MEPs and errors or faults sent by targets are reported
- * back to the consumer, thus interrupting the routing process.
- * In addition, this component is fully asynchronous and uses an exchange store to provide
- * full HA and recovery for clustered / persistent flows.
- *
- * @author gnodet
- * @version $Revision: 376451 $
- * @org.apache.xbean.XBean element="static-routing-slip"
- * description="A static Routing Slip"
- */
-public class StaticRoutingSlip extends EIPEndpoint {
-
- /**
- * List of target components used in the RoutingSlip
- */
- private ExchangeTarget[] targets;
- /**
- * The correlation property used by this component
- */
- private String correlation;
- /**
- * The current index of the target
- */
- private String index;
- /**
- * The id of the previous target exchange
- */
- private String previous;
-
- /**
- * @return Returns the targets.
- */
- public ExchangeTarget[] getTargets() {
- return targets;
- }
-
- /**
- * @param targets The targets to set.
- */
- public void setTargets(ExchangeTarget[] targets) {
- this.targets = targets;
- }
-
- /* (non-Javadoc)
- * @see org.apache.servicemix.eip.EIPEndpoint#validate()
- */
- public void validate() throws DeploymentException {
- super.validate();
- // Check target
- if (targets == null || targets.length == 0) {
- throw new IllegalArgumentException("targets should contain at least one ExchangeTarget");
- }
- // Create correlation properties
- correlation = "RoutingSlip.Correlation." + getService() + "." + getEndpoint();
- index = "RoutingSlip.Index." + getService() + "." + getEndpoint();
- previous = "RoutingSlip.Previous." + 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 InOut == false) {
- throw new IllegalStateException("Use an InOut MEP");
- }
- MessageExchange current = exchange;
- for (int i = 0; i < targets.length; i++) {
- InOut me = exchangeFactory.createInOutExchange();
- targets[i].configureTarget(me, getContext());
- if (i == 0) {
- MessageUtil.transferInToIn(current, me);
- } else {
- MessageUtil.transferOutToIn(current, me);
- }
- sendSync(me);
- if (i != 0) {
- done(current);
- }
- if (me.getStatus() == ExchangeStatus.DONE) {
- throw new IllegalStateException("Exchange status is " + ExchangeStatus.DONE);
- } else if (me.getStatus() == ExchangeStatus.ERROR) {
- fail(exchange, me.getError());
- return;
- } else if (me.getFault() != null) {
- Fault fault = MessageUtil.copyFault(me);
- MessageUtil.transferToFault(fault, exchange);
- done(me);
- sendSync(exchange);
- return;
- } else if (me.getOutMessage() == null) {
- throw new IllegalStateException("Exchange status is " + ExchangeStatus.ACTIVE + " but has no Out nor Fault message");
- }
- current = me;
- }
- MessageUtil.transferToOut(MessageUtil.copyOut(current), exchange);
- done(current);
- sendSync(exchange);
- }
-
- /* (non-Javadoc)
- * @see org.apache.servicemix.eip.EIPEndpoint#processAsync(javax.jbi.messaging.MessageExchange)
- */
- protected void processAsync(MessageExchange exchange) throws Exception {
- // This exchange comes from the consumer
- if (exchange.getRole() == MessageExchange.Role.PROVIDER) {
- if (exchange.getStatus() == ExchangeStatus.DONE) {
- String correlationId = (String) exchange.getProperty(correlation);
- if (correlationId == null) {
- throw new IllegalStateException(correlation + " property not found");
- }
- // Ack last target hit
- MessageExchange me = (MessageExchange) store.load(correlationId);
- done(me);
- } else if (exchange.getStatus() == ExchangeStatus.ERROR) {
- String correlationId = (String) exchange.getProperty(correlation);
- if (correlationId == null) {
- throw new IllegalStateException(correlation + " property not found");
- }
- // Ack last target hit
- MessageExchange me = (MessageExchange) store.load(correlationId);
- done(me);
- } else if (exchange instanceof InOut == false) {
- throw new IllegalStateException("Use an InOut MEP");
- } else {
- MessageExchange me = exchangeFactory.createInOutExchange();
- me.setProperty(correlation, exchange.getExchangeId());
- me.setProperty(index, new Integer(0));
- targets[0].configureTarget(me, getContext());
- store.store(exchange.getExchangeId(), exchange);
- MessageUtil.transferInToIn(exchange, me);
- send(me);
- }
- // The exchange comes from a target
- } else {
- String correlationId = (String) exchange.getProperty(correlation);
- String previousId = (String) exchange.getProperty(previous);
- Integer prevIndex = (Integer) exchange.getProperty(index);
- if (correlationId == null) {
- throw new IllegalStateException(correlation + " property not found");
- }
- if (prevIndex == null) {
- throw new IllegalStateException(previous + " property not found");
- }
- // This should never happen, as we can only send DONE
- if (exchange.getStatus() == ExchangeStatus.DONE) {
- throw new IllegalStateException("Exchange status is " + ExchangeStatus.DONE);
- // ERROR are sent back to the consumer
- } else if (exchange.getStatus() == ExchangeStatus.ERROR) {
- MessageExchange me = (MessageExchange) store.load(correlationId);
- fail(me, exchange.getError());
- // Ack the previous target
- if (previousId != null) {
- me = (MessageExchange) store.load(previousId);
- done(me);
- }
- // Faults are sent back to the consumer
- } else if (exchange.getFault() != null) {
- MessageExchange me = (MessageExchange) store.load(correlationId);
- me.setProperty(correlation, exchange.getExchangeId());
- store.store(exchange.getExchangeId(), exchange);
- MessageUtil.transferFaultToFault(exchange, me);
- send(me);
- // Ack the previous target
- if (previousId != null) {
- me = (MessageExchange) store.load(previousId);
- done(me);
- }
- // Out message, give it to next target or back to consumer
- } else if (exchange.getMessage("out") != null) {
- // This is the answer from the last target
- if (prevIndex.intValue() == targets.length - 1) {
- MessageExchange me = (MessageExchange) store.load(correlationId);
- me.setProperty(correlation, exchange.getExchangeId());
- store.store(exchange.getExchangeId(), exchange);
- MessageUtil.transferOutToOut(exchange, me);
- send(me);
- if (previousId != null) {
- me = (MessageExchange) store.load(previousId);
- done(me);
- }
- // We still have a target to hit
- } else {
- MessageExchange me = exchangeFactory.createInOutExchange();
- Integer curIndex = new Integer(prevIndex.intValue() + 1);
- me.setProperty(correlation, correlationId);
- me.setProperty(index, curIndex);
- me.setProperty(previous, exchange.getExchangeId());
- targets[curIndex.intValue()].configureTarget(me, getContext());
- store.store(exchange.getExchangeId(), exchange);
- MessageUtil.transferOutToIn(exchange, me);
- send(me);
- if (previousId != null) {
- me = (MessageExchange) store.load(previousId);
- done(me);
- }
- }
- // This should not happen
- } else {
- throw new IllegalStateException("Exchange status is " + ExchangeStatus.ACTIVE + " but has no Out nor Fault message");
- }
- }
- }
-
-}
+/*
+ * 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.InOut;
+import javax.jbi.messaging.MessageExchange;
+
+import org.apache.servicemix.eip.EIPEndpoint;
+import org.apache.servicemix.eip.support.ExchangeTarget;
+import org.apache.servicemix.eip.support.MessageUtil;
+
+/**
+ * A RoutingSlip component can be used to route an incoming In-Out exchange
+ * through a series of target services.
+ * This component implements the
+ * <a href="http://www.enterpriseintegrationpatterns.com/RoutingTable.html">Routing Slip</a>
+ * pattern, with the limitation that the routing table is static.
+ * This component only uses In-Out MEPs and errors or faults sent by targets are reported
+ * back to the consumer, thus interrupting the routing process.
+ * In addition, this component is fully asynchronous and uses an exchange store to provide
+ * full HA and recovery for clustered / persistent flows.
+ *
+ * @author gnodet
+ * @version $Revision: 376451 $
+ * @org.apache.xbean.XBean element="static-routing-slip"
+ * description="A static Routing Slip"
+ */
+public class StaticRoutingSlip extends EIPEndpoint {
+
+ /**
+ * List of target components used in the RoutingSlip
+ */
+ private ExchangeTarget[] targets;
+ /**
+ * The correlation property used by this component
+ */
+ private String correlation;
+ /**
+ * The current index of the target
+ */
+ private String index;
+ /**
+ * The id of the previous target exchange
+ */
+ private String previous;
+
+ /**
+ * @return Returns the targets.
+ */
+ public ExchangeTarget[] getTargets() {
+ return targets;
+ }
+
+ /**
+ * @param targets The targets to set.
+ */
+ public void setTargets(ExchangeTarget[] targets) {
+ this.targets = targets;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.servicemix.eip.EIPEndpoint#validate()
+ */
+ public void validate() throws DeploymentException {
+ super.validate();
+ // Check target
+ if (targets == null || targets.length == 0) {
+ throw new IllegalArgumentException("targets should contain at least one ExchangeTarget");
+ }
+ // Create correlation properties
+ correlation = "RoutingSlip.Correlation." + getService() + "." + getEndpoint();
+ index = "RoutingSlip.Index." + getService() + "." + getEndpoint();
+ previous = "RoutingSlip.Previous." + 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 InOut == false) {
+ throw new IllegalStateException("Use an InOut MEP");
+ }
+ MessageExchange current = exchange;
+ for (int i = 0; i < targets.length; i++) {
+ InOut me = exchangeFactory.createInOutExchange();
+ targets[i].configureTarget(me, getContext());
+ if (i == 0) {
+ MessageUtil.transferInToIn(current, me);
+ } else {
+ MessageUtil.transferOutToIn(current, me);
+ }
+ sendSync(me);
+ if (i != 0) {
+ done(current);
+ }
+ if (me.getStatus() == ExchangeStatus.DONE) {
+ throw new IllegalStateException("Exchange status is " + ExchangeStatus.DONE);
+ } else if (me.getStatus() == ExchangeStatus.ERROR) {
+ fail(exchange, me.getError());
+ return;
+ } else if (me.getFault() != null) {
+ Fault fault = MessageUtil.copyFault(me);
+ MessageUtil.transferToFault(fault, exchange);
+ done(me);
+ sendSync(exchange);
+ return;
+ } else if (me.getOutMessage() == null) {
+ throw new IllegalStateException("Exchange status is " + ExchangeStatus.ACTIVE + " but has no Out nor Fault message");
+ }
+ current = me;
+ }
+ MessageUtil.transferToOut(MessageUtil.copyOut(current), exchange);
+ done(current);
+ sendSync(exchange);
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.servicemix.eip.EIPEndpoint#processAsync(javax.jbi.messaging.MessageExchange)
+ */
+ protected void processAsync(MessageExchange exchange) throws Exception {
+ // This exchange comes from the consumer
+ if (exchange.getRole() == MessageExchange.Role.PROVIDER) {
+ if (exchange.getStatus() == ExchangeStatus.DONE) {
+ String correlationId = (String) exchange.getProperty(correlation);
+ if (correlationId == null) {
+ throw new IllegalStateException(correlation + " property not found");
+ }
+ // Ack last target hit
+ MessageExchange me = (MessageExchange) store.load(correlationId);
+ done(me);
+ } else if (exchange.getStatus() == ExchangeStatus.ERROR) {
+ String correlationId = (String) exchange.getProperty(correlation);
+ if (correlationId == null) {
+ throw new IllegalStateException(correlation + " property not found");
+ }
+ // Ack last target hit
+ MessageExchange me = (MessageExchange) store.load(correlationId);
+ done(me);
+ } else if (exchange instanceof InOut == false) {
+ throw new IllegalStateException("Use an InOut MEP");
+ } else {
+ MessageExchange me = exchangeFactory.createInOutExchange();
+ me.setProperty(correlation, exchange.getExchangeId());
+ me.setProperty(index, new Integer(0));
+ targets[0].configureTarget(me, getContext());
+ store.store(exchange.getExchangeId(), exchange);
+ MessageUtil.transferInToIn(exchange, me);
+ send(me);
+ }
+ // The exchange comes from a target
+ } else {
+ String correlationId = (String) exchange.getProperty(correlation);
+ String previousId = (String) exchange.getProperty(previous);
+ Integer prevIndex = (Integer) exchange.getProperty(index);
+ if (correlationId == null) {
+ throw new IllegalStateException(correlation + " property not found");
+ }
+ if (prevIndex == null) {
+ throw new IllegalStateException(previous + " property not found");
+ }
+ // This should never happen, as we can only send DONE
+ if (exchange.getStatus() == ExchangeStatus.DONE) {
+ throw new IllegalStateException("Exchange status is " + ExchangeStatus.DONE);
+ // ERROR are sent back to the consumer
+ } else if (exchange.getStatus() == ExchangeStatus.ERROR) {
+ MessageExchange me = (MessageExchange) store.load(correlationId);
+ fail(me, exchange.getError());
+ // Ack the previous target
+ if (previousId != null) {
+ me = (MessageExchange) store.load(previousId);
+ done(me);
+ }
+ // Faults are sent back to the consumer
+ } else if (exchange.getFault() != null) {
+ MessageExchange me = (MessageExchange) store.load(correlationId);
+ me.setProperty(correlation, exchange.getExchangeId());
+ store.store(exchange.getExchangeId(), exchange);
+ MessageUtil.transferFaultToFault(exchange, me);
+ send(me);
+ // Ack the previous target
+ if (previousId != null) {
+ me = (MessageExchange) store.load(previousId);
+ done(me);
+ }
+ // Out message, give it to next target or back to consumer
+ } else if (exchange.getMessage("out") != null) {
+ // This is the answer from the last target
+ if (prevIndex.intValue() == targets.length - 1) {
+ MessageExchange me = (MessageExchange) store.load(correlationId);
+ me.setProperty(correlation, exchange.getExchangeId());
+ store.store(exchange.getExchangeId(), exchange);
+ MessageUtil.transferOutToOut(exchange, me);
+ send(me);
+ if (previousId != null) {
+ me = (MessageExchange) store.load(previousId);
+ done(me);
+ }
+ // We still have a target to hit
+ } else {
+ MessageExchange me = exchangeFactory.createInOutExchange();
+ Integer curIndex = new Integer(prevIndex.intValue() + 1);
+ me.setProperty(correlation, correlationId);
+ me.setProperty(index, curIndex);
+ me.setProperty(previous, exchange.getExchangeId());
+ targets[curIndex.intValue()].configureTarget(me, getContext());
+ store.store(exchange.getExchangeId(), exchange);
+ MessageUtil.transferOutToIn(exchange, me);
+ send(me);
+ if (previousId != null) {
+ me = (MessageExchange) store.load(previousId);
+ done(me);
+ }
+ }
+ // This should not happen
+ } else {
+ throw new IllegalStateException("Exchange status is " + ExchangeStatus.ACTIVE + " but has no Out nor Fault message");
+ }
+ }
+ }
+
+}
Propchange: incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/patterns/StaticRoutingSlip.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/patterns/WireTap.java
URL: http://svn.apache.org/viewvc/incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/patterns/WireTap.java?rev=426392&r1=426391&r2=426392&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/patterns/WireTap.java (original)
+++ incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/patterns/WireTap.java Thu Jul 27 22:49:32 2006
@@ -1,261 +1,261 @@
-/*
- * 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.JBIException;
-import javax.jbi.management.DeploymentException;
-import javax.jbi.messaging.ExchangeStatus;
-import javax.jbi.messaging.InOnly;
-import javax.jbi.messaging.MessageExchange;
-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.store.Store;
-import org.w3c.dom.Document;
-
-/**
- *
- * A WireTap component can be used to forward a copy of the input message to a listener.
- * This component implements the
- * <a href="http://www.enterpriseintegrationpatterns.com/WireTap.html">WireTap</a>
- * pattern.
- * It can handle all 4 standard MEPs, but will only send an In-Only MEP to the listener.
- * In addition, this component is fully asynchronous and uses an exchange store to provide
- * full HA and recovery for clustered / persistent flows.
- *
- * @author gnodet
- * @version $Revision: 376451 $
- * @org.apache.xbean.XBean element="wire-tap"
- * description="A WireTap"
- */
-public class WireTap extends EIPEndpoint {
-
- /**
- * The main target destination which will receive the exchange
- */
- private ExchangeTarget target;
- /**
- * The listener destination for in messages
- */
- private ExchangeTarget inListener;
- /**
- * The listener destination for out messages
- */
- private ExchangeTarget outListener;
- /**
- * The listener destination for fault messages
- */
- private ExchangeTarget faultListener;
- /**
- * The correlation property used by this component
- */
- private String correlation;
-
- /**
- * @return Returns the target.
- */
- public ExchangeTarget getTarget() {
- return target;
- }
-
- /**
- * @param target The target to set.
- */
- public void setTarget(ExchangeTarget target) {
- this.target = target;
- this.wsdlExchangeTarget = target;
- }
-
- /**
- * @return Returns the faultListener.
- */
- public ExchangeTarget getFaultListener() {
- return faultListener;
- }
-
- /**
- * @param faultListener The faultListener to set.
- */
- public void setFaultListener(ExchangeTarget faultListener) {
- this.faultListener = faultListener;
- }
-
- /**
- * @return Returns the inListener.
- */
- public ExchangeTarget getInListener() {
- return inListener;
- }
-
- /**
- * @param inListener The inListener to set.
- */
- public void setInListener(ExchangeTarget inListener) {
- this.inListener = inListener;
- }
-
- /**
- * @return Returns the outListener.
- */
- public ExchangeTarget getOutListener() {
- return outListener;
- }
-
- /**
- * @param outListener The outListener to set.
- */
- public void setOutListener(ExchangeTarget outListener) {
- this.outListener = outListener;
- }
-
- /* (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");
- }
- // Create correlation property
- correlation = "WireTap.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());
- target.configureTarget(tme, getContext());
- sendSyncToListenerAndTarget(exchange, tme, inListener, "in");
- if (tme.getStatus() == ExchangeStatus.DONE) {
- done(exchange);
- } else if (tme.getStatus() == ExchangeStatus.ERROR) {
- fail(exchange, tme.getError());
- } else if (tme.getFault() != null) {
- sendSyncToListenerAndTarget(tme, exchange, faultListener, "fault");
- done(tme);
- } else if (tme.getMessage("out") != null) {
- sendSyncToListenerAndTarget(tme, exchange, outListener, "out");
- done(tme);
- } 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);
- }
- target.configureTarget(tme, getContext());
- // Set correlations
- exchange.setProperty(correlation, tme.getExchangeId());
- tme.setProperty(correlation, exchange.getExchangeId());
- // Put exchange to store
- store.store(exchange.getExchangeId(), exchange);
- // Send in to listener and target
- sendToListenerAndTarget(exchange, tme, inListener, "in");
- // Mimic the exchange on the other side and send to needed listener
- } else {
- String id = (String) exchange.getProperty(correlation);
- if (id == null) {
- if (exchange.getRole() == MessageExchange.Role.CONSUMER &&
- exchange.getStatus() != ExchangeStatus.ACTIVE) {
- // This must be a listener status, so ignore
- return;
- }
- 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);
- sendToListenerAndTarget(exchange, org, faultListener, "fault");
- // Reproduce answers to the other side
- } else if (exchange.getMessage("out") != null) {
- store.store(exchange.getExchangeId(), exchange);
- sendToListenerAndTarget(exchange, org, outListener, "out");
- } else {
- throw new IllegalStateException("Exchange status is " + ExchangeStatus.ACTIVE + " but has no Out nor Fault message");
- }
- }
- }
-
- private void sendToListenerAndTarget(MessageExchange source,
- MessageExchange dest,
- ExchangeTarget listener,
- String message) throws Exception {
- if (listener != null) {
- NormalizedMessage msg = MessageUtil.copy(source.getMessage(message));
- InOnly lme = exchangeFactory.createInOnlyExchange();
- if (store.hasFeature(Store.CLUSTERED)) {
- lme.setProperty(JbiConstants.STATELESS_CONSUMER, Boolean.TRUE);
- }
- listener.configureTarget(lme, getContext());
- MessageUtil.transferToIn(msg, lme);
- send(lme);
- MessageUtil.transferTo(msg, dest, message);
- send(dest);
- } else {
- MessageUtil.transferTo(source, dest, message);
- send(dest);
- }
- }
-
- private void sendSyncToListenerAndTarget(MessageExchange source,
- MessageExchange dest,
- ExchangeTarget listener,
- String message) throws Exception {
- if (listener != null) {
- NormalizedMessage msg = MessageUtil.copy(source.getMessage(message));
- InOnly lme = exchangeFactory.createInOnlyExchange();
- if (store.hasFeature(Store.CLUSTERED)) {
- lme.setProperty(JbiConstants.STATELESS_CONSUMER, Boolean.TRUE);
- }
- listener.configureTarget(lme, getContext());
- MessageUtil.transferToIn(msg, lme);
- sendSync(lme);
- MessageUtil.transferTo(msg, dest, message);
- sendSync(dest);
- } else {
- MessageUtil.transferTo(source, dest, message);
- sendSync(dest);
- }
- }
-
-}
+/*
+ * 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.JBIException;
+import javax.jbi.management.DeploymentException;
+import javax.jbi.messaging.ExchangeStatus;
+import javax.jbi.messaging.InOnly;
+import javax.jbi.messaging.MessageExchange;
+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.store.Store;
+import org.w3c.dom.Document;
+
+/**
+ *
+ * A WireTap component can be used to forward a copy of the input message to a listener.
+ * This component implements the
+ * <a href="http://www.enterpriseintegrationpatterns.com/WireTap.html">WireTap</a>
+ * pattern.
+ * It can handle all 4 standard MEPs, but will only send an In-Only MEP to the listener.
+ * In addition, this component is fully asynchronous and uses an exchange store to provide
+ * full HA and recovery for clustered / persistent flows.
+ *
+ * @author gnodet
+ * @version $Revision: 376451 $
+ * @org.apache.xbean.XBean element="wire-tap"
+ * description="A WireTap"
+ */
+public class WireTap extends EIPEndpoint {
+
+ /**
+ * The main target destination which will receive the exchange
+ */
+ private ExchangeTarget target;
+ /**
+ * The listener destination for in messages
+ */
+ private ExchangeTarget inListener;
+ /**
+ * The listener destination for out messages
+ */
+ private ExchangeTarget outListener;
+ /**
+ * The listener destination for fault messages
+ */
+ private ExchangeTarget faultListener;
+ /**
+ * The correlation property used by this component
+ */
+ private String correlation;
+
+ /**
+ * @return Returns the target.
+ */
+ public ExchangeTarget getTarget() {
+ return target;
+ }
+
+ /**
+ * @param target The target to set.
+ */
+ public void setTarget(ExchangeTarget target) {
+ this.target = target;
+ this.wsdlExchangeTarget = target;
+ }
+
+ /**
+ * @return Returns the faultListener.
+ */
+ public ExchangeTarget getFaultListener() {
+ return faultListener;
+ }
+
+ /**
+ * @param faultListener The faultListener to set.
+ */
+ public void setFaultListener(ExchangeTarget faultListener) {
+ this.faultListener = faultListener;
+ }
+
+ /**
+ * @return Returns the inListener.
+ */
+ public ExchangeTarget getInListener() {
+ return inListener;
+ }
+
+ /**
+ * @param inListener The inListener to set.
+ */
+ public void setInListener(ExchangeTarget inListener) {
+ this.inListener = inListener;
+ }
+
+ /**
+ * @return Returns the outListener.
+ */
+ public ExchangeTarget getOutListener() {
+ return outListener;
+ }
+
+ /**
+ * @param outListener The outListener to set.
+ */
+ public void setOutListener(ExchangeTarget outListener) {
+ this.outListener = outListener;
+ }
+
+ /* (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");
+ }
+ // Create correlation property
+ correlation = "WireTap.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());
+ target.configureTarget(tme, getContext());
+ sendSyncToListenerAndTarget(exchange, tme, inListener, "in");
+ if (tme.getStatus() == ExchangeStatus.DONE) {
+ done(exchange);
+ } else if (tme.getStatus() == ExchangeStatus.ERROR) {
+ fail(exchange, tme.getError());
+ } else if (tme.getFault() != null) {
+ sendSyncToListenerAndTarget(tme, exchange, faultListener, "fault");
+ done(tme);
+ } else if (tme.getMessage("out") != null) {
+ sendSyncToListenerAndTarget(tme, exchange, outListener, "out");
+ done(tme);
+ } 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);
+ }
+ target.configureTarget(tme, getContext());
+ // Set correlations
+ exchange.setProperty(correlation, tme.getExchangeId());
+ tme.setProperty(correlation, exchange.getExchangeId());
+ // Put exchange to store
+ store.store(exchange.getExchangeId(), exchange);
+ // Send in to listener and target
+ sendToListenerAndTarget(exchange, tme, inListener, "in");
+ // Mimic the exchange on the other side and send to needed listener
+ } else {
+ String id = (String) exchange.getProperty(correlation);
+ if (id == null) {
+ if (exchange.getRole() == MessageExchange.Role.CONSUMER &&
+ exchange.getStatus() != ExchangeStatus.ACTIVE) {
+ // This must be a listener status, so ignore
+ return;
+ }
+ 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);
+ sendToListenerAndTarget(exchange, org, faultListener, "fault");
+ // Reproduce answers to the other side
+ } else if (exchange.getMessage("out") != null) {
+ store.store(exchange.getExchangeId(), exchange);
+ sendToListenerAndTarget(exchange, org, outListener, "out");
+ } else {
+ throw new IllegalStateException("Exchange status is " + ExchangeStatus.ACTIVE + " but has no Out nor Fault message");
+ }
+ }
+ }
+
+ private void sendToListenerAndTarget(MessageExchange source,
+ MessageExchange dest,
+ ExchangeTarget listener,
+ String message) throws Exception {
+ if (listener != null) {
+ NormalizedMessage msg = MessageUtil.copy(source.getMessage(message));
+ InOnly lme = exchangeFactory.createInOnlyExchange();
+ if (store.hasFeature(Store.CLUSTERED)) {
+ lme.setProperty(JbiConstants.STATELESS_CONSUMER, Boolean.TRUE);
+ }
+ listener.configureTarget(lme, getContext());
+ MessageUtil.transferToIn(msg, lme);
+ send(lme);
+ MessageUtil.transferTo(msg, dest, message);
+ send(dest);
+ } else {
+ MessageUtil.transferTo(source, dest, message);
+ send(dest);
+ }
+ }
+
+ private void sendSyncToListenerAndTarget(MessageExchange source,
+ MessageExchange dest,
+ ExchangeTarget listener,
+ String message) throws Exception {
+ if (listener != null) {
+ NormalizedMessage msg = MessageUtil.copy(source.getMessage(message));
+ InOnly lme = exchangeFactory.createInOnlyExchange();
+ if (store.hasFeature(Store.CLUSTERED)) {
+ lme.setProperty(JbiConstants.STATELESS_CONSUMER, Boolean.TRUE);
+ }
+ listener.configureTarget(lme, getContext());
+ MessageUtil.transferToIn(msg, lme);
+ sendSync(lme);
+ MessageUtil.transferTo(msg, dest, message);
+ sendSync(dest);
+ } else {
+ MessageUtil.transferTo(source, dest, message);
+ sendSync(dest);
+ }
+ }
+
+}
Propchange: incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/patterns/WireTap.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/patterns/XPathSplitter.java
URL: http://svn.apache.org/viewvc/incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/patterns/XPathSplitter.java?rev=426392&r1=426391&r2=426392&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/patterns/XPathSplitter.java (original)
+++ incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/patterns/XPathSplitter.java Thu Jul 27 22:49:32 2006
@@ -1,159 +1,159 @@
-/*
- * 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.xml.namespace.NamespaceContext;
-import javax.xml.transform.Source;
-import javax.xml.transform.dom.DOMSource;
-import javax.xml.xpath.XPathFactory;
-import javax.xml.xpath.XPathFunctionResolver;
-
-import org.apache.servicemix.eip.support.AbstractSplitter;
-import org.apache.servicemix.expression.JAXPNodeSetXPathExpression;
-import org.apache.servicemix.expression.MessageVariableResolver;
-import org.apache.servicemix.jbi.jaxp.SourceTransformer;
-import org.w3c.dom.Node;
-import org.w3c.dom.NodeList;
-
-/**
- * The XPathSplitter component implements the
- * <a href="http://www.enterpriseintegrationpatterns.com/Sequencer.html">Splitter</a>
- * pattern using an xpath expression to split the incoming xml.
- *
- * @author gnodet
- * @version $Revision: 376451 $
- * @org.apache.xbean.XBean element="xpath-splitter"
- * description="An XPath Splitter"
- */
-public class XPathSplitter extends AbstractSplitter {
-
- /**
- * The xpath expression to use to split
- */
- private JAXPNodeSetXPathExpression xpathExpression = new JAXPNodeSetXPathExpression();
-
- /* (non-Javadoc)
- * @see org.apache.servicemix.eip.EIPEndpoint#validate()
- */
- public void validate() throws DeploymentException {
- super.validate();
- // Check xpath expression
- try {
- xpathExpression.afterPropertiesSet();
- } catch (Exception e) {
- throw new DeploymentException("Error validating xpath expression", e);
- }
- }
-
- /* (non-Javadoc)
- * @see org.apache.servicemix.components.eip.AbstractSplitter#split(javax.xml.transform.Source)
- */
- protected Source[] split(Source main) throws Exception {
- Node doc = new SourceTransformer().toDOMNode(main);
- NodeList nodes = (NodeList) xpathExpression.evaluateXPath(doc);
- Source[] parts = new Source[nodes.getLength()];
- for (int i = 0; i < parts.length; i++) {
- parts[i] = new DOMSource(nodes.item(i));
- }
- return parts;
- }
-
- /* (non-Javadoc)
- * @see org.apache.servicemix.expression.JAXPXPathExpression#getFactory()
- */
- public XPathFactory getFactory() {
- return xpathExpression.getFactory();
- }
-
- /* (non-Javadoc)
- * @see org.apache.servicemix.expression.JAXPXPathExpression#getFunctionResolver()
- */
- public XPathFunctionResolver getFunctionResolver() {
- return xpathExpression.getFunctionResolver();
- }
-
- /* (non-Javadoc)
- * @see org.apache.servicemix.expression.JAXPXPathExpression#getNamespaceContext()
- */
- public NamespaceContext getNamespaceContext() {
- return xpathExpression.getNamespaceContext();
- }
-
- /* (non-Javadoc)
- * @see org.apache.servicemix.expression.JAXPXPathExpression#getTransformer()
- */
- public SourceTransformer getTransformer() {
- return xpathExpression.getTransformer();
- }
-
- /* (non-Javadoc)
- * @see org.apache.servicemix.expression.JAXPXPathExpression#getVariableResolver()
- */
- public MessageVariableResolver getVariableResolver() {
- return xpathExpression.getVariableResolver();
- }
-
- /* (non-Javadoc)
- * @see org.apache.servicemix.expression.JAXPXPathExpression#getXPath()
- */
- public String getXPath() {
- return xpathExpression.getXPath();
- }
-
- /* (non-Javadoc)
- * @see org.apache.servicemix.expression.JAXPXPathExpression#setFactory(javax.xml.xpath.XPathFactory)
- */
- public void setFactory(XPathFactory factory) {
- xpathExpression.setFactory(factory);
- }
-
- /* (non-Javadoc)
- * @see org.apache.servicemix.expression.JAXPXPathExpression#setFunctionResolver(javax.xml.xpath.XPathFunctionResolver)
- */
- public void setFunctionResolver(XPathFunctionResolver functionResolver) {
- xpathExpression.setFunctionResolver(functionResolver);
- }
-
- /* (non-Javadoc)
- * @see org.apache.servicemix.expression.JAXPXPathExpression#setNamespaceContext(javax.xml.namespace.NamespaceContext)
- */
- public void setNamespaceContext(NamespaceContext namespaceContext) {
- xpathExpression.setNamespaceContext(namespaceContext);
- }
-
- /* (non-Javadoc)
- * @see org.apache.servicemix.expression.JAXPXPathExpression#setTransformer(org.apache.servicemix.jbi.jaxp.SourceTransformer)
- */
- public void setTransformer(SourceTransformer transformer) {
- xpathExpression.setTransformer(transformer);
- }
-
- /* (non-Javadoc)
- * @see org.apache.servicemix.expression.JAXPXPathExpression#setVariableResolver(org.apache.servicemix.expression.MessageVariableResolver)
- */
- public void setVariableResolver(MessageVariableResolver variableResolver) {
- xpathExpression.setVariableResolver(variableResolver);
- }
-
- /**
- * @org.apache.xbean.Property alias="xpath"
- */
- public void setXPath(String xpath) {
- xpathExpression.setXPath(xpath);
- }
-
-}
+/*
+ * 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.xml.namespace.NamespaceContext;
+import javax.xml.transform.Source;
+import javax.xml.transform.dom.DOMSource;
+import javax.xml.xpath.XPathFactory;
+import javax.xml.xpath.XPathFunctionResolver;
+
+import org.apache.servicemix.eip.support.AbstractSplitter;
+import org.apache.servicemix.expression.JAXPNodeSetXPathExpression;
+import org.apache.servicemix.expression.MessageVariableResolver;
+import org.apache.servicemix.jbi.jaxp.SourceTransformer;
+import org.w3c.dom.Node;
+import org.w3c.dom.NodeList;
+
+/**
+ * The XPathSplitter component implements the
+ * <a href="http://www.enterpriseintegrationpatterns.com/Sequencer.html">Splitter</a>
+ * pattern using an xpath expression to split the incoming xml.
+ *
+ * @author gnodet
+ * @version $Revision: 376451 $
+ * @org.apache.xbean.XBean element="xpath-splitter"
+ * description="An XPath Splitter"
+ */
+public class XPathSplitter extends AbstractSplitter {
+
+ /**
+ * The xpath expression to use to split
+ */
+ private JAXPNodeSetXPathExpression xpathExpression = new JAXPNodeSetXPathExpression();
+
+ /* (non-Javadoc)
+ * @see org.apache.servicemix.eip.EIPEndpoint#validate()
+ */
+ public void validate() throws DeploymentException {
+ super.validate();
+ // Check xpath expression
+ try {
+ xpathExpression.afterPropertiesSet();
+ } catch (Exception e) {
+ throw new DeploymentException("Error validating xpath expression", e);
+ }
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.servicemix.components.eip.AbstractSplitter#split(javax.xml.transform.Source)
+ */
+ protected Source[] split(Source main) throws Exception {
+ Node doc = new SourceTransformer().toDOMNode(main);
+ NodeList nodes = (NodeList) xpathExpression.evaluateXPath(doc);
+ Source[] parts = new Source[nodes.getLength()];
+ for (int i = 0; i < parts.length; i++) {
+ parts[i] = new DOMSource(nodes.item(i));
+ }
+ return parts;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.servicemix.expression.JAXPXPathExpression#getFactory()
+ */
+ public XPathFactory getFactory() {
+ return xpathExpression.getFactory();
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.servicemix.expression.JAXPXPathExpression#getFunctionResolver()
+ */
+ public XPathFunctionResolver getFunctionResolver() {
+ return xpathExpression.getFunctionResolver();
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.servicemix.expression.JAXPXPathExpression#getNamespaceContext()
+ */
+ public NamespaceContext getNamespaceContext() {
+ return xpathExpression.getNamespaceContext();
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.servicemix.expression.JAXPXPathExpression#getTransformer()
+ */
+ public SourceTransformer getTransformer() {
+ return xpathExpression.getTransformer();
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.servicemix.expression.JAXPXPathExpression#getVariableResolver()
+ */
+ public MessageVariableResolver getVariableResolver() {
+ return xpathExpression.getVariableResolver();
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.servicemix.expression.JAXPXPathExpression#getXPath()
+ */
+ public String getXPath() {
+ return xpathExpression.getXPath();
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.servicemix.expression.JAXPXPathExpression#setFactory(javax.xml.xpath.XPathFactory)
+ */
+ public void setFactory(XPathFactory factory) {
+ xpathExpression.setFactory(factory);
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.servicemix.expression.JAXPXPathExpression#setFunctionResolver(javax.xml.xpath.XPathFunctionResolver)
+ */
+ public void setFunctionResolver(XPathFunctionResolver functionResolver) {
+ xpathExpression.setFunctionResolver(functionResolver);
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.servicemix.expression.JAXPXPathExpression#setNamespaceContext(javax.xml.namespace.NamespaceContext)
+ */
+ public void setNamespaceContext(NamespaceContext namespaceContext) {
+ xpathExpression.setNamespaceContext(namespaceContext);
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.servicemix.expression.JAXPXPathExpression#setTransformer(org.apache.servicemix.jbi.jaxp.SourceTransformer)
+ */
+ public void setTransformer(SourceTransformer transformer) {
+ xpathExpression.setTransformer(transformer);
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.servicemix.expression.JAXPXPathExpression#setVariableResolver(org.apache.servicemix.expression.MessageVariableResolver)
+ */
+ public void setVariableResolver(MessageVariableResolver variableResolver) {
+ xpathExpression.setVariableResolver(variableResolver);
+ }
+
+ /**
+ * @org.apache.xbean.Property alias="xpath"
+ */
+ public void setXPath(String xpath) {
+ xpathExpression.setXPath(xpath);
+ }
+
+}
Propchange: incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/patterns/XPathSplitter.java
------------------------------------------------------------------------------
svn:eol-style = native
Modified: incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/support/AbstractAggregator.java
URL: http://svn.apache.org/viewvc/incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/support/AbstractAggregator.java?rev=426392&r1=426391&r2=426392&view=diff
==============================================================================
--- incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/support/AbstractAggregator.java (original)
+++ incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/support/AbstractAggregator.java Thu Jul 27 22:49:32 2006
@@ -1,294 +1,294 @@
-/*
- * 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.support;
-
-import java.util.Date;
-
-import javax.jbi.messaging.ExchangeStatus;
-import javax.jbi.messaging.InOnly;
-import javax.jbi.messaging.MessageExchange;
-import javax.jbi.messaging.NormalizedMessage;
-import javax.jbi.messaging.RobustInOnly;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.servicemix.eip.EIPEndpoint;
-import org.apache.servicemix.timers.Timer;
-import org.apache.servicemix.timers.TimerListener;
-
-import edu.emory.mathcs.backport.java.util.concurrent.ConcurrentHashMap;
-import edu.emory.mathcs.backport.java.util.concurrent.ConcurrentMap;
-import edu.emory.mathcs.backport.java.util.concurrent.locks.Lock;
-
-/**
- * Aggregator can be used to wait and combine several messages.
- * This component implements the
- * <a href="http://www.enterpriseintegrationpatterns.com/Aggregator.html">Aggregator</a>
- * pattern.
- *
- * TODO: keep list of closed aggregations for a certain time
- * TODO: distributed lock manager
- * TODO: persistent / transactional timer
- *
- * @author gnodet
- * @version $Revision: 376451 $
- */
-public abstract class AbstractAggregator extends EIPEndpoint {
-
- private static final Log log = LogFactory.getLog(AbstractAggregator.class);
-
- private ExchangeTarget target;
-
- private boolean rescheduleTimeouts;
-
- private boolean synchronous;
-
- private ConcurrentMap closedAggregates = new ConcurrentHashMap();
-
- /**
- * @return the synchronous
- */
- public boolean isSynchronous() {
- return synchronous;
- }
-
- /**
- * @param synchronous the synchronous to set
- */
- public void setSynchronous(boolean synchronous) {
- this.synchronous = synchronous;
- }
-
- /**
- * @return the rescheduleTimeouts
- */
- public boolean isRescheduleTimeouts() {
- return rescheduleTimeouts;
- }
-
- /**
- * @param rescheduleTimeouts the rescheduleTimeouts to set
- */
- public void setRescheduleTimeouts(boolean rescheduleTimeouts) {
- this.rescheduleTimeouts = rescheduleTimeouts;
- }
-
- /**
- * @return the target
- */
- public ExchangeTarget getTarget() {
- return target;
- }
-
- /**
- * @param target the target to set
- */
- public void setTarget(ExchangeTarget target) {
- this.target = target;
- }
-
- /* (non-Javadoc)
- * @see org.apache.servicemix.eip.EIPEndpoint#processSync(javax.jbi.messaging.MessageExchange)
- */
- protected void processSync(MessageExchange exchange) throws Exception {
- throw new IllegalStateException();
- }
-
- /* (non-Javadoc)
- * @see org.apache.servicemix.eip.EIPEndpoint#processAsync(javax.jbi.messaging.MessageExchange)
- */
- protected void processAsync(MessageExchange exchange) throws Exception {
- throw new IllegalStateException();
- }
-
- /* (non-Javadoc)
- * @see org.apache.servicemix.common.ExchangeProcessor#process(javax.jbi.messaging.MessageExchange)
- */
- public void process(MessageExchange exchange) throws Exception {
- // Skip DONE
- if (exchange.getStatus() == ExchangeStatus.DONE) {
- return;
- // Skip ERROR
- } else if (exchange.getStatus() == ExchangeStatus.ERROR) {
- return;
- // Handle an ACTIVE exchange as a PROVIDER
- } else if (exchange.getRole() == MessageExchange.Role.PROVIDER) {
- 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);
- final String correlationId = getCorrelationID(exchange, in);
- if (correlationId == null || correlationId.length() == 0) {
- throw new IllegalArgumentException("Could not retrieve correlation id for incoming exchange");
- }
- // Load existing aggregation
- Lock lock = getLockManager().getLock(correlationId);
- lock.lock();
- try {
- Object aggregation = store.load(correlationId);
- Date timeout = null;
- // Create a new aggregate
- if (aggregation == null) {
- if (isAggregationClosed(correlationId)) {
- // TODO: should we return an error here ?
- } else {
- aggregation = createAggregation(correlationId);
- timeout = getTimeout(aggregation);
- }
- } else if (isRescheduleTimeouts()) {
- timeout = getTimeout(aggregation);
- }
- // If the aggregation is not closed
- if (aggregation != null) {
- if (addMessage(aggregation, in, exchange)) {
- sendAggregate(correlationId, aggregation, false);
- } else {
- store.store(correlationId, aggregation);
- if (timeout != null) {
- if (log.isDebugEnabled()) {
- log.debug("Scheduling timeout at " + timeout + " for aggregate " + correlationId);
- }
- getTimerManager().schedule(new TimerListener() {
- public void timerExpired(Timer timer) {
- AbstractAggregator.this.onTimeout(correlationId);
- }
- }, timeout);
- }
- }
- }
- done(exchange);
- } finally {
- lock.unlock();
- }
- }
- // Handle an ACTIVE exchange as a CONSUMER
- } else if (exchange.getStatus() == ExchangeStatus.ACTIVE) {
- done(exchange);
- }
- }
-
- protected void sendAggregate(String correlationId,
- Object aggregation,
- boolean timeout) throws Exception {
- InOnly me = exchangeFactory.createInOnlyExchange();
- target.configureTarget(me, getContext());
- NormalizedMessage nm = me.createMessage();
- me.setInMessage(nm);
- buildAggregate(aggregation, nm, me, timeout);
- closeAggregation(correlationId);
- if (isSynchronous()) {
- sendSync(me);
- } else {
- send(me);
- }
- }
-
- protected void onTimeout(String correlationId) {
- if (log.isDebugEnabled()) {
- log.debug("Timeout expired for aggregate " + correlationId);
- }
- Lock lock = getLockManager().getLock(correlationId);
- lock.lock();
- try {
- Object aggregation = store.load(correlationId);
- if (aggregation != null) {
- sendAggregate(correlationId, aggregation, true);
- } else if (!isAggregationClosed(correlationId)) {
- throw new IllegalStateException("Aggregation is not closed, but can not be retrieved from the store");
- } else {
- if (log.isDebugEnabled()) {
- log.debug("Aggregate " + correlationId + " is closed");
- }
- }
- } catch (Exception e) {
- log.info("Caught exception while processing timeout aggregation", e);
- } finally {
- lock.unlock();
- }
- }
-
- /**
- * Check if the aggregation with the given correlation id is closed or not.
- * Called when the aggregation has not been found in the store.
- *
- * @param correlationId
- * @return
- */
- protected boolean isAggregationClosed(String correlationId) {
- // TODO: implement this using a persistent / cached behavior
- return closedAggregates.containsKey(correlationId);
- }
-
- /**
- * Mark an aggregation as closed
- * @param correlationId
- */
- protected void closeAggregation(String correlationId) {
- // TODO: implement this using a persistent / cached behavior
- closedAggregates.put(correlationId, Boolean.TRUE);
- }
-
- /**
- * Retrieve the correlation ID of the given exchange
- * @param exchange
- * @param message
- * @return the correlationID
- * @throws Exception
- */
- protected abstract String getCorrelationID(MessageExchange exchange, NormalizedMessage message) throws Exception;
-
- /**
- * Creates a new empty aggregation.
- * @param correlationID
- * @return a newly created aggregation
- */
- protected abstract Object createAggregation(String correlationID) throws Exception;
-
- /**
- * Returns the date when the onTimeout method should be called if the aggregation is not completed yet,
- * or null if the aggregation has no timeout.
- *
- * @param aggregate
- * @return
- */
- protected abstract Date getTimeout(Object aggregate);
-
- /**
- * Add a newly received message to this aggregation
- *
- * @param aggregate
- * @param message
- * @param exchange
- * @return <code>true</code> if the aggregate id complete
- */
- protected abstract boolean addMessage(Object aggregate,
- NormalizedMessage message,
- MessageExchange exchange) throws Exception;
-
- /**
- * Fill the given JBI message with the aggregation result.
- *
- * @param aggregate
- * @param message
- * @param exchange
- * @param timeout <code>false</code> if the aggregation has completed or <code>true</code> if this aggregation has timed out
- */
- protected abstract void buildAggregate(Object aggregate,
- NormalizedMessage message,
- MessageExchange exchange,
- boolean timeout) throws Exception;
-}
+/*
+ * 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.support;
+
+import java.util.Date;
+
+import javax.jbi.messaging.ExchangeStatus;
+import javax.jbi.messaging.InOnly;
+import javax.jbi.messaging.MessageExchange;
+import javax.jbi.messaging.NormalizedMessage;
+import javax.jbi.messaging.RobustInOnly;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.servicemix.eip.EIPEndpoint;
+import org.apache.servicemix.timers.Timer;
+import org.apache.servicemix.timers.TimerListener;
+
+import edu.emory.mathcs.backport.java.util.concurrent.ConcurrentHashMap;
+import edu.emory.mathcs.backport.java.util.concurrent.ConcurrentMap;
+import edu.emory.mathcs.backport.java.util.concurrent.locks.Lock;
+
+/**
+ * Aggregator can be used to wait and combine several messages.
+ * This component implements the
+ * <a href="http://www.enterpriseintegrationpatterns.com/Aggregator.html">Aggregator</a>
+ * pattern.
+ *
+ * TODO: keep list of closed aggregations for a certain time
+ * TODO: distributed lock manager
+ * TODO: persistent / transactional timer
+ *
+ * @author gnodet
+ * @version $Revision: 376451 $
+ */
+public abstract class AbstractAggregator extends EIPEndpoint {
+
+ private static final Log log = LogFactory.getLog(AbstractAggregator.class);
+
+ private ExchangeTarget target;
+
+ private boolean rescheduleTimeouts;
+
+ private boolean synchronous;
+
+ private ConcurrentMap closedAggregates = new ConcurrentHashMap();
+
+ /**
+ * @return the synchronous
+ */
+ public boolean isSynchronous() {
+ return synchronous;
+ }
+
+ /**
+ * @param synchronous the synchronous to set
+ */
+ public void setSynchronous(boolean synchronous) {
+ this.synchronous = synchronous;
+ }
+
+ /**
+ * @return the rescheduleTimeouts
+ */
+ public boolean isRescheduleTimeouts() {
+ return rescheduleTimeouts;
+ }
+
+ /**
+ * @param rescheduleTimeouts the rescheduleTimeouts to set
+ */
+ public void setRescheduleTimeouts(boolean rescheduleTimeouts) {
+ this.rescheduleTimeouts = rescheduleTimeouts;
+ }
+
+ /**
+ * @return the target
+ */
+ public ExchangeTarget getTarget() {
+ return target;
+ }
+
+ /**
+ * @param target the target to set
+ */
+ public void setTarget(ExchangeTarget target) {
+ this.target = target;
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.servicemix.eip.EIPEndpoint#processSync(javax.jbi.messaging.MessageExchange)
+ */
+ protected void processSync(MessageExchange exchange) throws Exception {
+ throw new IllegalStateException();
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.servicemix.eip.EIPEndpoint#processAsync(javax.jbi.messaging.MessageExchange)
+ */
+ protected void processAsync(MessageExchange exchange) throws Exception {
+ throw new IllegalStateException();
+ }
+
+ /* (non-Javadoc)
+ * @see org.apache.servicemix.common.ExchangeProcessor#process(javax.jbi.messaging.MessageExchange)
+ */
+ public void process(MessageExchange exchange) throws Exception {
+ // Skip DONE
+ if (exchange.getStatus() == ExchangeStatus.DONE) {
+ return;
+ // Skip ERROR
+ } else if (exchange.getStatus() == ExchangeStatus.ERROR) {
+ return;
+ // Handle an ACTIVE exchange as a PROVIDER
+ } else if (exchange.getRole() == MessageExchange.Role.PROVIDER) {
+ 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);
+ final String correlationId = getCorrelationID(exchange, in);
+ if (correlationId == null || correlationId.length() == 0) {
+ throw new IllegalArgumentException("Could not retrieve correlation id for incoming exchange");
+ }
+ // Load existing aggregation
+ Lock lock = getLockManager().getLock(correlationId);
+ lock.lock();
+ try {
+ Object aggregation = store.load(correlationId);
+ Date timeout = null;
+ // Create a new aggregate
+ if (aggregation == null) {
+ if (isAggregationClosed(correlationId)) {
+ // TODO: should we return an error here ?
+ } else {
+ aggregation = createAggregation(correlationId);
+ timeout = getTimeout(aggregation);
+ }
+ } else if (isRescheduleTimeouts()) {
+ timeout = getTimeout(aggregation);
+ }
+ // If the aggregation is not closed
+ if (aggregation != null) {
+ if (addMessage(aggregation, in, exchange)) {
+ sendAggregate(correlationId, aggregation, false);
+ } else {
+ store.store(correlationId, aggregation);
+ if (timeout != null) {
+ if (log.isDebugEnabled()) {
+ log.debug("Scheduling timeout at " + timeout + " for aggregate " + correlationId);
+ }
+ getTimerManager().schedule(new TimerListener() {
+ public void timerExpired(Timer timer) {
+ AbstractAggregator.this.onTimeout(correlationId);
+ }
+ }, timeout);
+ }
+ }
+ }
+ done(exchange);
+ } finally {
+ lock.unlock();
+ }
+ }
+ // Handle an ACTIVE exchange as a CONSUMER
+ } else if (exchange.getStatus() == ExchangeStatus.ACTIVE) {
+ done(exchange);
+ }
+ }
+
+ protected void sendAggregate(String correlationId,
+ Object aggregation,
+ boolean timeout) throws Exception {
+ InOnly me = exchangeFactory.createInOnlyExchange();
+ target.configureTarget(me, getContext());
+ NormalizedMessage nm = me.createMessage();
+ me.setInMessage(nm);
+ buildAggregate(aggregation, nm, me, timeout);
+ closeAggregation(correlationId);
+ if (isSynchronous()) {
+ sendSync(me);
+ } else {
+ send(me);
+ }
+ }
+
+ protected void onTimeout(String correlationId) {
+ if (log.isDebugEnabled()) {
+ log.debug("Timeout expired for aggregate " + correlationId);
+ }
+ Lock lock = getLockManager().getLock(correlationId);
+ lock.lock();
+ try {
+ Object aggregation = store.load(correlationId);
+ if (aggregation != null) {
+ sendAggregate(correlationId, aggregation, true);
+ } else if (!isAggregationClosed(correlationId)) {
+ throw new IllegalStateException("Aggregation is not closed, but can not be retrieved from the store");
+ } else {
+ if (log.isDebugEnabled()) {
+ log.debug("Aggregate " + correlationId + " is closed");
+ }
+ }
+ } catch (Exception e) {
+ log.info("Caught exception while processing timeout aggregation", e);
+ } finally {
+ lock.unlock();
+ }
+ }
+
+ /**
+ * Check if the aggregation with the given correlation id is closed or not.
+ * Called when the aggregation has not been found in the store.
+ *
+ * @param correlationId
+ * @return
+ */
+ protected boolean isAggregationClosed(String correlationId) {
+ // TODO: implement this using a persistent / cached behavior
+ return closedAggregates.containsKey(correlationId);
+ }
+
+ /**
+ * Mark an aggregation as closed
+ * @param correlationId
+ */
+ protected void closeAggregation(String correlationId) {
+ // TODO: implement this using a persistent / cached behavior
+ closedAggregates.put(correlationId, Boolean.TRUE);
+ }
+
+ /**
+ * Retrieve the correlation ID of the given exchange
+ * @param exchange
+ * @param message
+ * @return the correlationID
+ * @throws Exception
+ */
+ protected abstract String getCorrelationID(MessageExchange exchange, NormalizedMessage message) throws Exception;
+
+ /**
+ * Creates a new empty aggregation.
+ * @param correlationID
+ * @return a newly created aggregation
+ */
+ protected abstract Object createAggregation(String correlationID) throws Exception;
+
+ /**
+ * Returns the date when the onTimeout method should be called if the aggregation is not completed yet,
+ * or null if the aggregation has no timeout.
+ *
+ * @param aggregate
+ * @return
+ */
+ protected abstract Date getTimeout(Object aggregate);
+
+ /**
+ * Add a newly received message to this aggregation
+ *
+ * @param aggregate
+ * @param message
+ * @param exchange
+ * @return <code>true</code> if the aggregate id complete
+ */
+ protected abstract boolean addMessage(Object aggregate,
+ NormalizedMessage message,
+ MessageExchange exchange) throws Exception;
+
+ /**
+ * Fill the given JBI message with the aggregation result.
+ *
+ * @param aggregate
+ * @param message
+ * @param exchange
+ * @param timeout <code>false</code> if the aggregation has completed or <code>true</code> if this aggregation has timed out
+ */
+ protected abstract void buildAggregate(Object aggregate,
+ NormalizedMessage message,
+ MessageExchange exchange,
+ boolean timeout) throws Exception;
+}
Propchange: incubator/servicemix/trunk/servicemix-eip/src/main/java/org/apache/servicemix/eip/support/AbstractAggregator.java
------------------------------------------------------------------------------
svn:eol-style = native