Author: thomas.diesler(a)jboss.com Date: 2008-11-01 08:31:55 -0400 (Sat, 01 Nov 2008) New Revision: 2743 Added: projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/runtime/TransactionAssociation.java Modified: projects/spec/trunk/modules/cts/src/test/java/org/jbpm/test/cts/transaction/TxRequiredTest.java projects/spec/trunk/modules/dialects/api10/src/main/java/org/jbpm/dialect/api10/ProcessMarshaller.java projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/model/MessageImpl.java projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/runtime/MessageSender.java projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/runtime/TransactionInterceptor.java projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/service/HibernatePersistenceServiceImpl.java projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/service/MessageServiceImpl.java Log: Add TransactionAssociation Modified: projects/spec/trunk/modules/cts/src/test/java/org/jbpm/test/cts/transaction/TxRequiredTest.java =================================================================== --- projects/spec/trunk/modules/cts/src/test/java/org/jbpm/test/cts/transaction/TxRequiredTest.java 2008-11-01 12:04:46 UTC (rev 2742) +++ projects/spec/trunk/modules/cts/src/test/java/org/jbpm/test/cts/transaction/TxRequiredTest.java 2008-11-01 12:31:55 UTC (rev 2743) @@ -32,6 +32,7 @@ import org.jbpm.api.model.Process; import org.jbpm.api.model.ProcessDefinition; import org.jbpm.api.model.Group.GroupType; +import org.jbpm.api.model.Task.TaskType; import org.jbpm.api.model.builder.ProcessBuilder; import org.jbpm.api.service.ProcessBuilderService; import org.jbpm.api.test.CTSTestCase; @@ -69,9 +70,12 @@ ProcessBuilder procBuilder = ProcessBuilderService.locateProcessBuilder(); procBuilder.addProcess("RequiresTxTest"); procBuilder.addStartEvent("Start").addSequenceFlow("TaskA"); - procBuilder.addTask("TaskA").addGroupRef("TxRequired").addSequenceFlow("TaskB"); - procBuilder.addTask("TaskB").addGroupRef("TxRequired").addSequenceFlow("End").addEndEvent("End"); + procBuilder.addTask("TaskA", TaskType.Send).addOutMessageRef("TaskAMessage").addGroupRef("TxRequired").addSequenceFlow("TaskB"); + procBuilder.addTask("TaskB", TaskType.Send).addOutMessageRef("TaskBMessage").addGroupRef("TxRequired").addSequenceFlow("End"); procBuilder.addGroup(GroupType.Transaction, "TxRequired").addProperty(Constants.PROP_TX_TYPE, TxType.REQUIRED); + procBuilder.addProcessMessage("TaskAMessage").addToRef(getTestID()).addProperty("msgProp", "msgA"); + procBuilder.addProcessMessage("TaskBMessage").addToRef(getTestID()).addProperty("msgProp", "msgB");; + procBuilder.addEndEvent("End"); return procBuilder.getProcessDefinition(); } } Modified: projects/spec/trunk/modules/dialects/api10/src/main/java/org/jbpm/dialect/api10/ProcessMarshaller.java =================================================================== --- projects/spec/trunk/modules/dialects/api10/src/main/java/org/jbpm/dialect/api10/ProcessMarshaller.java 2008-11-01 12:04:46 UTC (rev 2742) +++ projects/spec/trunk/modules/dialects/api10/src/main/java/org/jbpm/dialect/api10/ProcessMarshaller.java 2008-11-01 12:31:55 UTC (rev 2743) @@ -397,8 +397,7 @@ private String getParticipant(Participant par) { - String parName = par.getName().getCanonicalName(); - return parName; + return par != null ? par.getName().getCanonicalName() : null; } private JAXBFlow getJAXBFlow(SequenceFlow flow) Modified: projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/model/MessageImpl.java =================================================================== --- projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/model/MessageImpl.java 2008-11-01 12:04:46 UTC (rev 2742) +++ projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/model/MessageImpl.java 2008-11-01 12:31:55 UTC (rev 2743) @@ -28,6 +28,7 @@ import javax.persistence.Basic; import javax.persistence.CascadeType; import javax.persistence.Entity; +import javax.persistence.FetchType; import javax.persistence.GeneratedValue; import javax.persistence.Id; import javax.persistence.OneToOne; @@ -59,10 +60,10 @@ @Basic private String name; - @OneToOne(cascade = { CascadeType.ALL }, targetEntity = ParticipantImpl.class) + @OneToOne(cascade = { CascadeType.ALL }, fetch = FetchType.EAGER, targetEntity = ParticipantImpl.class) private Participant fromRef; - @OneToOne(cascade = { CascadeType.ALL }, targetEntity = ParticipantImpl.class) + @OneToOne(cascade = { CascadeType.ALL }, fetch = FetchType.EAGER, targetEntity = ParticipantImpl.class) private Participant toRef; @OneToOne(cascade = { CascadeType.ALL }, targetEntity = PropertySupportImpl.class) Modified: projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/runtime/MessageSender.java =================================================================== --- projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/runtime/MessageSender.java 2008-11-01 12:04:46 UTC (rev 2742) +++ projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/runtime/MessageSender.java 2008-11-01 12:31:55 UTC (rev 2743) @@ -29,6 +29,7 @@ import org.jbpm.api.model.Message; import org.jbpm.api.model.Node; import org.jbpm.api.model.Participant; +import org.jbpm.api.model.Property; import org.jbpm.api.model.builder.MessageBuilder; import org.jbpm.api.runtime.ExecutionContext; import org.jbpm.api.runtime.Token; @@ -77,14 +78,17 @@ msgBuilder.newMessage(messageRef.getName()); msgBuilder.addToRef(targetID); - // Copy the properties from the execution context - // to the send message + // Copy the properties from the execution context to the send message for (String propName : messageRef.getPropertyNames()) { - Object value = exContext.getAttachment(propName); - if (value == null) - throw new IllegalStateException("Cannot obtain property: " + propName); - + Property prop = messageRef.getProperty(propName); + Object value = prop.getValue(); + if (prop.isCorrelation()) + { + value = exContext.getAttachment(propName); + if (value == null) + throw new IllegalStateException("Cannot obtain property: " + propName); + } msgBuilder.addProperty(propName, value); } MessageImpl msg = (MessageImpl)msgBuilder.getMessage(); @@ -94,4 +98,4 @@ MessageService msgService = engine.getService(MessageService.class); msgService.sendMessage(msg); } -} \ No newline at end of file +} Added: projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/runtime/TransactionAssociation.java =================================================================== --- projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/runtime/TransactionAssociation.java (rev 0) +++ projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/runtime/TransactionAssociation.java 2008-11-01 12:31:55 UTC (rev 2743) @@ -0,0 +1,47 @@ +/* + * JBoss, Home of Professional Open Source + * Copyright 2005, JBoss Inc., and individual contributors as indicated + * by the @authors tag. See the copyright.txt in the distribution for a + * full listing of individual contributors. + * + * This is free software; you can redistribute it and/or modify it + * under the terms of the GNU Lesser General Public License as + * published by the Free Software Foundation; either version 2.1 of + * the License, or (at your option) any later version. + * + * This software is distributed in the hope that it will be useful, + * but WITHOUT ANY WARRANTY; without even the implied warranty of + * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU + * Lesser General Public License for more details. + * + * You should have received a copy of the GNU Lesser General Public + * License along with this software; if not, write to the Free + * Software Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA + * 02110-1301 USA, or see the FSF site: http://www.fsf.org. + */ +package org.jbpm.ri.runtime; + +//$Id$ + +import org.hibernate.Transaction; + +/** + * A thread local association for the current transaction + * + * @author thomas.diesler(a)jboss.com + * @since 01-Nov-2008 + */ +public class TransactionAssociation +{ + private static final ThreadLocal<Transaction> txAssociation = new ThreadLocal<Transaction>(); + + public static void setTransaction(Transaction tx) + { + txAssociation.set(tx); + } + + public static Transaction getTransaction() + { + return txAssociation.get(); + } +} \ No newline at end of file Property changes on: projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/runtime/TransactionAssociation.java ___________________________________________________________________ Name: svn:keywords + Id Revision Name: svn:eol-style + LF Modified: projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/runtime/TransactionInterceptor.java =================================================================== --- projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/runtime/TransactionInterceptor.java 2008-11-01 12:04:46 UTC (rev 2742) +++ projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/runtime/TransactionInterceptor.java 2008-11-01 12:31:55 UTC (rev 2743) @@ -42,8 +42,6 @@ { final static Logger log = LoggerFactory.getLogger(TransactionInterceptor.class); - private PersistenceService service; - @Override public void execute(RuntimeContext rtContext) { @@ -53,10 +51,13 @@ // Begin the Tx Session session = token.getSession(); Transaction tx = session.beginTransaction(); + TransactionAssociation.setTransaction(tx); + try { // Load the node - PersistenceService service = getPersistenceService(node.getProcessEngine()); + ProcessEngine engine = node.getProcessEngine(); + PersistenceService service = engine.getService(PersistenceService.class); node = service.loadNode(session, node.getClass(), node.getKey()); // Call the next interceptor in the chain @@ -83,13 +84,9 @@ } throw rte; } + finally + { + TransactionAssociation.setTransaction(null); + } } - - private PersistenceService getPersistenceService(ProcessEngine engine) - { - if (service == null) - service = engine.getService(PersistenceService.class); - - return service; - } -} \ No newline at end of file +} Modified: projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/service/HibernatePersistenceServiceImpl.java =================================================================== --- projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/service/HibernatePersistenceServiceImpl.java 2008-11-01 12:04:46 UTC (rev 2742) +++ projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/service/HibernatePersistenceServiceImpl.java 2008-11-01 12:31:55 UTC (rev 2743) @@ -32,6 +32,7 @@ import org.hibernate.ObjectNotFoundException; import org.hibernate.Session; import org.hibernate.SessionFactory; +import org.hibernate.StaleStateException; import org.hibernate.Transaction; import org.hibernate.cfg.AnnotationConfiguration; import org.jbpm.api.NodeNotFoundException; @@ -208,6 +209,10 @@ session.delete(proc); tx.commit(); } + catch (StaleStateException ex) + { + System.out.println("FIXME: intermitent " + ex); + } finally { session.close(); Modified: projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/service/MessageServiceImpl.java =================================================================== --- projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/service/MessageServiceImpl.java 2008-11-01 12:04:46 UTC (rev 2742) +++ projects/spec/trunk/modules/ri/src/main/java/org/jbpm/ri/service/MessageServiceImpl.java 2008-11-01 12:31:55 UTC (rev 2743) @@ -23,10 +23,17 @@ // $Id$ +import javax.transaction.Status; +import javax.transaction.Synchronization; + +import org.hibernate.Transaction; import org.jbpm.api.client.MessageListener; import org.jbpm.api.client.ProcessEngine; import org.jbpm.api.model.Message; import org.jbpm.api.service.MessageService; +import org.jbpm.ri.runtime.TransactionAssociation; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; /** * The ProcessEngine sends messages through the MessageService. @@ -39,9 +46,64 @@ */ public class MessageServiceImpl extends MessageService implements MutableService { + // Provide logging + final Logger log = LoggerFactory.getLogger(MessageServiceImpl.class); + @Override public void setProcessEngine(ProcessEngine engine) { super.setProcessEngine(engine); } + + @Override + public void sendMessage(Message msg) + { + Transaction tx = TransactionAssociation.getTransaction(); + sendMessageTx(msg, null); + } + + private void sendMessageTx(Message msg, Transaction tx) + { + if (tx != null) + { + tx.registerSynchronization(new MessageWrapper(this, msg)); + } + else + { + super.sendMessage(msg); + } + } + + public class MessageWrapper implements Synchronization + { + private MessageServiceImpl service; + private Message msg; + + + protected MessageWrapper(MessageServiceImpl service, Message msg) + { + this.service = service; + this.msg = msg; + } + + @Override + public void afterCompletion(int status) + { + if (status != Status.STATUS_COMMITTED) + { + log.debug("Ignore message after completion: status=" + status + "," + msg); + } + else + { + log.debug("Send message after completion " + msg); + service.sendMessageTx(msg, null); + } + } + + @Override + public void beforeCompletion() + { + // nothing to do + } + } }