Spring Cloud Stream - @StremListener condition

Question

According to documentation: https://cloud.spring.io/spring-cloud-static/spring-cloud-stream/3.0.3.RELEASE/reference/html/spring-cloud-stream.html#_using_streamlistener_for_content_based_routing I can route the incoming message to a handler based on a condition like below:

@EnableBinding(MySink.class)
@EnableAutoConfiguration
public static class TestPojoWithAnnotatedArguments {

    @StreamListener(target = MySink.INPUT, condition = "headers['type']=='bogey'")
    public void receiveBogey(@Payload BogeyPojo bogeyPojo) {
       // handle the message
    }

    @StreamListener(target = MySink.INPUT, condition = "headers['type']=='bacall'")
    public void receiveBacall(@Payload BacallPojo bacallPojo) {
       // handle the message
    }

    @StreamListener(target = MySink.ANOTHER_INPUT, condition = "headers['type']=='bacall'")
    public void receiveBacall(@Payload BacallPojo bacallPojo) {
       // handle the message
    }
}

How do I provide a handler that's called when none of the conditions match?

If I have 2 handlers, first one with a condition and second one without any, both the handlers are called when the first one's condition matches. How do i avoid this?

It looks like this is not possible. The handleRequestMessage(..) delivers to all matching handlers and has no default handler concept judging from the spring-cloud-stream source code: github.com/spring-cloud/spring-cloud-stream/blob/master/… — Jerald Baker
Please don' t ask variation of the same question in two different places - stackoverflow.com/questions/61188379/…. It is taking away the time of people who contribute and confuses the readers when answers come. — Oleg Zhurakousky
Also, keep in mind that annotation-based model is all but deprecated. Please go through these 3 post to better understand the direction we're going. The last post also provides more details on content-based routing. spring.io/blog/2019/10/14/…, spring.io/blog/2019/10/17/…, spring.io/blog/2019/10/31/spring-cloud-stream-event-routing — Oleg Zhurakousky
I went through the posts. I have multiple sources (similar to what i would have with target = Sink.INPUT, target = Sink.INPUT2, and so on), and each of those sources produce multiple type of messages. So, i want to be able to route based on some condition, specific to each source. I see only a generic property "spring.cloud.function.routing-expression" which will be applied to all sources. Also, what about the scenario of default handler? It would be really great if you could point me to an example/test, as I am unable to find any. — Jerald Baker
I have modified my question with the code based on the multiple input types per source binding. — Jerald Baker

Oleg Zhurakousky Oleg Zhurakousky · Accepted Answer · 2020-04-13T16:47:14

We probably need to modify the section you're referring to as it is somewhat outdated.

Also, we can not (should not) do any kind of routing based on the payload type, since the data comes in from the wire in the serialised form such as byte[]. I discuss it in details in this old post. But you can definitely use other parts of the incoming Message as routing condition. The recommended best practice is to rely on Message Headers.

So let's look at the sample:

@Bean
public Function<String, String> uppercase() {
    return v -> v.toUpperCase();
}

@Bean
public Function<String, String> lowercase() {
    return v -> v.toLowerCase();
}

@Bean
public Function<String, String> reverse() {
    return v -> new StringBuilder(v).reverse().toString();
}

. . .and indeed a single routing-expression property. You only need one expression, since however complex or simple your condition is it can be encoded with standard Spring SpEL

--spring.cloud.function.routing-expression=headers['type'] == 'upper' ? 'uppercase' : (headers['type'] == 'lower' ? 'lowercase' : ''reverse)

What will happen is, the incoming Message's header with the name type will be evaluated. And if its value is 'upper' it will go to 'uppercase' function; if 'lower' to 'lowercase' function and default to 'reverse'.

Hope that helps.

Spring Cloud Stream - @StremListener condition

1 Answers