Random Access Queue

[Mark Fisher]

If I understand correctly, I believe that this issue should address your problem without the need for a new channel implementation: http://jira.springframework.org/browse/INT-143

The first draft of that implementation should be available by the end of today (or tomorrow at the latest). I'll post back here and would be very interested in your feedback - if that does in fact address your need.

-Mark

[amsmota]

How's the situation on this? I noticed today there are lot's of updates in the SVN. (neverthless, my channel is working ok for me...)

[Mark Fisher]

Check out the recently added ResponseCorrelator. There are still a number of possibilities for extending this (see the JIRA comments for a start: http://jira.springframework.org/browse/INT-143), but I'm interested in how this may apply to your use case.

Thanks,
Mark

[amsmota]

Hello again:

I started today to test the ResponseCorrelator and I'm having a heck of a time doing it. First, let me tell the things my RandomAccessChannel was doing and I can't accomplish with ResponseCorrelator.

- a non-blocking request-response mechanism. When I try to get a response from the queue I want to return something similar to a 404 - Not Found, so the user can keep requesting the resource for the response whenever he wants until it gets it.

- a iterator over the messages currently on hold in the queue. That way, a user can query the responseChannel without a msgId in which case he receives a list of all the msgId's currently on the queue, if working with a http connector in the form of hiperlinks to those msgs (very restfull )

Besides this "wishes", I don't even am successful in simply testing it with a asynchronous request / response test, the id's never match...

But I'm trying a little more, I was testing with the M3 from the repository, I'll try now with the version I downloaded from the trunk.

Thanks a lot for your time.

[amsmota]

So basically I can rewrite my code to use a ResponseCorrelator instead of my Response channel, but that way I cannot specify like I have

<channel id="input"/>
<channel id="responses"/>
<endpoint input-channel="input" handler-ref="serviceimpl" handler-method="execute" default-output-channel="responses"/>

or with my RandomAccessChannel

<beans:bean id="responses" class="RandomAccessChannel"/>


because I have to hand code the ResponseCorrelator that has it's own queue. Also I added a iterator to ResponseCorrelator, RetrievalBlockingMessageStore, MessageStore, SimpleMessageStore. And I miss a peek method, but I can have the same efect by looking for a null after a ResponseCorrelator.getResponse(key).

I'll keep on testing tomorrow.

[Mark Fisher]

You should be able to use the same basic configuration. Then just add a ResponseCorrelator that is subscribed to the "responses" channel. If you provide a reference to that correlator in your client code, then you can call responseCorrelator.getResponse(correlationId).

Nevertheless, what I think we need to introduce is a way to do this at a higher level so that these configuration details are not so tedious. Also, it should be very consistent with the RequestReplyTemplate usage. There are a couple JIRA issues related to this. Feel free to add comments:
http://jira.springframework.org/browse/INT-126
http://jira.springframework.org/browse/INT-143

Thanks,
Mark

[amsmota]

Well, I have added this to the RequestReplyTemplate and it works for me.

	private final ResponseCorrelator correlator;

	public RequestReplyTemplate(MessageChannel requestChannel, ResponseCorrelator correlator, ExecutorService executor) {
		Assert.notNull(requestChannel, "'requestChannel' must not be null");
		Assert.notNull(correlator, "'correlator' must not be null");
		Assert.notNull(executor, "'executor' must not be null");
		this.requestChannel = requestChannel;
		this.correlator = correlator;
		this.executor = executor;
	}

	public RequestReplyTemplate(MessageChannel requestChannel, ResponseCorrelator correlator) {
		this(requestChannel, correlator, Executors.newSingleThreadExecutor());
	}
	
	
	public void requestCorrelator(final Message<?> requestMessage) {		
		ReplyHandler replyHandler = new ReplyHandler() {
			public void handle(Message<?> replyMessage, MessageHeader originalMessageHeader) {
				correlator.handle(replyMessage);
			}
		};
		request(requestMessage, replyHandler);
		
	}

It misses some checks yet.

However, reading the comments on the Jira it's not this that you want, or at least is only partially what you want?

[amsmota]

'm afraid (and apologize for) I'm not understanding the big picture here. For example, I don't know what you mean by "subscribe". And I also can't figure out what does the <handler/> tag also.

But nevertheless, this is the situation of my tests. In my config i had:

	<channel id="input"/>
	<channel id="responses"/>

	<endpoint input-channel="input" handler-ref="serviceimpl"
		handler-method="execute" default-output-channel="responses"/>
	
	<beans:bean id="serviceimpl"
		class="com.meridianp2p.stf.service.testservice.TestService"/>
	
	<beans:bean id="resource"
		class="com.meridianp2p.stf.service.testservice.TestResource">
		<beans:property name="requestChannel" ref="input"/>
		<beans:property name="responseChannel" ref="responses"/>
	</beans:bean>

and now I have

	<channel id="input"/>
	<channel id="responses"/>

	<endpoint input-channel="input" handler-ref="serviceimpl"
		handler-method="execute" default-output-channel="responses"/>
	
	<endpoint input-channel="responses" handler-ref="correlator"/>
	
	<beans:bean id="correlator" class="org.springframework.integration.handler.ResponseCorrelator"/>
	
	<beans:bean id="serviceimpl"
		class="com.meridianp2p.stf.service.testservice.TestService"/>
	
	<beans:bean id="resource"
		class="com.meridianp2p.stf.service.testservice.TestResource">
		<beans:property name="requestChannel" ref="input"/>
		<beans:property name="responseChannel" ref="responses"/>
		<beans:property name="correlator" ref="correlator"/>
	</beans:bean>

It's working ok with the changes I've made to the ResponseCorrelator, but I don't see the point in having a response channel and a ResponseCorrelator to do only one thing.

Other thing that worries me is the impact that all this will have when I change my intra-VM queues to JMS queues...

[amsmota]

Continuing my changes, I finished the synchronous-that-falls-down-to-asynchronous-with-correlationId-when-timedout message type, which I integrated in the RequestReplyTemplate. Now probably this is getting too complex and should be separated like suggested in Jira INT-143.

I feel a little uncomfortable with this kind of configuration

<endpoint input-channel="input" handler-ref="serviceimpl" handler-method="execute" default-output-channel="responses"/>

<endpoint input-channel="responses" handler-ref="correlator"/>

but I don't know how to do it better.

Anyhow I have to move to other implementations so I only be testing this furthermore in my "spare" time. If you want me to publish the code here I can clean it up a little, but probably there are better ways to do what I did.

If I can be of assistance please let me know.

[Mark Fisher]

Thanks so much for all of the feedback.

I am thinking that the RequestReplyTemplate and ResponseCorrelator need to be encapsulated. It seems that all of this can be hidden inside a Messaging Gateway implementation (see http://eaipatterns.com/MessagingGateway.html). Something like the following would be used instead of your endpoint definitions:

<gateway id="serviceGateway"
         input-channel="input"
         handler="serviceimpl"
         method="execute"
         response-channel="responses"/>

This might internally use RequestReplyTemplate and ResponseCorrelator, but the result would be a proxy for "serviceimpl" whose bean name is "serviceGateway".

If an anonymous, temporary channel is sufficient, then the "response-channel" can be left out. Also, if the proxy should just implement MessageHandler, then the "method" should be left out.

Your idea of falling back to async after a timeout is interesting and could probably be available as one strategy used by the gateway.

Regards,
Mark