AsyncOperationBean


1   package org.example;
2   
3   import javax.ejb.EJBException  ;
4   import javax.ejb.MessageDrivenBean  ;
5   import javax.ejb.MessageDrivenContext  ;
6   import javax.jms.JMSException  ;
7   import javax.jms.MapMessage  ;
8   import javax.jms.Message  ;
9   import javax.jms.MessageListener  ;
10  
11  import org.apache.commons.logging.Log;
12  import org.apache.commons.logging.LogFactory;
13  
14  import org.jbpm.db.GraphSession;
15  import org.jbpm.db.JbpmSession;
16  import org.jbpm.db.JbpmSessionFactory;
17  import org.jbpm.graph.exe.Token;
18  import org.jbpm.graph.exe.ProcessInstance;
19  
20  public class AsyncOperationBean implements MessageDrivenBean  , MessageListener   {
21    
22    private transient MessageDrivenContext   context;
23    
24    /** JMS property for the operation name */
25    public static final String   OPERATION_PROPERTY = "Operation";
26    /** Item name for the process identifier */
27    public static final String   PROCESS_ID_ITEM = "_$processId";
28    /** Item name for the token identifier */
29    public static final String   TOKEN_ID_ITEM = "_$tokenId";
30    
31    private static JbpmSessionFactory jbpmSessionFactory = 
32      JbpmSessionFactory.buildJbpmSessionFactory();
33    private static final Log log = LogFactory.getLog(AsyncOperationBean.class);
34  
35    public void setMessageDrivenContext(MessageDrivenContext   context) throws EJBException   {
36      this.context = context;
37    }
38    
39    public void ejbCreate() {
40    }
41    
42    public void onMessage(Message   message) {
43      if (message instanceof MapMessage  ) {
44        MapMessage   mapMessage = (MapMessage  ) message;
45        try {
46          // get operation name from message header
47          String   operation = mapMessage.getStringProperty(OPERATION_PROPERTY);
48          // get process and token identifiers from message body
49          long processId = mapMessage.getLong(PROCESS_ID_ITEM);
50          long tokenId = mapMessage.getLong(TOKEN_ID_ITEM);
51          log.debug("received message: operation=" + operation + ", tokenId" + tokenId);
52          
53          // execute the indicated operation
54          executeOperation(operation);
55          
56          // send a signal to the process
57          signalToken(processId, tokenId);
58        }
59        catch (JMSException   e) {
60          log.error("could not read jms message", e);
61          context.setRollbackOnly();
62        }
63      }
64      else {
65        log.warn("message of wrong type: " + message.getClass().getName());
66      }
67    }
68    
69    protected void executeOperation(String   operation) {
70      log.info("executing operation: " + operation);
71    }
72    
73    protected void signalToken(long processId, long tokenId) {
74      JbpmSession jbpmSession = jbpmSessionFactory.openJbpmSession();
75      jbpmSession.beginTransaction();
76      
77      // load process instance from the database
78      GraphSession graphSession = jbpmSession.getGraphSession();
79      ProcessInstance processInstance = graphSession.loadProcessInstance(processId);
80      log.debug("loaded process instance: " + processId);
81      // lock the process instance to avoid race condition
82      graphSession.lockProcessInstance(processInstance);
83      
84      // now load the token
85      Token token = graphSession.loadToken(tokenId);
86      log.debug("loaded token: " + tokenId);
87      // signal we're done here
88      token.signal();
89      log.debug("signaled token: " + tokenId);
90      
91      jbpmSession.commitTransaction();
92      jbpmSession.close();
93    }
94  
95    public void ejbRemove() throws EJBException   {
96    }
97  }
98
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags