How to add a field from body to a condition

I use the @StreamListener annotation to listen to the topic. How can I add a field from body to the condition? An example of an entity that is sent to the topic:

public class Event {

    private String eventId = UUID.randomUUID().toString();
    private LocalDateTime eventTime = LocalDateTime.now();
    private Entity entity;
}

public class Entity {

    private String id = UUID.randomUUID().toString();
    private String name;
}

Answer

As you can see from version 3.0, you should avoid using filtering based on the message payload. Notice these lines from the documentation:

The preceding code is perfectly valid. It compiles and deploys without any issues, yet it never produces the result you expect.

That is because you are testing something that does not yet exist in a state you expect. That is because the payload of the message is not yet converted from the wire format (byte[]) to the desired type

So, unless you use a SPeL expression that evaluates raw data (for example, the value of the first byte in the byte array), use message header-based expressions (such as condition = “headers[‘type’]==’dog'”).

So, for your case, you could add one more header into your messages and filter by using conditions on your header, e.g:

Message<Event> message = MessageBuilder.withPayload(event)
                .setHeader(KafkaHeaders.MESSAGE_KEY, event.getRequestId().toString().getBytes(StandardCharsets.UTF_8))
                .setHeader("entityName", event.getEnity().getName().getBytes(StandardCharsets.UTF_8))
                .build();
this.streamBridge.send("binding", message);

And now your condition would be condition = "headers['entityName']=='ABCName'"

Notice: Annotation-based programming model. Basically the @EnableBinding, @StreamListener and all related annotations are now deprecated from version 3.x