Sample code for request/reply design by using Spring Cloud Stream

vadivelan

Hi, I am new to solace and my application is using solace request/reply by using old approach.

I have planned to use spring cloud stream in my project for the below scenarios.

please refer the below use case and help me with samples:

I have three microservice services in my application

ServiceA -- > Request the message to ServiceB processing and waiting for the response.
ServiceB --> Read the payload from topic/queue and apply the some business rule validation.
ServiceB is publish the message to ServiceC if business validation is success else ServiceB is reply the error message to ServiceA
ServiceC --> ServiceC is read the message and apply some business process then send the final response to ServiceA.

marc

@giri is correct, request-reply isn't explicitly supported in Spring Cloud Stream but you can kind of roll your own.

I haven't yet done it myself, but it should work if you do this (I don't think I'm missing anything..):

• Use the solace_correlationId and solace_replyTo headers
• Have you requestor pre-define a replyTo topic space and subscribe to it..for example: domain/app/app-instance/reply/> (Follow topic best practices that you can find elsewhere)
• Set the solace_correlationId to a unique value and the solace_replyToin the msg header to something like domain/app/app-instance/reply/<correlation-id>
• Have the replier respond to the solace_replyTo topic using the "dynamic publishing" capabilities of spring cloud stream which you can learn more about in this codelab. I'd probably use the BindersHeaders.TARGET_DESTINATION option since each response would be on a unique topic...no reason to have spring cache the channel info.

Hope that helps!

marc

I decided to try this out and wrote a requestor/replier in this repo: https://github.com/Mrc0113/cloud-stream-request-reply-solace

Feel free to check it out. Note you'd definitely want something in the requestor that probably runs on a schedule to check for outstanding requests that need to be resent or something like that.

giri

Hi @vadivelan, your requirement fits the classic request-response messaging pattern.

Synchronous request-response may not fit in the scheme of SCS as it is more geared for asynchronous interactions.

However, request-response can be implemented asynchronously, with a response being returned at some unknown later time but the request, response, and any other intermediate messages all carry the same correlation-id. This is something that can be controlled in the code. This is a “sync over async”, or “sync/async” pattern.

giri

And of course, you can use the solace header property 'solace_replyTo' to control message routing. Please check out https://github.com/SolaceProducts/solace-spring-cloud/tree/master/solace-spring-cloud-starters/solace-spring-cloud-stream-starter for more information.

marc

@giri is correct, request-reply isn't explicitly supported in Spring Cloud Stream but you can kind of roll your own.

I haven't yet done it myself, but it should work if you do this (I don't think I'm missing anything..):

• Use the solace_correlationId and solace_replyTo headers
• Have you requestor pre-define a replyTo topic space and subscribe to it..for example: domain/app/app-instance/reply/> (Follow topic best practices that you can find elsewhere)
• Set the solace_correlationId to a unique value and the solace_replyToin the msg header to something like domain/app/app-instance/reply/<correlation-id>
• Have the replier respond to the solace_replyTo topic using the "dynamic publishing" capabilities of spring cloud stream which you can learn more about in this codelab. I'd probably use the BindersHeaders.TARGET_DESTINATION option since each response would be on a unique topic...no reason to have spring cache the channel info.

Hope that helps!

marc

I decided to try this out and wrote a requestor/replier in this repo: https://github.com/Mrc0113/cloud-stream-request-reply-solace

Feel free to check it out. Note you'd definitely want something in the requestor that probably runs on a schedule to check for outstanding requests that need to be resent or something like that.

as63010

Thanks Team for your support. Please clarify BinderHeaders.TARGET_DESTINATION is support only topic name or binder name also same like StreamBridge as we need to support multi VPN in our scenario.

marc

Hi @as63010,

BinderHeaders.TARGET_DESTINATION is just to specify the topic that an individual message is sent to. That message is then handed to the binder that sends to that topic. If you want to publish to multiple binders then you could either use StreamBridge or you could have different send methods that each have their own bindings configured since each binding allows you to configure a different binder. Depending on how many binders you need I think the latter sounds like a nice option b/c I believe that would result in the framework verifying connectivity of each binding on startup. I haven't seen anyone do this yet so please share if you get this working and which solution you end up trying

as63010

@marc ,

We able to send the message to different binder/VPN using streambridge with bindername and it is working. Please review below and share your feedback.

