SpringCloudStream - how to create delayed exchange?

Question

I want to create delayed exchange from rabbitMQ plugin.

At first I tried to use spring binders as I can see, that delayed-exchange flag is available:

spring:
  cloud:
    stream:              
      bindings:
        delayed-ex:
          group: update-delayed
          contentType: application/json
          consumer:
            max-attempts: 1
      rabbit:
        bindings:
          delayed-ex:
            consumer:
              transacted: true
              auto-bind-dlq: true
              republish-to-dlq: true
              delayed-exchange: true
              requeue-rejected: false

Unfortunately, this config is creating me simple topic exchange without any flags.

I can create bean with:

@Bean
public CustomExchange delayedExchange() {
    final Map<String, Object> args = new HashMap<>();
    args.put("x-delayed-type", "topic");     
    return new CustomExchange("delayed-ex", "x-delayed-message", true, false, args);
}

And it is working with current config, but it is giving me stacktraces about existing exchanges which I want to override:

Caused by: com.rabbitmq.client.ShutdownSignalException: channel error; protocol method: #method(reply-code=406, reply-text=PRECONDITION_FAILED - inequivalent arg 'type' for exchange 'delayed-ex' in vhost '/': received 'topic' but current is ''x-delayed-message'', class-id=40, method-id=10)

Another thing (don't know if it is a problem) I want to have consumer and producer in the same service, just for having buffer for my messages.

Also, I checked source of delayed-exchange flag, and saw that this flag is doing absolutely nothing (Am I wrong?), because isDelayed methods from AbstractExchange is not called.

Is it possible to declare delayed exchange with type x-delayed-message and do it by configuration, instead of declaring bean? Also I want to remove that stacktraces

ByeBye ByeBye · Accepted Answer · 2017-09-04T08:24:57

The problem was with wrong version of spring-rabbit dependency. I used 1.5.6.RELEASE which do not use isDelayed() method. Changing it into 1.7.3.RELEASE version fix the problem.

I deleted my custom bean and create this config:

spring  
  cloud:
    stream:              
      bindings:
        delayed-ex:
          group: update-delayed
          contentType: application/json
          consumer:
            max-attempts: 1
      rabbit:
        bindings:
          delayed-ex:
            consumer:
              transacted: true
              auto-bind-dlq: true
              republish-to-dlq: true
              delayed-exchange: true
              requeue-rejected: false     
            producer:
              delayed-exchange: true

Now, exchange and queues are created automatically by config without any stacktraces.

SpringCloudStream - how to create delayed exchange?

1 Answers