1 of 1 people found this helpful
Hi, basically you have two different things to solve, one is about how to locate the correspondant session and the other is how to choose the right process to notify the event right?
For locating the right process to signal and to restore the right session we usualy implement a business layer that store a business key and information about your workitem or processid.
Take a look at the following example, using that layer you can also map a JMS message to a specific event or workitem to complete. I think that you can use the same approach provided in the example.
Hope it helps!
Hi mauricio, thanks for the reply. You identified my issues accurately. I agree that I need to implement a business layer to locate the right process and to restore the right session. A question was how to locate the right process and restore the right session. I was also thinking about storing some info (e.g., process instance id) or some serialized objects and restore them before sending a signal event. I just didn't know how.
I went through your example code, but I couldn't find code where a process instance is located or a session is restoreed based on some stored key data. It is possible I didn't see it. Could you please advise which code does this function?
Basically, I'd like to Knowledge base APIs doing that kind of job.
Many thanks again!
I am implementing a similar solution and agree with Mauricio that you need to implement a business layer. I am not as familiar with Drools and took a different approach which I will describe briefly. I implemented the Selective Consumer pattern as described in Hohpe and Woolf's Enterprise Integration Patterns book. In my case processes were started before the business key was assigned, so my messages do not always pass around the same business key. This makes it harder to link them back to a running instance of a process. I ended up writing a manager component in mule that receives JMS messages that may need to go to JBPM processes. This component manages the start and end of process instances. When it receives a messages it extracts the business key from the message and looks to a self contained map to determine if this key is related to an active process. If we find a match I pass the message to the active state in the process. These are defined as custom nodes in the process. The class implements ExternalActivityBehavior. It is responsible for making sure it received the right message. The value it returns is used by a decision node to determine if we need to keep waiting or can move on to the next wait state.
This is just one approach. I welcome feedback from those who have been at this longer than I have. It is working for me, but I am struggling with group timers which I am trying to implement in order to kill the process if the right message does not show up in a specified period of time. Hope this adds an option for you to pursue.
Thanks, Mauricio and Jim! After receiving replies from both of you, I did give some thoughts on this, and read some of blogs that Kris V. was involved. Here is my plan:
In my business layer, when a JMS message arrives (it can be Web service, EJB, POJO, etc. - does not matter), it parses the message/object for a unique correlation id (the correlation id can consist of many field values to make it unique if necessary) and other key fields. It uses the emf to persist (cache) the knowledge session, and saves the knowledge session id and process instance id (after startprocess) along with the correlation id into a mapping database table. So, in the mapping table has the correlation id - session id - process instance id and other key fields per table row.
Later, an external client (partner) sends a JMS message that has the correlation id that matches the above correlation id. Using the correlation id from the message, it will retrieve the session id and process instance id from the mapping table. Then, using the JPAKnowledgeService.loadStatefulKnowledgeSession (session_id, kbase, null, env), it will restore the session. Since the session can have multiple process instances, it will locate the right process instance from the session using the ksession.getProcessInstance (process id).
Once it gets the process instance, using the event type (the event type can also come from the external message, or some other mapping is possible), it can inject the intermedidate message event to the right place, using the signal event, e.g., processInstance.signalEvent("event Type", data); Once the process is complete, we can purge the corresponding data from the mapping table.
If I don't use the emf, I think I can use the kbase.getStatefulKnowledgeSessions() to get a collection of sessions - there is an assumption that the kbase is static. Then, in a loop, I can find the right session by comparing the session.getId() with the session id from the mapping table. Since the kbase is static which is per JVM, it may cause some issues to run the jBPM 5 in a clustered environment. So, I prefer to restore the session from the database (persistence).
That's the basic idea. Any comments will be appreciated.
I used to use that BPEL, the correlation support is part of the process configuration (usign some BPEL IDE), not part of business layer. In jBPM 5, it seems to me that I need to handle the correlation support as above.
As I addressed above, to support an async message event (external), I store a unique correlation id, with a session id and a process instance id. When an intermediate message arrives, I restore a session and a procss instance is based on the incoming message correlation id field. Using the following code:
ksession.signalEvent(eventType, eventData, restoredProcessInstanceId);
It worked nicely, but there is a problem. If the async message event is in a reusable sub-process (not in the top-level process), the above method won't work because the sub-process has its own process id. I looked at the jBPM ProcessInstanceInfo database table, there is no link between the parent and the sub-process.
So, how do I get the sub-process process id when the top-level hit the async message event and returns back to the process caller?
Any suggestion will be appreciated.
OK, it looks like I self-answered my question above.
Now, I understand why Krisv V. said that use kession.signalEvent for the external aync event.
So, I used
ksession.signalEvent(eventType, eventData); // without the top-level restoredProcessInstanceId.
Then, the jBPM 5 automatically finds the corresponding sub-process with the intermediate message event.
It worked as long as I don't share the ksession with another top-level bpmn process.
Sorry for the direct question (and my poor english), actually i'm to end of working day.
Even just a request correlated to a response by an id defined in the payload of the messages.
Maybe tomorrow i add more details about what i want to do...
Hi i'm back with some delay.
My question is: can the process engine handle message correlation automaticaly in according to BPMN 2.0 spec (section 8.3.2)? or i must to handle message correlation externaly to proccess engine?
Let me explain, i have understood that a process can wait a message by a intermediate catch message event or by a receive task.
In the first case my application can signal the arrival of message by only
because if i use
only the start (and not the intermediate catch) message event is triggered
In second case my application can signal the arrival of message by
it is correct?
then, if my application needs to be aware of the process id or the workItemId for signaling the arrival of a message when the correlationProperty come into play?
another question is: what is the difference at the conceptual level from intermediate catch message event and receive task? when is appropriate to use one rather than other?
After my previous post, using ksession.signalEvent(eventType, eventData) without the process instance id, I realized that the method does NOT work well. The jBPM engine is supposed to find the proper process instance id in the session, but many times it did NOT point the correct process instance id when there are some left-over data in the EventType database table (due to multiple process instances for the same process definition are running (of course they have the same eventType name) or previous sessions failed and did not clean up the database table entries). When I used the intermediate message event in a nested subprocess, the engine confused more. Once the engine returned a wrong process instance id, a wrong process instance ened up being processed. Then, subsequent process instances got null pointer exceptions, etc. It was NOT pretty.
I reported this problem, see http://community.jboss.org/thread/166536, that the name query ProcessInstanceWaitingForEvent used by the ksession.signalEvent didn't return with the correct process/subprocess instance which has the intermediate event type.
So, my team solved the issues as follows:
1. When we start the process, we store a unique correlation id, a session id and a process instance id (it will be the top-level process instance id) in a database table (e.g., correlation id, session id, process instance id).
2. When an intermediate message arrives, we restore the session using the incomig correlation id from the database table. Using this table, we also find the top-level process instance id.
3. Using this process id, we query the EventType database table to find the process or sub-process which has the proper event type. The EventTypes database table has two columns: instance id and element.
As you can see, the 1356 row points the 1357 row. In my opinion, the proper way to find a correct pid is drilling down the chain from the top-level pid which is getten from the startProcess. Instead of embedding the next pid part of the text data in the Element field, if we have the third column to point the next pid, we could use SQL chain query. Using the current schema, I need to get the 356 instance id, parse the element text (after the colon), then, get the row 1357 instanceid, so on....
Since we don't want to add additional column at this time, we created a function with two parameters: eventType, topLevelProcessInstanceId:
- the function queries the EventType database table using the topLevelProcessInstanceId.
- it parses the element column.
- If the element contains the same eventType, it returns the processInstanceId.
- Otherwise, it gets a sub-string after ":" (e.g., 1357). Use this sub-string (need to convert it to long) as a parameter along with the eventType (unchanged), and call the same function recursively until the element is equal to the event type.
4. After the function, we call ksession.signalEvent(eventType, eventData, foundProcessInstanceId);
As you noticed, this method may end up querying multiple times depending on the depth of sub-processes. There could be some performance penalties. But, it works reliably regardless the EventTypes database table entries.
If jBPM engine itself supports the correlation out-of-the-box, it would be great. But for now, we need to code for the correlation support.
Any suggestions would be appreciated.
Thanks Bi'm a engineering student that currently preparing the thesis and your opinion is greatly appreciated.
i have understood that the messages's correlation must be handled at application level, as in your solution.
Maybe, for avoiding, the recursion to find the correct pid you can use a ProcessEventListener that persists the processsInstanceId of any process that is waiting for something in a table defined according to your needs, but this is only a solution that I thought at the time.
thanks again, if i will work in this area i will write here my solution.
Hi community members,
I am also facing the same problem regarding the correlation key from parent process to sub-process. I have a parent process, which is using a standalone sub-process with its own process-id. The question is how to signal an event which is located in sub-process from outside?
If we use
ksession.signalEvent(eventType, eventData); // without the top-level restoredProcessInstanceId.
how process engine knows which sub-process instance to notify? I see byungwoojun provides a custom solution, storing the event type and process/sub-process instance ids in EventTypes table.
I would like to know if jBPM 6.0.0 version supports this use-case, so we do not have to do it manually?
Thanks in advance,