client/src/main/java/io/hoplin/DefaultQueueConsumer.java
/**
   * If you don't send the ack back, the consumer continues to fetch subsequent messages; however,
   * when you disconnect the consumer, all the messages will still be in the queue. Messages are not
   * consumed until RabbitMQ receives the corresponding ack.
   * <p>
   * Note: A message must be acknowledged only once;
   * </p>
   */
If you don't send the ack back, the consumer continues to fetch subsequent messages; however,
when you disconnect the consumer, all the messages will still be in the queue. Messages are not
consumed until RabbitMQ receives the corresponding ack.

Note: A message must be acknowledged only once;

If you don ' t send the ack back the consumer continues to fetch subsequent messages ; however when you disconnect the consumer all the messages will still be in the queue . Messages are not consumed until RabbitMQ receives the corresponding ack . Note : A message must be acknowledged only once ;
@SuppressWarnings("unchecked")
  @Override
  public void handleDelivery(final String consumerTag, final Envelope envelope,
      final AMQP.BasicProperties properties, byte[] body) {
    metrics.markMessageReceived();
    metrics.incrementReceived(body.length);
    final MessageContext context = MessageContext
        .create(queue, consumerTag, envelope, properties, body);
    CompletableFuture.runAsync(() -> {
      AckStrategy ack;
      try {
        ack = ackFromOptions(queueOptions);
        final JsonMessagePayloadCodec codec = new JsonMessagePayloadCodec(handlers.keySet());
        final MessagePayload<?> message = codec.deserialize(body, MessagePayload.class);
        final Object val = message.getPayload();
        final Class<?> targetClass = message.getTypeAsClass();
        final Collection<MethodReference<?>> consumers = handlers.get(targetClass);
        final List<Throwable> exceptions = new ArrayList<>();
        int invokedHandlers = 0;
        final boolean batchRequest = isBatchedRequest(context);
        Reply<?> reply = null;
        for (final MethodReference reference : consumers) {
          reply = null;
          try {
            final BiFunction<Object, MessageContext, Reply<?>> handler = reference.getHandler();
            final Class<?> root = reference.getRootType();
            if (root == targetClass) {
              ++invokedHandlers;
              reply = execute(context, val, handler);
            } else { 
              final Optional<?> castedValue = safeCast(val, targetClass);
              if (castedValue.isPresent()) {
                ++invokedHandlers;
                reply = execute(context, castedValue.get(), handler);
              }
            }
            if (log.isDebugEnabled()) {
              log.debug("reply : {}", reply);
            }
            if (batchRequest) {
              break;
            }
          } catch (final Exception e) {
            exceptions.add(e);
            log.error("Handler error for message  : " + message, e);
          }
        }
        if (invokedHandlers == 0) {
          throw new HoplinRuntimeException("No handlers defined for type : " + targetClass);
        }
        if (batchRequest) {
          final String replyTo = properties.getReplyTo();
          final String correlationId = properties.getCorrelationId();
          final Map<String, Object> headers = properties.getHeaders();
          final Object batchId = headers.get("x-batch-id");
          headers.put("x-batch-correlationId", correlationId);
          if (log.isDebugEnabled()) {
            log.debug("BatchIncoming context        >  {}", context);
            log.debug("BatchIncoming replyTo        >  {}", replyTo);
            log.debug("BatchIncoming correlationId  >  {}", correlationId);
            log.debug("BatchIncoming batchId        >  {}", batchId);
          }
          final JobExecutionInformation executionInfo = context.getExecutionInfo();
          if (log.isDebugEnabled()) {
            log.debug("Handler time : {}", executionInfo.asElapsedMillis());
          }
          if (reply != null && !reply.isExceptional()) {
            publisher.basicPublish(getChannel(), "", replyTo, reply.getValue(), headers);
            ack = AcknowledgmentStrategies.BASIC_ACK.strategy();
          }
        }
        if (reply != null && reply.isExceptional()) {
          ack = errorStrategy.handleConsumerError(context, reply.getException());
        }
      } catch (final Exception e) {
        log.error("Unable to process message", e);
        try {
          ack = errorStrategy.handleConsumerError(context, e);
        } catch (final Exception ex2) {
          log.error("Exception in error strategy", ex2);
          ack = AcknowledgmentStrategies.BASIC_ACK.strategy();
        }
      }
      AckStrategy.acknowledge(getChannel(), context, ack);
    }, executor);
  }
