Help needed with integration scenario

[gekkothelizard]

Hi! I have ran into a situation which I feel is a quite common case but I can't figure out how to implement elegantly using SEI. Fortunately it's not very complicated so I'll try describe it here, and would be very grateful for help from people with more experience with the framework.

I have a Dossier domain entity that has an Application representing a loan application to a bank. The Application has several Applicants. I want to do the following: grab the Dossier from the DossierService, and run a BRE check on each applicant checking if they are older the 18 (keep in mind that this is just an exercise, it's of course too simple a task to really use a BRE for), then submit the Dossier back to the DossierService. I already implemented this by implementing the iteration over all Applicants in the BRE adapter.
Now I want to modify the implementation to use a Splitter for splitting the Applicants into multiple messages so the BRE adapter will only handle one Applicant. My message chain looks the following now (this is not working yet, so please don't mind errors in there, I mainly put it here to summarize what I'm trying to do):

<int:chain input-channel="businessRuleChannel" output-channel="dossierServiceChannel">
		<int:transformer expression="payload.getApplication().getApplicants().getApplicantList()"/>
		<int:splitter apply-sequence="true"/>
		<int:service-activator method="checkAgeOfApplicant">
			<bean class="nl.vwpfs.pos.esb.BusinessRuleEngine"/>
		</int:service-activator>
		<int:aggregator>
			<bean class="nl.vwpfs.pos.esb.ApplicantAggregator"/>
		</int:aggregator>
</int:chain>

The problem I ran into is that the aggregator will produce a List<Applicant> in the end but I what I need is an updated Dossier object. As I see it my Dossier object was lost when the Splitter transformed my message which had a Dossier payload into a set of messages with Applicant payloads. Is there a way to "save" my original message, so I can use the Dossier in the end. I definetaly want to output the whole (updated) Dossier from the chain.

One more note: In this simple case where everything is an in-memory object the Dossier will actually be updated because it still holds reference to the Applicant objects. But If I would convert this to a more realistic scenario then this would not be the case, so I want to simulate that update operation.

[Amol Nayak]

The problem I ran into is that the aggregator will produce a List<Applicant> in the end but I what I need is an updated Dossier object

What does your class nl.vwpfs.pos.esb.ApplicantAggregator do? What is the return type of you aggregating method? Behavior you are referring to as a List of Applicant is the default one when no aggregating bean is specified by you (DefaultAggregatingMessageGroupProcessor does that work).

[gekkothelizard]

I want the ApplicantAggregator to take a list of Applicants as an input parameter and return the original Dossier updated with those applicants. The problem is I can't get the Dossier instance in the ApplicantAggregator, because it is not contained in the messages being aggregated. Here is the code of the aggregator:

public class ApplicantAggregator {
	
	@Aggregator
	public Dossier aggregateApplicants(List<Applicant> applicants) {
		Application.Applicants applicantWrapper = new Application.Applicants();
		applicantWrapper.getApplicantList().addAll(applicants);

                //this is the dossier instance I need to get access to
		dossier.getApplication().setApplicants(applicantWrapper);
		return dossier;
	}
}

Note that I can't create a new Dossier here, because the Dossier contains a lot of other stuff besides the applicants, so I need to update the same Dossier that I split previously with the splitter in the message chain.

[Amol Nayak]

I haven't tried this out myself (will do a small sample now though). But i am thinking of using a claim check for this purpose.
This is what i feel can address the problem

1. You have a service activator at the beginning of the chain that will send the message to a claim check in and will wait till a response with the claim check (the original message's id) is returned. Service activator's method return the payload as is from the invoked method. The claim check in will store the payload in the message store and will generate the message with the key, which is the same as messages id. We will actually not get hold of value and can ignore it as it would be same as the id of the message we will be splitting.

2. The flow will continue as is through the splitter, BRE and then will reach the aggregator.

3. Here in the aggregating bean we will need the get the correlation id from the messages header as a parameter to the bean's aggregate method along with the List of payload, this value of correlation id will be same as the id of the original message.

4. We use this id to send a message to the claim check out and get the message with original payload on the reply channel/output channel.

5. Now once you have dossier object and you already have the List of Applicants as the parameter to your aggregate method, you can update the dossier and return it.

Note: We may not actually directly interact with the claim check in and out's input channels but will have a gateway that in turn will do the job for us.

Let me know you views and i'll also try preparing a small sample and post that here later to demonstrate the concept (if it works fine )

[gekkothelizard]

Thanks for your answer! I definitely see that the <claim-check-in> is the component I need, but I don't understand something:

The <claim-check-in> will store my dossier but it will send a new message to the output channel with only a UUID as the payload. How will I get the Dossier to the Splitter if I already hid it in the <claim-check-in>. As for the aggregator at the end I wouldn't mind if it had a direct reference to the message store, but I need the Dossier for input to the splitter.
I suspect that this is what you are solving with the <service-activator> in point 1, but I don't understand how it would work. Can you clarify it for me bit? A sample xml would be very helpful - it doesn't have to actually work. Thanks!

[Gary Russell]

Copy the applicants to a header before claim check in; move the claim check id to a header after checkin and move the applicants back to the payload.

I don't think I'd use a claim check here though; it might be simpler to just send a copy of the dossier to the aggregator - move the aggregator out of the chain; make the businessRuleChannel a <publish-subscribe /> channel and <bridge /> it to the aggregator. Use a custom release strategy on the aggregator to wait for the dossier as well as the applicants.

[gekkothelizard]

Hello Guys! Sorry for not responding for such a long time, I had a lot on my hands lately. Among others I just completed the SEI certification, so I'm armed with new knowledge
I managed to solve the problem using the help you gave me and some additional learning. Here is my solution: I used a publish-subscribe channel and two aggregators. The PS channel sends the Dossier to both the businessRuleChannel and straight to the end of the processing where the second aggregator will store it until the subresults arrive. The chain handling the business rule engine invocation also contains an aggregator, to gather the Applicants into a list. So the second aggregator is only aggregating two object: the Dossier and the List of Applicants. You can see the code below. Please write if you have any suggestions how this implementation could be improved. The important requirement is that the administration around the BRE call should be self-contained so more such steps can be added to the flow easily.

PS. I think that the claim check in/out pattern is really suited for this situation, but I really can't figure out how to use it in practice. I either loose the real Dossier before I would split it into Applicants or I loose the id of the stored dossier. Strangely I also couldn't find a single example of this pattern in practice.

<beans ...>
	<int:publish-subscribe-channel id="testProcessingOperationChannel" datatype="nl.vwpfs.pos.xmlcdm.dossier.Dossier"/>

	<int:bridge input-channel="testProcessingOperationChannel" output-channel="businessRuleChannel"/>
	<int:bridge input-channel="testProcessingOperationChannel" output-channel="dossierReassemblyChannel"/>

	<int:channel id="businessRuleChannel" datatype="nl.vwpfs.pos.xmlcdm.dossier.Dossier"/>
	<int:chain input-channel="businessRuleChannel" output-channel="dossierReassemblyChannel">
		<int:header-enricher>
			<int:header name="DossierID" expression="payload.getUid()"/>
		</int:header-enricher>
		<int:splitter apply-sequence="true" expression="payload.getApplication().getApplicants().getApplicantList()"/>
		<int:service-activator method="checkAgeOfApplicant">
			<bean class="nl.vwpfs.pos.esb.BusinessRuleEngine"/>
		</int:service-activator>
		<int:aggregator/>
	</int:chain>

	<int:channel id="dossierReassemblyChannel"/>
	<int:aggregator
			input-channel="dossierReassemblyChannel" output-channel="dossierServiceChannel"

			ref="dossierReassembler"
			method="processMessageGroup"

			correlation-strategy="dossierReassembler"
			release-strategy="dossierReassembler">
	</int:aggregator>


	<int:channel id="dossierServiceChannel" datatype="nl.vwpfs.pos.xmlcdm.dossier.Dossier"/>
	<int:service-activator
			id="dossierServiceServiceActivator"
			input-channel="dossierServiceChannel"
			ref="dossierService"
			method="update"
			output-channel="nullChannel"/>

	<context:annotation-config/>

	<import resource="beanDefinitionContext.xml"/>

</beans>

And the dossier reassebling aggregator:

class DossierReassembler extends AbstractAggregatingMessageGroupProcessor implements CorrelationStrategy, ReleaseStrategy {

	@Override
	Object getCorrelationKey(Message<?> message) {
		if (message.getPayload() instanceof Dossier) {
			return ((Dossier) message.getPayload()).getUid()
		} else {
			if (!message.headers["DossierID"]) throw new IllegalArgumentException("DossierID not present in header of message with payload type " + message.payload.class)
			return message.headers["DossierID"]
		}
	}

	@Override
	boolean canRelease(MessageGroup group) {
		//group should contain one list of applicants and one dossier
		boolean applicantListArrived = group.marked.find { Message msg -> msg.payload instanceof List<Applicant> } != null
		boolean dossierListArrived = group.marked.find { Message msg -> msg.payload instanceof Dossier } != null
		return applicantListArrived && dossierListArrived
	}


	@Override
	protected Object aggregatePayloads(MessageGroup group, Map<String, Object> defaultHeaders) {
		List<Applicant> applicantList = group.marked.find { Message msg -> msg.payload instanceof List<Applicant> }.payload as List<Applicant>
		Dossier dossier = group.marked.find { Message msg -> msg.payload instanceof Dossier }.payload as Dossier
		dossier.application.applicants.applicantList.clear()
		dossier.application.applicants.applicantList.addAll(applicantList)
		return dossier
	}
}

[Gary Russell]

I think what you have is cleaner, but you could do it with a claim check and without any custom components (aside from your agechecker) using something like this...

<int:chain input-channel="chan1" output-channel="chan2">
	<int:header-enricher>
		<int:header name="saveApplicants" expression="payload.getApplication().getApplicants().getApplicantList()" />
	</int:header-enricher>
	<int:claim-check-in message-store="messageStore" />
	<int:header-enricher>
		<int:header name="claimCheck" expression="payload" />
	</int:header-enricher>
	<int:transformer expression="headers['saveApplicants']" />
	<int:header-filter header-names="saveApplicants" />
	<int:splitter />
	<int:service-activator ref="ageChecker" />
	<int:aggregator />
	<int:header-enricher>
		<int:header name="saveApplicants" expression="payload" />
	</int:header-enricher>
	<int:transformer expression="headers['claimCheck']" />
	<int:claim-check-out />
	<int:header-filter header-names="claimCheck" />
	<int:transformer>
 	    <int-groovy:script><![CDATA[
			payload.application.applicants.applicantList.clear()
			payload.application.applicants.applicantList.addAll(headers['saveApplicants'])
			payload			
	 ]]></int-groovy:script>
	</int:transformer>
	<int:header-filter header-names="saveApplicants" />
</int:chain>

[gekkothelizard]

Thanks Gary!

I was thinking about you suggestion to switch the payload with the headers, but didn't like it, because it requires a lot of basically boilerplate code. Doing all that switching really feels like a workaround. An interesting question is I think, how can the claim check in/out components be used without doing such workarounds. I don't necessarily mean this particular scenario. I have the feeling that the facet that the claim check in replaces the payload will be an issue in any scenario.