bindings:
initializer|validatorAndEnrichment|processor-in-0:
destination: TEMPS.Q
binder: atomic
group: AtomicProcessor
consumer:
concurrency: 5
atomicService1External-out-0:
destination: post/atomicservice1/response/external
binder: external
atomicService1Internal-out-0:
destination: post/atomicservice1/response/internal
binder: internal

        log.info("processor: " + msg.getPayload() +  msg.getHeaders());
        String bindingName = (String)msg.getHeaders().get("ReplyToBindingName");  // **atomicService1External-out-0**
        String payload = msg.getPayload().concat(" Added in sendRequest ");
        Message responseMsg = MessageBuilder.withPayload(payload).copyHeaders(msg.getHeaders()).build();
        streamBridge.send(bindingName, responseMsg);

marc

Hi @as63010,

Great, glad it's working for you! I think this pattern works pretty well but you'll just want to be cautious that the event you receive always has a "ReplyToBindingName" that will match the config of your app. It feels a bit too coupled for my liking....I think I'd go a different route.

There aren't many examples out there as it's a new enhancement in 2021, but StreamBridge also offers a method where you can specify the binder you want to send to. Check it out here. If you were to go this route then your inbound messages can just specify a topic to reply on (which could be dynamic and allow you to handle an infinite amount of them) and then your app just needs to have some logic to know what binder to send to. Make sense?

as63010

Thanks @marc. This enhancement on spring cloud stream is already released? We will use new method and get the topic name from inbound message and pass and internally we get the bindername as you suggested.

public boolean send(String bindingName, @Nullable String binderName, Object data)

marc

Yep! That code is in the main branch which is the latest version

as63010

Thanks @marc . We have used older version earlier updated to latest version and works fine.

marc

Excellent, thanks for the update @as63010

MartinL

Hi Marc,

I tried your solution and it also works fine for me. But I changed it a bit and send now a REST request to this dynamic topic, but the resonse is coming to your cloud-stream-replier. Do you have me an idea how I can get the response back to my REST request? I am trying it now since many hours without success. I think I am getting crazy soon

marc

Hi @MartinL,

Just to clarify are you trying to have a REST app make a request to the broker and then have a Spring Cloud Stream app handle that request, process it and reply?

If yes, then you'll want to verify a few things:
1. Make sure your VPN is in Microgateway mode (this way your REST app will wait for a response)
2. Make sure your Spring Cloud Stream app is replying to the topic specified in the "solace_replyTo" header AND that it is setting the "solace_correlationId" to match the one incoming on the request. Both of those items would be required for the microgateway to properly pass the response back to the requestor.

Hope that helps!

marc

@MartinL I'll also add that @JamilAhmed implemented a use case that sounds similar to yours (REST request in via MicroGateway and request handled in the backend by Cloud Stream apps) in this repo: https://github.com/itsJamilAhmed/scs-credit-card-demo

That might be of interest to check out

MartinL

Hello Marc,

your thoughts a message before were right. That's what I already did (2. solace_replyTo and solace_correlationId) and then I tried to get it on the backend with a function and a consumer, but I had no success to get the response from the function that should call a SOAP service. But now I recognized that you do it already with the Microgateway. I tried to configure it this way - a link I received from @Hong - and it worked great. But I am asking myself, if I loose the normal messaging possibilities with Cloud Streams with the Microgateway setting. Do I have to install then something else? If my backend webservice is behind a firewall, does it make sence to opend special ports also or is it enough to open the port for the backend service. I will also try your second solution as soon as possible. Thanks a lot for your help!!!

marc

Hi @MartinL,

Do I have to install then something else?

Nope, one of the benefits of our broker is you don't need to install plugins

If my backend webservice is behind a firewall, does it make sence to opend special ports also or is it enough to open the port for the backend service.

This really depends on your security constraints and what already has access to call the service via the microgateway on the broker. You'll likely wnat some security in place. I'd recommend checking out our Security Overview docs. Especially the Authorization section which covers how you can authorize who can connect and what resources (topics/queues) they have access to.

Hope that helps!

MartinL

Hello Marc,
I hope this is my last question, sorry! I did all this way it worked, when I was at home, but not behind my firewall etc. What I found out is that everything seems to be enable, but the operation status is Down. I made two screenshot. But it changed for example, if I did change the url in the rc_get to "localhost" and suddenly it was up. Is it possible that your broker tries to connect to this url and can n't reach it, because the ports are not already open to reach it???

MartinL

@marc Do n't you have a tipp why this operational state is always down. Opened now ports, all traffic is running just the broker hangs in this operational state. Changed it also to POST but the same effect.

MartinL

@marc Solved the problem thanks to your tipp here https://solace.community/discussion/comment/768#Comment_768 with checking the details. the url could not be resolved. Changed it to ip and it work

marc

Hi @MartinL, thanks for the update and glad you figured it out! Sorry I wasn't able to get back to you sooner.