Qfyilyi

What I am trying to do is, When I receive a message at consumer, I filter that message based on there content the filter will send messages to 3 different queues with the help of three different producers.

this is the configuration for three queues :

  cloud:

    stream:

      bindings:

        QuickScore:

          concurrency: 5

          destination: quickScore

          content-type: application/json

          group: quickScoreGroup

          maxConcurrency: 10

          recoveryInterval: 10000

        SuitabilityScore:

          concurrency: 5

          destination: suitabilityScore

          content-type: application/json

          group: suitabilityScore

          maxConcurrency: 10

          recoveryInterval: 10000

        CompletenessScore:

          concurrency: 5

          destination: completenessScore

          content-type: application/json

          group: completenessScore

          maxConcurrency: 10

          recoveryInterval: 10000



  rabbitmq:

    host: ${rabbitmq.host:localhost}

    username: guest

    password: guest

    port: 5672

Custom channel for queues

public interface CustomChannels {



  @Output("QuickScore")

  MessageChannel publishMessageToQuickScore();



  @Input("QuickScore")

  SubscribableChannel receivedAtQuickScore();



  @Output("CompletenessScore")

  MessageChannel publishMessageToCompletenessScore();



  @Input("CompletenessScore")

  SubscribableChannel receivedAtCompletenessScore();



  @Output("SuitabilityScore")

  MessageChannel publishMessageToSuitabilityScore();



  @Input("SuitabilityScore")

  SubscribableChannel receivedAtSuitabilityScore();

}

producers for queues :

@Autowired

  private CustomChannels customChannels;



  public void sendToQuickScore(UpdatedFieldsEntity updatedFieldsEntity) {

    customChannels

        .publishMessageToQuickScore().send(MessageBuilder.withPayload(updatedFieldsEntity).build());

    log.info("sending to Quick score" + updatedFieldsEntity.toString());

  }



  public void sendToCompletenessScore(UpdatedFieldsEntity updatedFieldsEntity) {

    customChannels

        .publishMessageToCompletenessScore()

        .send(MessageBuilder.withPayload(updatedFieldsEntity).build());

    log.info("sending to completeness score" + updatedFieldsEntity.toString());

  }



  public void sendToSuitabilityScore(UpdatedFieldsEntity updatedFieldsEntity) {

    customChannels

        .publishMessageToSuitabilityScore()

        .send(MessageBuilder.withPayload(updatedFieldsEntity).build());

    log.info("sending to suitability score" + updatedFieldsEntity.toString());

  }

}

And this is how I am filtering and publishing in different queues :

 @Autowired

  private EventProducer producer;



  public UpdatedFieldsEntity CheckUpdatedKey(UpdatedFieldsEntity updatedFieldsEntity)

      throws RezoomexException {

    logger.info("nn Checking UpdateKeys " + updatedFieldsEntity.toString());

    if (updatedFieldsEntity == null) {

      RezoomexException exception = new RezoomexException("update message is null");

      throw exception;

    }

    for (UpdatedFields updatedFields : updatedFieldsEntity.getUpdatedFields()) {

      UpdateKey element = updatedFields.getUpdateKey();

      if (element.toString().equals(TECHNOLOGIES_UNDER_SKILLSET) || element.toString()

          .equals(TOTAL_EXPERIENCE_VALUE)

          || element.toString().equals(TECHNOLOGIES) || element.toString()

          .equals(TOTAL_EXPERIENCE_SUFFIX)) {

        IsThreeScores = true;

      }



    }

    if (IsThreeScores == true) {

      logger.info("nnn publishing message to all Q");

      producer.sendToQuickScore(updatedFieldsEntity);

      producer.sendToSuitabilityScore(updatedFieldsEntity);

      producer.sendToCompletenessScore(updatedFieldsEntity);

      IsThreeScores = false;

    } else {

      logger.info("nnn publishing message to 2 Q");

      producer.sendToSuitabilityScore(updatedFieldsEntity);

      producer.sendToCompletenessScore(updatedFieldsEntity);

    }

    return updatedFieldsEntity;

  }

}

The very first time all queues consumes the message , but at second time any of the three queues throws exception as :

    org.springframework.amqp.rabbit.listener.exception.ListenerExecutionFailedException: Listener threw exception



Caused by: org.springframework.integration.MessageDispatchingException: Dispatcher has no subscribers

    at org.springframework.integration.dispatcher.UnicastingDispatcher.doDispatch(UnicastingDispatcher.java:154) ~[spring-integration-core-4.3.14.RELEASE.jar!/:4.3.14.RELEASE]

    at org.springframework.integration.dispatcher.UnicastingDispatcher.dispatch(UnicastingDispatcher.java:121) ~[spring-integration-core-4.3.14.RELEASE.jar!/:4.3.14.RELEASE]

    at org.springframework.integration.channel.AbstractSubscribableChannel.doSend(AbstractSubscribableChannel.java:89) ~[spring-integration-core-4.3.14.RELEASE.jar!/:4.3.14.RELEASE]

    ... 43 common frames omitted

The problem is because you are using same channel for Input and Output (i.e consuming message and posting message in queue from same channel).
Have different Channel for Consuming for eg:-

  @Output("QuickScore")

  MessageChannel publishMessageToQuickScore();



  @Input("Score")

  SubscribableChannel receivedAtQuickScore();

In your code change Channel name for @input or @output.