diff --git a/pom.xml b/pom.xml
index a98557da76..1538fd1c3d 100644
--- a/pom.xml
+++ b/pom.xml
@@ -29,6 +29,7 @@
spring-integration-ftp
spring-integration-sftp
spring-integration-twitter
+ spring-integration-activiti
scm:svn:https://src.springframework.org/svn/spring-integration/trunk
diff --git a/spring-integration-activiti/pom.xml b/spring-integration-activiti/pom.xml
new file mode 100644
index 0000000000..61b22c10a5
--- /dev/null
+++ b/spring-integration-activiti/pom.xml
@@ -0,0 +1,102 @@
+
+
+ 4.0.0
+
+ org.springframework.integration
+ spring-integration-parent
+ 2.0.0.BUILD-SNAPSHOT
+
+ org.springframework.integration
+ spring-integration-activiti
+ jar
+ Spring Integration Activiti Support
+
+
+
+ commons-dbcp
+ commons-dbcp
+ 1.2.2
+
+
+ cglib
+ cglib-nodep
+ 2.2
+
+
+ commons-lang
+ commons-lang
+ 2.5
+
+
+ org.springframework
+ spring-core
+ 3.0.3.RELEASE
+
+
+ org.springframework
+ spring-orm
+ 3.0.3.RELEASE
+
+
+ org.springframework
+ spring-beans
+ 3.0.3.RELEASE
+
+
+ org.springframework
+ spring-test
+ 3.0.3.RELEASE
+
+
+ org.springframework.integration
+ spring-integration-event
+ ${project.version}
+
+
+ org.springframework.integration
+ spring-integration-jms
+ ${project.version}
+
+
+
+ org.springframework.integration
+ spring-integration-core
+ ${project.version}
+
+
+ org.springframework
+ spring-context
+ 3.0.3.RELEASE
+
+
+ org.springframework
+ spring-context-support
+ 3.0.3.RELEASE
+
+
+ org.springframework
+ spring-aop
+ 3.0.3.RELEASE
+
+
+ com.h2database
+ h2
+ 1.2.132
+
+
+ org.activiti
+ activiti-engine
+ 5.0.beta1
+
+
+
+
+
+
diff --git a/spring-integration-activiti/src/main/java/org/springframework/integration/activiti/ActivitiConstants.java b/spring-integration-activiti/src/main/java/org/springframework/integration/activiti/ActivitiConstants.java
new file mode 100644
index 0000000000..fca24687b4
--- /dev/null
+++ b/spring-integration-activiti/src/main/java/org/springframework/integration/activiti/ActivitiConstants.java
@@ -0,0 +1,35 @@
+/* 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.springframework.integration.activiti;
+
+/**
+ * @author Josh Long
+ */
+public class ActivitiConstants {
+
+ /**
+ * In order for the gateway to correctly signal execution to Activiti, it needs to the executionId so that it can look up the {@link org.activiti.engine.runtime.Execution} instance.
+ * the executionId is expected to be under this header.
+ */
+ public static final String WELL_KNOWN_EXECUTION_ID_HEADER_KEY = "activiti_spring_integration_executionId";
+
+ /**
+ * Assuming #updateProcessVariablesFromReplyMessageHeaders is true, then any {@link org.springframework.integration.MessageHeaders} header key that starts with String will be propagated as an Activiti process variable.
+ */
+ public static final String WELL_KNOWN_SPRING_INTEGRATION_HEADER_PREFIX = "activiti_spring_integration_";
+
+ /**
+ * This is the key under which we will look up the custom processDefinitionName up. This value will be used to spawn
+ */
+ public static final String WELL_KNOWN_PROCESS_DEFINITION_NAME_HEADER_KEY = "activiti_spring_integration_processDefinitionName";
+}
diff --git a/spring-integration-activiti/src/main/java/org/springframework/integration/activiti/ProcessSupport.java b/spring-integration-activiti/src/main/java/org/springframework/integration/activiti/ProcessSupport.java
new file mode 100644
index 0000000000..d0ab59de42
--- /dev/null
+++ b/spring-integration-activiti/src/main/java/org/springframework/integration/activiti/ProcessSupport.java
@@ -0,0 +1,55 @@
+/* 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.springframework.integration.activiti;
+
+import org.springframework.integration.MessageHeaders;
+
+import java.util.*;
+import java.util.concurrent.ConcurrentSkipListSet;
+
+
+/**
+ * Provides utility logic to take headers from a {@link org.springframework.integration.MessageHeaders} instance and propagate them as
+ * values that can be used as process variables for a {@link org.activiti.engine.runtime.ProcessInstance}.
+ *
+ * @author Josh Long
+ */
+public class ProcessSupport {
+ private final int headerPrefixLength = ActivitiConstants.WELL_KNOWN_SPRING_INTEGRATION_HEADER_PREFIX.length();
+ private Collection blackList;
+
+ public ProcessSupport() {
+ this.blackList = new ConcurrentSkipListSet(Arrays.asList(ActivitiConstants.WELL_KNOWN_EXECUTION_ID_HEADER_KEY, ActivitiConstants.WELL_KNOWN_PROCESS_DEFINITION_NAME_HEADER_KEY));
+ }
+
+ public Map processVariablesFromMessageHeaders(MessageHeaders msg) {
+ return this.processVariablesFromMessageHeaders(new HashSet(), msg);
+ }
+
+ public Map processVariablesFromMessageHeaders(Set whiteListOfMustCopyMessageHeaderKeyNames, MessageHeaders msgHeaders) {
+ Map procVars = new HashMap();
+
+ Set headers = msgHeaders.keySet();
+ Set wl = (whiteListOfMustCopyMessageHeaderKeyNames == null) ? new HashSet() : whiteListOfMustCopyMessageHeaderKeyNames;
+
+ for (String messageHeaderKey : headers) {
+ if ((!blackList.contains(messageHeaderKey) && messageHeaderKey.startsWith(ActivitiConstants.WELL_KNOWN_SPRING_INTEGRATION_HEADER_PREFIX)) || wl.contains(messageHeaderKey)) {
+ String pvName = messageHeaderKey.startsWith(ActivitiConstants.WELL_KNOWN_SPRING_INTEGRATION_HEADER_PREFIX) ? messageHeaderKey.substring(headerPrefixLength) : messageHeaderKey;
+ procVars.put(pvName, msgHeaders.get(messageHeaderKey));
+ }
+ }
+
+ return procVars;
+ }
+}
diff --git a/spring-integration-activiti/src/main/java/org/springframework/integration/activiti/adapter/ActivitiOutboundChannelAdapter.java b/spring-integration-activiti/src/main/java/org/springframework/integration/activiti/adapter/ActivitiOutboundChannelAdapter.java
new file mode 100644
index 0000000000..4c50100bc3
--- /dev/null
+++ b/spring-integration-activiti/src/main/java/org/springframework/integration/activiti/adapter/ActivitiOutboundChannelAdapter.java
@@ -0,0 +1,86 @@
+/* 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.springframework.integration.activiti.adapter;
+
+import org.activiti.engine.ProcessEngine;
+import org.springframework.integration.Message;
+import org.springframework.integration.activiti.ActivitiConstants;
+import org.springframework.integration.activiti.ProcessSupport;
+import org.springframework.integration.context.IntegrationObjectSupport;
+import org.springframework.integration.core.MessageHandler;
+import org.springframework.util.Assert;
+
+import java.util.Map;
+
+
+/**
+ * Supports spawning a {@link org.activiti.engine.runtime.ProcessInstance} as a result of a trigger {@link org.springframework.integration.Message}.
+ *
+ * The component also supports propagating headers as processVariables.
+ *
+ * This support is simialar to the classic EIP book's "Process Manager" pattern.
+ *
+ * Thanks to Mark Fisherfor the idea.
+ *
+ * @author Josh Long
+ */
+public class ActivitiOutboundChannelAdapter extends IntegrationObjectSupport implements MessageHandler {
+ /**
+ * A reference to the {@link ProcessEngine} (see {@link org.activiti.engine.impl.cfg.spring.ProcessEngineFactoryBean}
+ */
+ private ProcessEngine processEngine;
+
+ /**
+ * Do you want all flows that come into this component to launch the same business process? Hard code the process name here.
+ * If this is null, the component will expect a well known header value and use that to spawn the process definition name.
+ */
+ private String processDefinitionName;
+
+ /**
+ * Provides convenience methods
+ */
+ private ProcessSupport processSupport = new ProcessSupport();
+
+ @Override
+ protected void onInit() throws Exception {
+ Assert.notNull(this.processEngine, "'processEngine' is required!");
+ }
+
+ @SuppressWarnings("unused")
+ public void setProcessEngine(ProcessEngine processEngine) {
+ this.processEngine = processEngine;
+ }
+
+ @SuppressWarnings("unused")
+ public void setProcessDefinitionName(String processDefinitionName) {
+ this.processDefinitionName = processDefinitionName;
+ }
+
+ public void handleMessage(Message message) {
+ Map procVars = processSupport.processVariablesFromMessageHeaders(message.getHeaders());
+
+ String procName = (String) message.getHeaders().get(ActivitiConstants.WELL_KNOWN_PROCESS_DEFINITION_NAME_HEADER_KEY);
+
+ if ((procName == null) || procName.trim().equals("")) {
+ procName = this.processDefinitionName;
+ }
+
+ Assert.isTrue(procName != null,
+ "you must specify a processDefinitionName, either through " +
+ "an inbound header mapped to the key " + ActivitiConstants.WELL_KNOWN_PROCESS_DEFINITION_NAME_HEADER_KEY +
+ ", or on the 'process-definition-name' property of this adapter"
+ );
+
+ processEngine.getRuntimeService().startProcessInstanceByKey(this.processDefinitionName, procVars);
+ }
+}
diff --git a/spring-integration-activiti/src/main/java/org/springframework/integration/activiti/config/ActivitiNamespaceHandler.java b/spring-integration-activiti/src/main/java/org/springframework/integration/activiti/config/ActivitiNamespaceHandler.java
new file mode 100644
index 0000000000..4dd987e658
--- /dev/null
+++ b/spring-integration-activiti/src/main/java/org/springframework/integration/activiti/config/ActivitiNamespaceHandler.java
@@ -0,0 +1,74 @@
+/*
+ * Copyright 2010 the original author or authors
+ *
+ * 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.springframework.integration.activiti.config;
+
+import org.springframework.beans.factory.support.AbstractBeanDefinition;
+import org.springframework.beans.factory.support.BeanDefinitionBuilder;
+import org.springframework.beans.factory.xml.AbstractSingleBeanDefinitionParser;
+import org.springframework.beans.factory.xml.NamespaceHandlerSupport;
+import org.springframework.beans.factory.xml.ParserContext;
+import org.springframework.integration.activiti.adapter.ActivitiOutboundChannelAdapter;
+import org.springframework.integration.activiti.gateway.ActivityBehaviorMessagingGatewayFactoryBean;
+import org.springframework.integration.config.xml.AbstractOutboundChannelAdapterParser;
+import org.springframework.integration.config.xml.IntegrationNamespaceUtils;
+import org.w3c.dom.Element;
+
+
+/**
+ * Provides namespace support for defining a Gateway that communicates from Acitviti to Spring Integration.
+ *
+ * @author Josh Long
+ */
+@SuppressWarnings("unused")
+public class ActivitiNamespaceHandler extends NamespaceHandlerSupport {
+
+ public void init() {
+ this.registerBeanDefinitionParser("inbound-gateway", new ActivitiInboundGatewayParser());
+ this.registerBeanDefinitionParser("outbound-channel-adapter", new ActivitiOutboundChannelAdapterParser());
+ }
+
+ private static class ActivitiOutboundChannelAdapterParser extends AbstractOutboundChannelAdapterParser {
+ @Override
+ protected AbstractBeanDefinition parseConsumer(Element element, ParserContext parserContext) {
+ BeanDefinitionBuilder builder = BeanDefinitionBuilder.genericBeanDefinition(ActivitiOutboundChannelAdapter.class.getName());
+ IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "process-engine");
+ IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "process-definition-name");
+ return builder.getBeanDefinition();
+ }
+ }
+
+ private static class ActivitiInboundGatewayParser extends AbstractSingleBeanDefinitionParser {
+ @Override
+ protected String getBeanClassName(Element element) {
+ return ActivityBehaviorMessagingGatewayFactoryBean.class.getName();
+ }
+
+ @Override
+ protected boolean shouldGenerateIdAsFallback() {
+ return true;
+ }
+
+ @Override
+ protected void doParse(Element element, ParserContext parserContext, BeanDefinitionBuilder builder) {
+ IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "request-channel");
+ IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "reply-channel");
+ IntegrationNamespaceUtils.setReferenceIfAttributeDefined(builder, element, "process-engine");
+ IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "update-process-variables-from-reply-message-headers");
+ IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "async");
+ IntegrationNamespaceUtils.setValueIfAttributeDefined(builder, element, "forward-process-variables-as-message-headers");
+ }
+ }
+}
diff --git a/spring-integration-activiti/src/main/java/org/springframework/integration/activiti/gateway/ActivityBehaviorMessagingGatewayFactoryBean.java b/spring-integration-activiti/src/main/java/org/springframework/integration/activiti/gateway/ActivityBehaviorMessagingGatewayFactoryBean.java
new file mode 100644
index 0000000000..1a136374ba
--- /dev/null
+++ b/spring-integration-activiti/src/main/java/org/springframework/integration/activiti/gateway/ActivityBehaviorMessagingGatewayFactoryBean.java
@@ -0,0 +1,109 @@
+package org.springframework.integration.activiti.gateway;
+
+import org.activiti.engine.ProcessEngine;
+
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.BeanFactoryAware;
+import org.springframework.beans.factory.BeanNameAware;
+import org.springframework.beans.factory.FactoryBean;
+
+import org.springframework.integration.MessageChannel;
+
+import org.springframework.transaction.PlatformTransactionManager;
+
+
+public class ActivityBehaviorMessagingGatewayFactoryBean implements FactoryBean, BeanFactoryAware, BeanNameAware {
+ private volatile boolean updateProcessVariablesFromReplyMessageHeaders = false;
+ private volatile boolean forwardProcessVariablesAsMessageHeaders = false;
+ private volatile PlatformTransactionManager platformTransactionManager;
+ private volatile MessageChannel requestChannel;
+ private volatile MessageChannel replyChannel;
+ private volatile ProcessEngine processEngine;
+ private volatile boolean async;
+ private BeanFactory beanFactory;
+ private String beanName;
+
+ @SuppressWarnings("unused")
+ public void setAsync(boolean async) {
+ this.async = async;
+ }
+
+ @SuppressWarnings("unused")
+ public void setProcessEngine(ProcessEngine processEngine) {
+ this.processEngine = processEngine;
+ }
+
+ @SuppressWarnings("unused")
+ public void setUpdateProcessVariablesFromReplyMessageHeaders(boolean updateProcessVariablesFromReplyMessageHeaders) {
+ this.updateProcessVariablesFromReplyMessageHeaders = updateProcessVariablesFromReplyMessageHeaders;
+ }
+
+ @SuppressWarnings("unused")
+ public void setForwardProcessVariablesAsMessageHeaders(boolean forwardProcessVariablesAsMessageHeaders) {
+ this.forwardProcessVariablesAsMessageHeaders = forwardProcessVariablesAsMessageHeaders;
+ }
+
+ @SuppressWarnings("unused")
+ public void setPlatformTransactionManager(PlatformTransactionManager platformTransactionManager) {
+ this.platformTransactionManager = platformTransactionManager;
+ }
+
+ @SuppressWarnings("unused")
+ public void setRequestChannel(MessageChannel requestChannel) {
+ this.requestChannel = requestChannel;
+ }
+
+ @SuppressWarnings("unused")
+ public void setBeanFactory(BeanFactory beanFactory)
+ throws BeansException {
+ this.beanFactory = beanFactory;
+ }
+
+ @SuppressWarnings("unused")
+ public void setReplyChannel(MessageChannel replyChannel) {
+ this.replyChannel = replyChannel;
+ }
+
+ public Object getObject() throws Exception {
+ if (this.async) {
+ AsyncActivityBehaviorMessagingGateway asyncActivityBehaviorMessagingGateway = new AsyncActivityBehaviorMessagingGateway();
+ asyncActivityBehaviorMessagingGateway.setForwardProcessVariablesAsMessageHeaders(this.forwardProcessVariablesAsMessageHeaders);
+ asyncActivityBehaviorMessagingGateway.setUpdateProcessVariablesFromReplyMessageHeaders(this.updateProcessVariablesFromReplyMessageHeaders);
+ asyncActivityBehaviorMessagingGateway.setPlatformTransactionManager(this.platformTransactionManager);
+ asyncActivityBehaviorMessagingGateway.setReplyChannel(this.replyChannel);
+ asyncActivityBehaviorMessagingGateway.setProcessEngine(this.processEngine);
+ asyncActivityBehaviorMessagingGateway.setRequestChannel(this.requestChannel);
+ asyncActivityBehaviorMessagingGateway.setBeanFactory(this.beanFactory);
+ asyncActivityBehaviorMessagingGateway.setBeanName(this.beanName);
+ asyncActivityBehaviorMessagingGateway.afterPropertiesSet();
+
+ return asyncActivityBehaviorMessagingGateway;
+ } else {
+ SyncActivityBehaviorMessagingGateway syncActivityBehaviorMessagingGateway = new SyncActivityBehaviorMessagingGateway();
+ syncActivityBehaviorMessagingGateway.setForwardProcessVariablesAsMessageHeaders(this.forwardProcessVariablesAsMessageHeaders);
+ syncActivityBehaviorMessagingGateway.setUpdateProcessVariablesFromReplyMessageHeaders(this.updateProcessVariablesFromReplyMessageHeaders);
+ syncActivityBehaviorMessagingGateway.setPlatformTransactionManager(this.platformTransactionManager);
+ syncActivityBehaviorMessagingGateway.setReplyChannel(this.replyChannel);
+ syncActivityBehaviorMessagingGateway.setProcessEngine(this.processEngine);
+ syncActivityBehaviorMessagingGateway.setRequestChannel(this.requestChannel);
+ syncActivityBehaviorMessagingGateway.setBeanFactory(this.beanFactory);
+ syncActivityBehaviorMessagingGateway.setBeanName(this.beanName);
+ syncActivityBehaviorMessagingGateway.afterPropertiesSet();
+
+ return syncActivityBehaviorMessagingGateway;
+ }
+ }
+
+ public Class getObjectType() {
+ return this.async ? AsyncActivityBehaviorMessagingGateway.class : SyncActivityBehaviorMessagingGateway.class;
+ }
+
+ public boolean isSingleton() {
+ return true;
+ }
+
+ public void setBeanName(String name) {
+ this.beanName = name;
+ }
+}
diff --git a/spring-integration-activiti/src/main/java/org/springframework/integration/activiti/gateway/AsyncActivityBehaviorMessagingGateway.java b/spring-integration-activiti/src/main/java/org/springframework/integration/activiti/gateway/AsyncActivityBehaviorMessagingGateway.java
new file mode 100644
index 0000000000..c4677d0f17
--- /dev/null
+++ b/spring-integration-activiti/src/main/java/org/springframework/integration/activiti/gateway/AsyncActivityBehaviorMessagingGateway.java
@@ -0,0 +1,263 @@
+/* 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.springframework.integration.activiti.gateway;
+
+
+import org.activiti.engine.ProcessEngine;
+import org.activiti.engine.RuntimeService;
+import org.activiti.engine.impl.bpmn.ReceiveTaskActivity;
+import org.activiti.engine.impl.cfg.spring.ProcessEngineFactoryBean;
+import org.activiti.pvm.activity.ActivityBehavior;
+import org.activiti.pvm.activity.ActivityExecution;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.BeanFactoryAware;
+import org.springframework.beans.factory.BeanNameAware;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.integration.*;
+import org.springframework.integration.activiti.ActivitiConstants;
+import org.springframework.integration.activiti.ProcessSupport;
+import org.springframework.integration.config.ConsumerEndpointFactoryBean;
+import org.springframework.integration.core.*;
+import org.springframework.integration.endpoint.AbstractEndpoint;
+import org.springframework.integration.scheduling.PollerMetadata;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.scheduling.support.PeriodicTrigger;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.util.Assert;
+
+import org.activiti.engine.runtime.Execution ;
+import java.util.Map;
+import java.util.Set;
+
+
+/**
+ * This class is plugged into an Activiti workflow. let's us plugin a custom {@link ActivityBehavior}.
+ * We need to build an {@link ActivityBehavior} that can send and receive the message,
+ * propagating the {@code executionId} and potentially process variables/ header variables.
+ *
+ * Simply svn co http://svn.codehaus.org/activiti/activiti/branches/alpha4-spring-integration-adapter that repository and then mvn clean install it.
+ *
+ * The class forwards ("pushes") control from a BPM process (in-thread) to a Spring Integration channel where, of course, Spring Integration acts as a client and
+ * can push the execution forward anyway it wants to.
+ *
+ * Possible use cases include forwarding the job through an outbound JMS adapter or gateway, forwarding the job through an XMPP adapter, or forwarding the job through
+ * to an outbound email adapter.
+ *
+ *
+ * The only requirement for the reply is that the {@link org.springframework.integration.Message} arrive on the #replyChannel and that it contain a header of
+ * {@link org.springframework.integration.activiti.ActivitiConstants#WELL_KNOWN_EXECUTION_ID_HEADER_KEY} (which the outbound {@link org.springframework.integration.Message} will have)
+ * so that the Activiti runtime can signal that execution has completed successfully.
+ *
+ * Thanks to Dave Syer and Tom Baeyens for the help brainstorming.
+ *
+ * @author Josh Long
+ * @see ReceiveTaskActivity the {@link ActivityBehavior} impl that ships w/ Activiti that has the machinery to wake up when signaled
+ * @see ProcessEngine the process engine instance is required to be able to use this namespace
+ * @see ProcessEngineFactoryBean - use this class to create the aforementioned ProcessEngine instance!
+ */
+public class AsyncActivityBehaviorMessagingGateway extends ReceiveTaskActivity implements BeanFactoryAware, BeanNameAware, ActivityBehavior, InitializingBean {
+ /**
+ * Used to handle sending in a standard way
+ */
+ private MessagingTemplate messagingTemplate = new MessagingTemplate();
+
+ /**
+ * This is the channel on which we expect requests - {@link Execution}s from Activiti - to arrive
+ */
+ private volatile MessageChannel requestChannel;
+
+ /**
+ * This is the channel on which we expect to send replies - ie, the result of our work in
+ * Spring Integration - back to Activiti, which should be waiting for the results
+ */
+ private volatile MessageChannel replyChannel;
+
+ /**
+ * Injected from Spring or some other mechanism. Recommended approach is through a {@link ProcessEngineFactoryBean}
+ */
+ private volatile ProcessEngine processEngine;
+
+ /**
+ * Should we update the process variables based on the reply {@link org.springframework.integration.Message}'s {@link org.springframework.integration.MessageHeaders}?
+ */
+ private volatile boolean updateProcessVariablesFromReplyMessageHeaders = false;
+
+ /**
+ * Should we pass the workflow process variables as message headers when we send a message into the Spring Integration framework?
+ */
+ private volatile boolean forwardProcessVariablesAsMessageHeaders = false;
+
+ /**
+ * Forwarded to the {@link org.springframework.integration.core.MessagingTemplate} instance.
+ */
+ private volatile PlatformTransactionManager platformTransactionManager;
+
+ /**
+ * A reference to the {@link org.springframework.beans.factory.BeanFactory} that's hosting this component. Spring will inject this reference automatically assuming
+ * this object is hosted in a Spring context.
+ */
+ private volatile BeanFactory beanFactory;
+
+ /**
+ * The process engine instance that controls the Activiti PVM. Recommended creation is through {@link ProcessEngineFactoryBean}
+ */
+ private RuntimeService processService;
+
+ /**
+ * Provides common logic for things like sifting through inbound message headers and arriving at process variable candidates
+ */
+ private ProcessSupport processSupport = new ProcessSupport();
+ private String beanName;
+
+ @SuppressWarnings("unused")
+ public void setPlatformTransactionManager(PlatformTransactionManager platformTransactionManager) {
+ this.platformTransactionManager = platformTransactionManager;
+ }
+
+ @SuppressWarnings("unused")
+ public void setRequestChannel(MessageChannel requestChannel) {
+ this.requestChannel = requestChannel;
+ }
+
+ @SuppressWarnings("unused")
+ public void setReplyChannel(MessageChannel replyChannel) {
+ this.replyChannel = replyChannel;
+ }
+
+ @SuppressWarnings("unused")
+ public void setProcessEngine(ProcessEngine processEngine) {
+ this.processEngine = processEngine;
+ }
+
+ @SuppressWarnings("unused")
+ public void setForwardProcessVariablesAsMessageHeaders(boolean forwardProcessVariablesAsMessageHeaders) {
+ this.forwardProcessVariablesAsMessageHeaders = forwardProcessVariablesAsMessageHeaders;
+ }
+
+ @SuppressWarnings("unused")
+ public void setUpdateProcessVariablesFromReplyMessageHeaders(boolean updateProcessVariablesFromReplyMessageHeaders) {
+ this.updateProcessVariablesFromReplyMessageHeaders = updateProcessVariablesFromReplyMessageHeaders;
+ }
+
+ public void setBeanFactory(BeanFactory beanFactory)
+ throws BeansException {
+ this.beanFactory = beanFactory;
+ }
+
+ @Override
+ public void signal(ActivityExecution execution, String signalName, Object data) throws Exception {
+
+ if (data instanceof MessageHeaders) {
+ MessageHeaders messageHeaders = (MessageHeaders) data;
+
+ if (this.updateProcessVariablesFromReplyMessageHeaders) {
+ Map vars = execution .getVariables();
+ Set existingVars = vars.keySet();
+
+ Map procVars = this.processSupport.processVariablesFromMessageHeaders(existingVars, messageHeaders);
+
+ for (String key : procVars.keySet())
+ execution.setVariable(key, procVars.get(key));
+ }
+ }
+
+ super.signal(execution, signalName, data);
+ }
+
+ /**
+ * This is the main interface method from {@link ActivityBehavior}. It will be called when the BPMN process executes the node referencing this logic.
+ *
+ * @param execution the {@link ActivityExecution} as given to use by the engine
+ * @throws Exception
+ */
+ public void execute(ActivityExecution execution) throws Exception {
+ String executionId = execution.getId();
+
+ MessageBuilder messageBuilder = MessageBuilder.withPayload(execution).setHeader(ActivitiConstants.WELL_KNOWN_EXECUTION_ID_HEADER_KEY, executionId).setCorrelationId(executionId);
+
+ if (this.forwardProcessVariablesAsMessageHeaders) {
+ Map variables = execution.getVariables();
+
+ if ((variables != null) && (variables.size() > 0)) {
+ messageBuilder = messageBuilder.copyHeadersIfAbsent(variables);
+ }
+ }
+
+ Message msg = messageBuilder.setReplyChannel(replyChannel).build();
+
+ this.messagingTemplate.send(this.requestChannel, msg);
+ }
+
+ /**
+ * Verify the presence of references to a request and reply {@link org.springframework.integration.MessageChannel},
+ * the {@link ProcessEngine}, and setup the {@link org.springframework.integration.core.MessageHandler} that handles the replies
+ *
+ * @throws Exception
+ */
+ public void afterPropertiesSet() throws Exception {
+ Assert.state(this.replyChannel != null, "'replyChannel' can't be null!");
+ Assert.state(this.requestChannel != null, "'requestChannel' can't be null!");
+ Assert.state(this.processEngine != null, "'processEngine' can't be null!");
+
+ processService = this.processEngine.getRuntimeService();
+
+ if (this.platformTransactionManager != null) {
+ this.messagingTemplate.setTransactionManager(this.platformTransactionManager);
+ }
+
+ MessageHandler handler = new ReplyMessageHandler();
+
+ PollerMetadata pollerMetadata = new PollerMetadata();
+ pollerMetadata.setReceiveTimeout(-1);
+ pollerMetadata.setTransactionManager(this.platformTransactionManager);
+ pollerMetadata.setTrigger(new PeriodicTrigger(10));
+
+ ConsumerEndpointFactoryBean consumerEndpointFactoryBean = new ConsumerEndpointFactoryBean();
+ consumerEndpointFactoryBean.setAutoStartup(false);
+
+ if (this.replyChannel instanceof PollableChannel)
+ consumerEndpointFactoryBean.setPollerMetadata(pollerMetadata);
+
+ consumerEndpointFactoryBean.setBeanFactory(this.beanFactory);
+ consumerEndpointFactoryBean.setHandler(handler);
+ consumerEndpointFactoryBean.setInputChannel(this.replyChannel);
+ consumerEndpointFactoryBean.setBeanName(this.beanName);
+
+ AbstractEndpoint correlator = consumerEndpointFactoryBean.getObject();
+
+ if (correlator != null) {
+ correlator.start();
+ }
+ }
+
+ public void setBeanName(String s) {
+ this.beanName = s;
+ }
+
+ /**
+ * This class listens for results on the reply channel and causes the flow of execution to proceed inside the business process
+ */
+ class ReplyMessageHandler implements MessageHandler {
+ public void handleMessage(Message message) throws MessageHandlingException, MessageDeliveryException {
+ try {
+ MessageHeaders messageHeaders = message.getHeaders();
+ String executionId = (String) message.getHeaders().get(ActivitiConstants.WELL_KNOWN_EXECUTION_ID_HEADER_KEY);
+ Execution execution = processService.findExecutionById(executionId);
+ processEngine.getRuntimeService().signal(execution.getId(), AsyncActivityBehaviorMessagingGateway.class.getName() , messageHeaders);
+ } catch (Throwable throwable) {
+ throw new RuntimeException(throwable);
+ }
+ }
+ }
+}
diff --git a/spring-integration-activiti/src/main/java/org/springframework/integration/activiti/gateway/SyncActivityBehaviorMessagingGateway.java b/spring-integration-activiti/src/main/java/org/springframework/integration/activiti/gateway/SyncActivityBehaviorMessagingGateway.java
new file mode 100644
index 0000000000..558f9e594f
--- /dev/null
+++ b/spring-integration-activiti/src/main/java/org/springframework/integration/activiti/gateway/SyncActivityBehaviorMessagingGateway.java
@@ -0,0 +1,272 @@
+/* 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.springframework.integration.activiti.gateway;
+
+
+import org.activiti.engine.ProcessEngine;
+import org.activiti.engine.RuntimeService;
+import org.activiti.engine.impl.runtime.ExecutionEntity;
+import org.activiti.engine.runtime.Execution;
+import org.activiti.pvm.activity.ActivityBehavior;
+import org.activiti.pvm.activity.ActivityExecution;
+import org.apache.commons.lang.StringUtils;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.BeanFactoryAware;
+import org.springframework.beans.factory.BeanNameAware;
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.integration.*;
+import org.springframework.integration.activiti.ActivitiConstants;
+import org.springframework.integration.activiti.ProcessSupport;
+import org.springframework.integration.config.ConsumerEndpointFactoryBean;
+import org.springframework.integration.core.*;
+import org.springframework.integration.endpoint.AbstractEndpoint;
+import org.springframework.integration.scheduling.PollerMetadata;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.scheduling.support.PeriodicTrigger;
+import org.springframework.transaction.PlatformTransactionManager;
+import org.springframework.util.Assert;
+
+import java.util.Map;
+import java.util.Set;
+
+
+/**
+ * This class is plugged into an Activiti workflow. let's us plugin a custom {@link ActivityBehavior}.
+ * We need to build an {@link ActivityBehavior} that can send and receive the message,
+ * propagating the {@code executionId} and potentially process variables/ header variables.
+ *
+ * Simply svn co http://svn.codehaus.org/activiti/activiti/branches/alpha4-spring-integration-adapter that repository and then mvn clean install it.
+ *
+ * The class forwards ("pushes") control from a BPM process (in-thread) to a Spring Integration channel where, of course, Spring Integration acts as a client and
+ * can push the execution forward anyway it wants to.
+ *
+ * Possible use cases include forwarding the job through an outbound JMS adapter or gateway, forwarding the job through an XMPP adapter, or forwarding the job through
+ * to an outbound email adapter.
+ *
+ *
+ * The only requirement for the reply is that the {@link org.springframework.integration.Message} arrive on the #replyChannel and that it contain a header of
+ * {@link org.springframework.integration.activiti.ActivitiConstants#WELL_KNOWN_EXECUTION_ID_HEADER_KEY} (which the outbound {@link org.springframework.integration.Message} will have)
+ * so that the Activiti runtime can signal that execution has completed successfully.
+ *
+ * Thanks to Dave Syer and Tom Baeyens for the help brainstorming.
+ *
+ * This is very much like {@link org.springframework.integration.activiti.gateway.AsyncActivityBehaviorMessagingGateway} except that it assumes that
+ * the request/reply sequence will happen in the same Activiti transaction, so don't keep it waiting! Use the {@link org.springframework.integration.activiti.gateway.AsyncActivityBehaviorMessagingGateway}
+ * which is modeled as a wait-state and deals perfectly with asynchronous continuations.
+ *
+ * @author Josh Long
+ * @see org.activiti.engine.impl.bpmn.ReceiveTaskActivity the {@link org.activiti.pvm.activity.ActivityBehavior} impl that ships w/ Activiti that has the machinery to wake up when signaled
+ * @see org.activiti.engine.ProcessEngine the process engine instance is required to be able to use this namespace
+ * @see org.activiti.engine.impl.cfg.spring.ProcessEngineFactoryBean - use this class to create the aforementioned ProcessEngine instance!
+ */
+public class SyncActivityBehaviorMessagingGateway implements BeanFactoryAware, BeanNameAware, ActivityBehavior, InitializingBean {
+ /**
+ * Used to handle sending in a standard way
+ */
+ private MessagingTemplate messagingTemplate = new MessagingTemplate();
+
+ /**
+ * This is the channel on which we expect requests - {@link org.activiti.engine.runtime.Execution}s from Activiti - to arrive
+ */
+ private volatile MessageChannel requestChannel;
+
+ /**
+ * This is the channel on which we expect to send replies - ie, the result of our work in
+ * Spring Integration - back to Activiti, which should be waiting for the results
+ */
+ private volatile MessageChannel replyChannel;
+
+ /**
+ * Injected from Spring or some other mechanism. Recommended approach is through a {@link org.activiti.engine.impl.cfg.spring.ProcessEngineFactoryBean}
+ */
+ private volatile ProcessEngine processEngine;
+
+ /**
+ * Should we update the process variables based on the reply {@link org.springframework.integration.Message}'s {@link org.springframework.integration.MessageHeaders}?
+ */
+ private volatile boolean updateProcessVariablesFromReplyMessageHeaders = false;
+
+ /**
+ * Should we pass the workflow process variables as message headers when we send a message into the Spring Integration framework?
+ */
+ private volatile boolean forwardProcessVariablesAsMessageHeaders = false;
+
+ /**
+ * Forwarded to the {@link org.springframework.integration.core.MessagingTemplate} instance.
+ */
+ private volatile PlatformTransactionManager platformTransactionManager;
+
+ /**
+ * A reference to the {@link org.springframework.beans.factory.BeanFactory} that's hosting this component. Spring will inject this reference automatically assuming
+ * this object is hosted in a Spring context.
+ */
+ private volatile BeanFactory beanFactory;
+
+ /**
+ * The process engine instance that controls the Activiti PVM.
+ */
+ private RuntimeService runtimeService;
+
+ /**
+ * Provides common logic for things like sifting through inbound message headers and arriving at process variable candidates
+ */
+ private ProcessSupport processSupport = new ProcessSupport();
+ private String beanName;
+
+ @SuppressWarnings("unused")
+ public void setPlatformTransactionManager(PlatformTransactionManager platformTransactionManager) {
+ this.platformTransactionManager = platformTransactionManager;
+ }
+
+ @SuppressWarnings("unused")
+ public void setRequestChannel(MessageChannel requestChannel) {
+ this.requestChannel = requestChannel;
+ }
+
+ @SuppressWarnings("unused")
+ public void setReplyChannel(MessageChannel replyChannel) {
+ this.replyChannel = replyChannel;
+ }
+
+ @SuppressWarnings("unused")
+ public void setProcessEngine(ProcessEngine processEngine) {
+ this.processEngine = processEngine;
+ }
+
+ @SuppressWarnings("unused")
+ public void setForwardProcessVariablesAsMessageHeaders(boolean forwardProcessVariablesAsMessageHeaders) {
+ this.forwardProcessVariablesAsMessageHeaders = forwardProcessVariablesAsMessageHeaders;
+ }
+
+ @SuppressWarnings("unused")
+ public void setUpdateProcessVariablesFromReplyMessageHeaders(boolean updateProcessVariablesFromReplyMessageHeaders) {
+ this.updateProcessVariablesFromReplyMessageHeaders = updateProcessVariablesFromReplyMessageHeaders;
+ }
+
+ public void setBeanFactory(BeanFactory beanFactory)
+ throws BeansException {
+ this.beanFactory = beanFactory;
+ }
+
+ /**
+ * This is the main interface method from {@link ActivityBehavior}. It will be called when the BPMN process executes the node referencing this logic.
+ *
+ * @param execution the {@link ActivityExecution} as given to use by the engine
+ * @throws Exception
+ */
+ public void execute(ActivityExecution execution) throws Exception {
+ ExecutionEntity dbExecution = ((ExecutionEntity) execution);
+ String executionId = dbExecution.getId();
+ MessageBuilder messageBuilder = MessageBuilder.withPayload(execution)
+ .setHeader(ActivitiConstants.WELL_KNOWN_EXECUTION_ID_HEADER_KEY, executionId)
+ .setCorrelationId(executionId);
+
+ if (this.forwardProcessVariablesAsMessageHeaders) {
+ Map variables = dbExecution.getVariables();
+
+ if ((variables != null) && (variables.size() > 0)) {
+ messageBuilder = messageBuilder.copyHeadersIfAbsent(variables);
+ }
+ }
+
+ Message msg = messageBuilder.setReplyChannel(replyChannel).build();
+ Message response = this.messagingTemplate.sendAndReceive(requestChannel, msg);
+
+ handleReply(dbExecution, response);
+
+ }
+
+ /**
+ * The transaction won't have committed, so there's simply no need to
+ * @param execution the execution
+ * @param msg the inbound message
+ * @throws Exception escape hatch exception
+ */
+ protected void handleReply(ExecutionEntity execution, Message msg)
+ throws Exception {
+ MessageHeaders messageHeaders = msg.getHeaders();
+
+ if (this.updateProcessVariablesFromReplyMessageHeaders) {
+ Map vars = execution.getVariables();
+ Set existingVars = vars.keySet();
+ Map procVars = this.processSupport.processVariablesFromMessageHeaders(existingVars, messageHeaders);
+
+ for (String varName : procVars.keySet())
+ execution.getProcessInstance().setVariable(varName, procVars.get(varName));
+ }
+ }
+
+ /**
+ * Verify the presence of references to a request and reply {@link MessageChannel},
+ * the {@link ProcessEngine}, and setup the {@link org.springframework.integration.core.MessageHandler} that handles the replies
+ *
+ * @throws Exception
+ */
+ public void afterPropertiesSet() throws Exception {
+ Assert.state(this.replyChannel != null, "'replyChannel' can't be null!");
+ Assert.state(this.requestChannel != null, "'requestChannel' can't be null!");
+ Assert.state(this.processEngine != null, "'processEngine' can't be null!");
+
+ runtimeService = this.processEngine.getRuntimeService();
+
+ if (this.platformTransactionManager != null) {
+ this.messagingTemplate.setTransactionManager(this.platformTransactionManager);
+ }
+
+ MessageHandler handler = new ReplyMessageHandler();
+
+ PollerMetadata pollerMetadata = new PollerMetadata();
+ pollerMetadata.setReceiveTimeout(-1);
+ pollerMetadata.setTransactionManager(this.platformTransactionManager);
+ pollerMetadata.setTrigger(new PeriodicTrigger(10));
+
+ ConsumerEndpointFactoryBean consumerEndpointFactoryBean = new ConsumerEndpointFactoryBean();
+ consumerEndpointFactoryBean.setAutoStartup(false);
+
+ if (this.replyChannel instanceof PollableChannel) {
+ consumerEndpointFactoryBean.setPollerMetadata(pollerMetadata);
+ }
+
+ consumerEndpointFactoryBean.setBeanFactory(this.beanFactory);
+ consumerEndpointFactoryBean.setHandler(handler);
+ consumerEndpointFactoryBean.setInputChannel(this.replyChannel);
+ consumerEndpointFactoryBean.setBeanName(this.beanName);
+
+ AbstractEndpoint correlator = consumerEndpointFactoryBean.getObject();
+
+ if (correlator != null) {
+ correlator.start();
+ }
+ }
+
+ public void setBeanName(String s) {
+ this.beanName = s;
+ }
+
+ /**
+ * This class listens for results on the reply channel and causes the flow of execution to proceed inside the business process
+ */
+ class ReplyMessageHandler implements MessageHandler {
+ public void handleMessage(Message message) throws MessageHandlingException, MessageDeliveryException {
+ try {
+ MessageHeaders messageHeaders = message.getHeaders();
+ String executionId = (String) message.getHeaders().get(ActivitiConstants.WELL_KNOWN_EXECUTION_ID_HEADER_KEY);
+ Execution execution = runtimeService.findExecutionById(executionId);
+ processEngine.getRuntimeService().signal(execution.getId(), StringUtils.EMPTY, messageHeaders);
+ } catch (Throwable throwable) {
+ throw new RuntimeException(throwable);
+ }
+ }
+ }
+}
diff --git a/spring-integration-activiti/src/main/resources/META-INF/MANIFEST.MF b/spring-integration-activiti/src/main/resources/META-INF/MANIFEST.MF
new file mode 100644
index 0000000000..e69de29bb2
diff --git a/spring-integration-activiti/src/main/resources/META-INF/spring.handlers b/spring-integration-activiti/src/main/resources/META-INF/spring.handlers
new file mode 100644
index 0000000000..57bcf69e07
--- /dev/null
+++ b/spring-integration-activiti/src/main/resources/META-INF/spring.handlers
@@ -0,0 +1 @@
+http\://www.springframework.org/schema/integration/activiti=org.springframework.integration.activiti.config.ActivitiNamespaceHandler
\ No newline at end of file
diff --git a/spring-integration-activiti/src/main/resources/META-INF/spring.schemas b/spring-integration-activiti/src/main/resources/META-INF/spring.schemas
new file mode 100644
index 0000000000..31d7c9a8a5
--- /dev/null
+++ b/spring-integration-activiti/src/main/resources/META-INF/spring.schemas
@@ -0,0 +1,2 @@
+http\://www.springframework.org/schema/integration/activiti/spring-integration-activiti-2.0.xsd=org/springframework/integration/activiti/config/spring-integration-activiti-2.0.xsd
+http\://www.springframework.org/schema/integration/activiti/spring-integration-activiti.xsd=org/springframework/integration/activiti/config/spring-integration-activiti-2.0.xsd
diff --git a/spring-integration-activiti/src/main/resources/org/springframework/integration/activiti/config/spring-integration-activiti-2.0.xsd b/spring-integration-activiti/src/main/resources/org/springframework/integration/activiti/config/spring-integration-activiti-2.0.xsd
new file mode 100644
index 0000000000..01b76f170d
--- /dev/null
+++ b/spring-integration-activiti/src/main/resources/org/springframework/integration/activiti/config/spring-integration-activiti-2.0.xsd
@@ -0,0 +1,119 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ Defines an inbound Activiti Messaging Gateway.
+
+ This bean should be referenced from a BPMN workflow artifact. Once entered, execution of the task will be dispatched to Spring Integration
+ where Spring Integration can then do whatever it likes so long as execution ultimately travels through the 'reply-channel'.
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/spring-integration-activiti/src/test/java/org/springframework/integration/activiti/GatewayTest.java b/spring-integration-activiti/src/test/java/org/springframework/integration/activiti/GatewayTest.java
new file mode 100644
index 0000000000..c062de97bf
--- /dev/null
+++ b/spring-integration-activiti/src/test/java/org/springframework/integration/activiti/GatewayTest.java
@@ -0,0 +1,48 @@
+/* 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.springframework.integration.activiti;
+
+import org.activiti.engine.ProcessEngine;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.AbstractJUnit4SpringContextTests;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+import java.util.HashMap;
+import java.util.Map;
+
+
+/**
+ * This component demonstrates creating a {@link org.springframework.integration.activiti.gateway.AsyncActivityBehaviorMessagingGateway} (factoried from Spring)
+ * and exposed for use in a BPMN 2 process.
+ *
+ * @author Josh Long
+ */
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+public class GatewayTest extends AbstractJUnit4SpringContextTests {
+
+ @Autowired private ProcessEngine processEngine ;
+
+ @Test
+ public void testGateway() throws Throwable {
+ processEngine.getRepositoryService().createDeployment().addClasspathResource("processes/si_gateway_example.bpmn20.xml").deploy();
+ Map vars = new HashMap();
+ vars.put("customerId", 232);
+ processEngine.getRuntimeService().startProcessInstanceByKey("sigatewayProcess", vars);
+ Thread.sleep(10000);
+
+ }
+}
diff --git a/spring-integration-activiti/src/test/java/org/springframework/integration/activiti/OutboundAdapterTest.java b/spring-integration-activiti/src/test/java/org/springframework/integration/activiti/OutboundAdapterTest.java
new file mode 100644
index 0000000000..e3f3c0ca44
--- /dev/null
+++ b/spring-integration-activiti/src/test/java/org/springframework/integration/activiti/OutboundAdapterTest.java
@@ -0,0 +1,44 @@
+/* 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.springframework.integration.activiti;
+
+import org.junit.Ignore;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.integration.MessageChannel;
+import org.springframework.integration.core.MessagingTemplate;
+import org.springframework.integration.support.MessageBuilder;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
+
+
+@ContextConfiguration
+@RunWith(SpringJUnit4ClassRunner.class)
+
+public class OutboundAdapterTest {
+
+
+ @Value( "#{tc}")
+ private MessageChannel messageChannel ;
+
+ private MessagingTemplate messagingTemplate = new MessagingTemplate();
+
+ @Test
+ public void testOutboundAdapter() throws Throwable {
+ this.messagingTemplate.send( this.messageChannel , MessageBuilder.withPayload( "hello, from "+ System.currentTimeMillis())
+ .setHeader( ActivitiConstants.WELL_KNOWN_PROCESS_DEFINITION_NAME_HEADER_KEY+"customerId",2324).build());
+ }
+
+}
diff --git a/spring-integration-activiti/src/test/java/org/springframework/integration/activiti/impls/AnnouncingServiceActivator.java b/spring-integration-activiti/src/test/java/org/springframework/integration/activiti/impls/AnnouncingServiceActivator.java
new file mode 100644
index 0000000000..71ac909fbd
--- /dev/null
+++ b/spring-integration-activiti/src/test/java/org/springframework/integration/activiti/impls/AnnouncingServiceActivator.java
@@ -0,0 +1,24 @@
+/* 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.springframework.integration.activiti.impls;
+
+import org.springframework.integration.Message;
+import org.springframework.integration.annotation.ServiceActivator;
+
+
+public class AnnouncingServiceActivator {
+ @ServiceActivator
+ public void hello(Message msg) throws Throwable {
+ System.out.println("got it ");
+ }
+}
diff --git a/spring-integration-activiti/src/test/java/org/springframework/integration/activiti/impls/DelayedchoServiceActivator.java b/spring-integration-activiti/src/test/java/org/springframework/integration/activiti/impls/DelayedchoServiceActivator.java
new file mode 100644
index 0000000000..636b7bd34d
--- /dev/null
+++ b/spring-integration-activiti/src/test/java/org/springframework/integration/activiti/impls/DelayedchoServiceActivator.java
@@ -0,0 +1,43 @@
+/* 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.springframework.integration.activiti.impls;
+
+import org.springframework.integration.Message;
+import org.springframework.integration.activiti.ActivitiConstants;
+import org.springframework.integration.annotation.ServiceActivator;
+import org.springframework.integration.support.MessageBuilder;
+
+
+/**
+ * This is a trivial component that demonstrates that the flow of control lives outside of the BPMN process (ie, that we have truly implemented a wait-state)
+ * and that gateway does the right thing if you send back message headers whose key corresponds with process variables, or new message headers that you want as process variables.
+ *
+ * The only requirement for any component wishing to reply to an Activiti business process is that there be a header named using the static variable
+ * {@link org.springframework.integration.activiti.ActivitiConstants#WELL_KNOWN_EXECUTION_ID_HEADER_KEY}.
+ *
+ * @author Josh Long
+ */
+public class DelayedchoServiceActivator {
+
+ @ServiceActivator
+ public Message sayHello(Message requestComingFromActiviti)
+ throws Throwable {
+ System.out.println("entering ServiceActivator:sayHello");
+ Thread.sleep(5 * 1000);
+ System.out.println("exiting ServiceActivator:sayHello");
+ return MessageBuilder.withPayload(requestComingFromActiviti.getPayload()).
+ copyHeadersIfAbsent(requestComingFromActiviti.getHeaders()).setHeader(
+ ActivitiConstants.WELL_KNOWN_SPRING_INTEGRATION_HEADER_PREFIX + "test", "1 + 1").
+ build();
+ }
+}
diff --git a/spring-integration-activiti/src/test/java/org/springframework/integration/activiti/impls/SimpleCustomActivityBehavior.java b/spring-integration-activiti/src/test/java/org/springframework/integration/activiti/impls/SimpleCustomActivityBehavior.java
new file mode 100644
index 0000000000..e7bc049383
--- /dev/null
+++ b/spring-integration-activiti/src/test/java/org/springframework/integration/activiti/impls/SimpleCustomActivityBehavior.java
@@ -0,0 +1,49 @@
+/* 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.springframework.integration.activiti.impls;
+
+import org.activiti.engine.impl.runtime.ExecutionEntity;
+
+import org.activiti.pvm.activity.ActivityBehavior;
+import org.activiti.pvm.activity.ActivityExecution;
+import org.activiti.pvm.process.PvmTransition;
+
+import org.springframework.beans.factory.InitializingBean;
+
+import java.util.List;
+
+
+/**
+ * A simple component that implements {@link ActivityBehavior}
+ *
+ * @author Josh Long
+ */
+public class SimpleCustomActivityBehavior implements ActivityBehavior, InitializingBean {
+ public void execute(ActivityExecution dbExecution)
+ throws Exception {
+ System.out.println("Hello from a custom ActivityBehavior hosted in the Spring context. " + this);
+
+
+ for (String varName : dbExecution.getVariables().keySet())
+ System.out.println(varName + "=" + dbExecution.getVariable(varName));
+
+
+ List transitions = dbExecution.getActivity().getOutgoingTransitions();
+
+ dbExecution.take(((null == transitions) || (transitions.size() == 0)) ? null : transitions.get(0));
+ }
+
+ public void afterPropertiesSet() throws Exception {
+ System.out.println("Starting " + SimpleCustomActivityBehavior.class.getName());
+ }
+}
diff --git a/spring-integration-activiti/src/test/java/org/springframework/integration/activiti/impls/SimpleProcessTriggeringMessageSource.java b/spring-integration-activiti/src/test/java/org/springframework/integration/activiti/impls/SimpleProcessTriggeringMessageSource.java
new file mode 100644
index 0000000000..0332686c67
--- /dev/null
+++ b/spring-integration-activiti/src/test/java/org/springframework/integration/activiti/impls/SimpleProcessTriggeringMessageSource.java
@@ -0,0 +1,45 @@
+/* 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.springframework.integration.activiti.impls;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.springframework.integration.Message;
+import org.springframework.integration.activiti.ActivitiConstants;
+import org.springframework.integration.core.MessageSource;
+import org.springframework.integration.support.MessageBuilder;
+
+
+/**
+ * Triggers a BPMN process that has process variables.
+ *
+ * @author Josh Long
+ */
+public class SimpleProcessTriggeringMessageSource implements MessageSource {
+
+ private final Log logger = LogFactory.getLog(SimpleProcessTriggeringMessageSource.class.getName());
+
+ public Message receive() {
+ try {
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ logger.debug("couldn't sleep 1000 ms", e);
+ }
+ return MessageBuilder.withPayload(
+ "hello from " + System.currentTimeMillis()
+ ).
+ setHeader(ActivitiConstants.WELL_KNOWN_SPRING_INTEGRATION_HEADER_PREFIX + "customerId", 232).
+ setHeader( ActivitiConstants.WELL_KNOWN_PROCESS_DEFINITION_NAME_HEADER_KEY, "helloWorldProcess" ).build();
+
+ }
+}
diff --git a/spring-integration-activiti/src/test/resources/org/springframework/integration/activiti/GatewayTest-context.xml b/spring-integration-activiti/src/test/resources/org/springframework/integration/activiti/GatewayTest-context.xml
new file mode 100644
index 0000000000..bc0bef42c4
--- /dev/null
+++ b/spring-integration-activiti/src/test/resources/org/springframework/integration/activiti/GatewayTest-context.xml
@@ -0,0 +1,87 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/spring-integration-activiti/src/test/resources/org/springframework/integration/activiti/OutboundAdapterTest-context.xml b/spring-integration-activiti/src/test/resources/org/springframework/integration/activiti/OutboundAdapterTest-context.xml
new file mode 100644
index 0000000000..e802cd7c19
--- /dev/null
+++ b/spring-integration-activiti/src/test/resources/org/springframework/integration/activiti/OutboundAdapterTest-context.xml
@@ -0,0 +1,80 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+ classpath:/processes/hello-world.bpmn20.xml
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/spring-integration-activiti/src/test/resources/processes/hello-world.bpmn20.xml b/spring-integration-activiti/src/test/resources/processes/hello-world.bpmn20.xml
new file mode 100644
index 0000000000..4fa42827e3
--- /dev/null
+++ b/spring-integration-activiti/src/test/resources/processes/hello-world.bpmn20.xml
@@ -0,0 +1,24 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/spring-integration-activiti/src/test/resources/processes/si_gateway_example.bpmn20.xml b/spring-integration-activiti/src/test/resources/processes/si_gateway_example.bpmn20.xml
new file mode 100644
index 0000000000..475d1ff93d
--- /dev/null
+++ b/spring-integration-activiti/src/test/resources/processes/si_gateway_example.bpmn20.xml
@@ -0,0 +1,24 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/spring-integration-activiti/template.mf b/spring-integration-activiti/template.mf
new file mode 100644
index 0000000000..d92598ec33
--- /dev/null
+++ b/spring-integration-activiti/template.mf
@@ -0,0 +1,15 @@
+Bundle-SymbolicName: org.springframework.integration.activiti
+Bundle-Name: Spring Integration Activiti Support
+Bundle-Vendor: SpringSource
+Bundle-ManifestVersion: 2
+Import-Template:
+ org.apache.commons.logging;version="[1.1.1, 2.0.0)",
+ org.apache.commons.lang.*;version="[2.5.0, 3.0.0)",
+ org.springframework.integration.*;version="[2.0.0, 2.0.1)",
+ org.springframework.beans.*;version="[3.0.0, 4.0.0)",
+ org.springframework.context;version="[3.0.0, 4.0.0)",
+ org.springframework.core.*;version="[3.0.0, 4.0.0)",
+ org.springframework.util;version="[3.0.0, 4.0.0)",
+ org.jivesoftware.*;version="[3.1.0, 4.0.0)",
+ javax.*;version="0",
+ org.w3c.dom.*;version="0"
diff --git a/src/assembly/distribution.xml b/src/assembly/distribution.xml
index 6b3326c9a5..5463bf3959 100644
--- a/src/assembly/distribution.xml
+++ b/src/assembly/distribution.xml
@@ -82,6 +82,7 @@
org.springframework.integration:spring-integration-sftp
org.springframework.integration:spring-integration-ftp
org.springframework.integration:spring-integration-twitter
+ org.springframework.integration:spring-integration-activiti
dist