Using Spring Cloud Stream Source to send method results to stream

Question

I'm trying to create a Spring Cloud Stream Source Bean inside a Spring Boot Application that simply sends the results of a method to a stream (underlying Kafka topic is bound to the stream).

Most of the Stream samples I've seen use @InboundChannelAdapter annotation to send data to the stream using a poller. But I don't want to use a poller. I've tried setting the poller to an empty array but the other problem is that when using @InboundChannelAdapter you are unable to have any method parameters.

The overall concept of what I am trying to do is read from an inbound stream. Do some async processing, then post the result to an outbound stream. So using a processor doesn't seem to be an option either. I am using @StreamListener with a Sink channel to read the inbound stream and that works.

Here is some code i've been trying but this doesn't work at all. I was hoping it would be this simple because my Sink was but maybe it isn't. Looking for someone to point me to an example of a source that isn't a Processor (i.e. doesn't require listening on an inbound channel) and doesn't use @InboundChannelAdapter or to give me some design tips to accomplish what I need to do in a different way. Thanks!

@EnableBinding(Source.class)
public class JobForwarder {

   @ServiceActivator(outputChannel = Source.OUTPUT)
   @SendTo(Source.OUTPUT)
   public String forwardJob(String message) {
       log.info(String.format("Forwarding a job message [%s] to queue [%s]", message, Source.OUTPUT));
       return message;
   }
}

Answer 1

Your orginal requirement can be achieved through the below steps.

1. Create your custom Bound Interface (you can use the default @EnableBinding(Source.class) as well)

   public interface CustomSource {
       String OUTPUT = "customoutput";
   
       @Output(CustomSource.OUTPUT)
       MessageChannel output();
   }
   

2. Inject your bound channel

   @Component
   @EnableBinding(CustomSource.class)
   public class CustomOutputEventSource {
   
       @Autowired
       private CustomSource customSource;
   
       public void sendMessage(String message) {
           customSource.output().send(MessageBuilder.withPayload(message).build());
       }
   }
   

3. Test it

   @RunWith(SpringRunner.class)
   @SpringBootTest
   public class CustomOutputEventSourceTest {
   
       @Autowired
       CustomOutputEventSource output;
   
       @Test
       public void sendMessage() {
           output.sendMessage("Test message from JUnit test");
       }
   }
   

Answer 2

So if you don't want to use a Poller, what causes the forwardJob() method to be called?

You can't just call the method and expect the result to go to the output channel.

With your current configuration, you need an inputChannel on the service containing your inbound message (and something to send a message to that channel). It doesn't have to be bound to a transport; it can be a simple MessageChannel @Bean.

Or, you could use a @Publisher to publish the result of the method invocation (as well as being returned to the caller) - docs here.

@Publisher(channel = Source.OUTPUT)

Answer 3

Thanks for the input. It took me a while to get back to the problem. I did try reading the documentation for @Publisher. It looked to be exactly what I needed but I just couldn't get the proper beans initialized to get it wired properly.

To answer your question the forwardJob() method is called after some async processing of the input.

Eventually I just implemented using spring-kafka library directly and that was much more explicit and felt easier to get going. I think we are going to stick to kafka as the only channel binding so I think we'll stick with that library.

However, we did eventually get the spring-cloud-stream library working quite simply. Here was the code for a single source without a poller.

@Component
@EnableBinding(Source.class)
public class JobForwarder {

    private Source source;

    @Autowired
    public ScheduledJobForwarder(Source source) {
        this.source = source;
    }

    public void forwardScheduledJob(String message) {
        log.info(String.format("Forwarding a job message [%s] to queue [%s]", message, Source.OUTPUT));
        source.output().send(MessageBuilder.withPayload(message).build());
    }
}