@ SuppressWarnings ( "unchecked" ) @ Override public void handleDelivery ( final String consumerTag , final Envelope envelope , final AMQP . BasicProperties properties , byte [ ] body ) { metrics . markMessageReceived ( ) ; metrics . incrementReceived ( body . length ) ; final MessageContext context = MessageContext . create ( queue , consumerTag , envelope , properties , body ) ; CompletableFuture . runAsync ( ( ) -> { AckStrategy ack ; try { ack = ackFromOptions ( queueOptions ) ; final JsonMessagePayloadCodec codec = new JsonMessagePayloadCodec ( handlers . keySet ( ) ) ; final MessagePayload < ? > message = codec . deserialize ( body , MessagePayload . class ) ; final Object val = message . getPayload ( ) ; final Class < ? > targetClass = message . getTypeAsClass ( ) ; final Collection < MethodReference < ? > > consumers = handlers . get ( targetClass ) ; final List < Throwable > exceptions = new ArrayList < > ( ) ; int invokedHandlers = 0 ; final boolean batchRequest = isBatchedRequest ( context ) ; Reply < ? > reply = null ; for ( final MethodReference reference : consumers ) { reply = null ; try { final BiFunction < Object , MessageContext , Reply < ? > > handler = reference . getHandler ( ) ; final Class < ? > root = reference . getRootType ( ) ; if ( root == targetClass ) { ++ invokedHandlers ; reply = execute ( context , val , handler ) ; } else { final Optional < ? > castedValue = safeCast ( val , targetClass ) ; if ( castedValue . isPresent ( ) ) { ++ invokedHandlers ; reply = execute ( context , castedValue . get ( ) , handler ) ; } } if ( log . isDebugEnabled ( ) ) { log . debug ( "reply : {}" , reply ) ; } if ( batchRequest ) { break ; } } catch ( final Exception e ) { exceptions . add ( e ) ; log . error ( "Handler error for message  : " + message , e ) ; } } if ( invokedHandlers == 0 ) { throw new HoplinRuntimeException ( "No handlers defined for type : " + targetClass ) ; } if ( batchRequest ) { final String replyTo = properties . getReplyTo ( ) ; final String correlationId = properties . getCorrelationId ( ) ; final Map < String , Object > headers = properties . getHeaders ( ) ; final Object batchId = headers . get ( "x-batch-id" ) ; headers . put ( "x-batch-correlationId" , correlationId ) ; if ( log . isDebugEnabled ( ) ) { log . debug ( "BatchIncoming context        >  {}" , context ) ; log . debug ( "BatchIncoming replyTo        >  {}" , replyTo ) ; log . debug ( "BatchIncoming correlationId  >  {}" , correlationId ) ; log . debug ( "BatchIncoming batchId        >  {}" , batchId ) ; } final JobExecutionInformation executionInfo = context . getExecutionInfo ( ) ; if ( log . isDebugEnabled ( ) ) { log . debug ( "Handler time : {}" , executionInfo . asElapsedMillis ( ) ) ; } if ( reply != null && ! reply . isExceptional ( ) ) { publisher . basicPublish ( getChannel ( ) , "" , replyTo , reply . getValue ( ) , headers ) ; ack = AcknowledgmentStrategies . BASIC_ACK . strategy ( ) ; } } if ( reply != null && reply . isExceptional ( ) ) { ack = errorStrategy . handleConsumerError ( context , reply . getException ( ) ) ; } } catch ( final Exception e ) { log . error ( "Unable to process message" , e ) ; try { ack = errorStrategy . handleConsumerError ( context , e ) ; } catch ( final Exception ex2 ) { log . error ( "Exception in error strategy" , ex2 ) ; ack = AcknowledgmentStrategies . BASIC_ACK . strategy ( ) ; } } AckStrategy . acknowledge ( getChannel ( ) , context , ack ) ; } , executor ) ; }
If you don't send the ack back, the consumer continues to fetch subsequent messages; however,
 when you disconnect the consumer, all the messages will still be in the queue.
If you don ' t send the ack back the consumer continues to fetch subsequent messages ; however when you disconnect the consumer all the messages will still be in the queue .
// Down cast if necessary // can't have multiple handlers for batched requests // TODO : This should be handled better