Supplier binding is not working with spring cloud stream rabbit

Question

We have a source like follows and we are using spring cloud stream rabbit binder 3.0.1.RELEASE.

@Component
public class Handlers {

  private EmitterProcessor<String> sourceGenerator = EmitterProcessor.create();

  public void emitData(String str){
    sourceGenerator.onNext(str);
  }

  @Bean
  public Supplier<Flux<String>> generate() {
    return () -> sourceGenerator;
  }

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


}

application.yml

spring:
  profiles: dev
  cloud:
    stream:
      function:
        definition: generate;process
        bindings:
          generate-out-0: source1
          process-in-0: source1
          process-out-0: processed

        bindingServiceProperties:
          defaultBinder: local_rabbit

      binders:
        local_rabbit:
          type: rabbit
          environment:
            spring:
              rabbitmq:
                host: localhost
                port: 5672
                username: guest
                password: guest
                virtual-host: /

While calling emitData method, we are not seeing data in RabbitMQ queue. We also observed that consumer binding is working. That we checked by means of direct sending of messages into a consumer linked queue through RabbitMQ Admin. But supplier binding is not working.

Also, we observed that Supplier without Flux is working fine with the same application.yml configuration. Are we missing any configuration here?

Even test case with TestChannelBinderConfiguration is working fine as follows.

@Slf4j
@TestPropertySource(
        properties = {"spring.cloud.function.definition = generate|process"}
)
public class HandlersTest extends AbstractTest {
  @Autowired
  private OutputDestination outputDestination;

  @Test
  public void testGeneratorAndProcessor() {
      final String testStr = "test"; 
      handlers.emitData(testStr);

      Object eventObj;
      final Message<byte[]> message = outputDestination.receive(1000);

      assertNotNull(message, "processing timeout");
      eventObj = message.getPayload();

      assertEquals(new String((byte[]) eventObj), testStr.toUpperCase());
  }
}

How do you invoke emitData? To be specific, how do you get reference to Handlers? — Oleg Zhurakousky
Handlers are being used like normal autowired bean from other class to call `emitdata' method. — Mayank Rupareliya

Oleg Zhurakousky Oleg Zhurakousky · Accepted Answer · 2020-01-08T14:58:07

When you say we are not seeing data in RabbitMQ queue. . .. Which queue are you talking about? When using AMQP, messages are sent to exchanges and if such exchange is not bound to any queue the message is dropped, hence my question. Did you actually bind generate-out-0 exchange to a queue?

In any event, I just tested it and all is working as expected. Here is the complete code.

@SpringBootApplication
public class SimpleStreamApplication {

    public static void main(String[] args) throws Exception {
        ApplicationContext context = SpringApplication.run(SimpleStreamApplication.class);
        SimpleStreamApplication app = context.getBean(SimpleStreamApplication.class);
        app.emitData("Hello");
    }

    private EmitterProcessor<String> sourceGenerator = EmitterProcessor.create();

    public void emitData(String str) {
        sourceGenerator.onNext(str);
    }

    @Bean
    public Supplier<Flux<String>> generate() {
        return () -> sourceGenerator;
    }
}

Supplier binding is not working with spring cloud stream rabbit

3 Answers