client/src/main/java/io/hoplin/rpc/DefaultRpcServer.java
/**
   * Setup consumer in 'server' mode
   *
   * @param handler
   */
Setup consumer in 'server' mode
@param handler
Setup consumer in ' server ' mode @param handler
@SuppressWarnings("unchecked")
  private void consumeRequest(final Function<I, O> handler) {
    try {
      final AMQP.Queue.BindOk bindStatus = channel
          .queueBind(requestQueueName, exchange, routingKey);
      log.info("consumeRequest requestQueueName : {}, {}", requestQueueName, bindStatus);
      int prefetchSize = 1;
      if (binding.getOptions() != null) {
        final QueueOptions options = binding.getOptions();
        prefetchSize = options.getPrefetchCount();
      }
      if (prefetchSize < 1) {
        if (executor instanceof ThreadPoolExecutor) {
          prefetchSize = ((ThreadPoolExecutor) executor).getCorePoolSize();
        } else {
          prefetchSize = 1;
        }
      }
      log.info("QOS prefetch set to : {}", prefetchSize);
      channel.basicQos(prefetchSize);
      channel.basicConsume(requestQueueName, false,
          new RpcResponderConsumer(channel, handler, executor, metrics));
    } catch (final Exception e) {
      throw new HoplinRuntimeException("Unable to start RPC server consumer", e);
    }
  }
@ SuppressWarnings ( "unchecked" ) private void consumeRequest ( final Function < I , O > handler ) { try { final AMQP . Queue . BindOk bindStatus = channel . queueBind ( requestQueueName , exchange , routingKey ) ; log . info ( "consumeRequest requestQueueName : {}, {}" , requestQueueName , bindStatus ) ; int prefetchSize = 1 ; if ( binding . getOptions ( ) != null ) { final QueueOptions options = binding . getOptions ( ) ; prefetchSize = options . getPrefetchCount ( ) ; } if ( prefetchSize < 1 ) { if ( executor instanceof ThreadPoolExecutor ) { prefetchSize = ( ( ThreadPoolExecutor ) executor ) . getCorePoolSize ( ) ; } else { prefetchSize = 1 ; } } log . info ( "QOS prefetch set to : {}" , prefetchSize ) ; channel . basicQos ( prefetchSize ) ; channel . basicConsume ( requestQueueName , false , new RpcResponderConsumer ( channel , handler , executor , metrics ) ) ; } catch ( final Exception e ) { throw new HoplinRuntimeException ( "Unable to start RPC server consumer" , e ) ; } }
Setup consumer in 'server' mode
 @param handler
Setup consumer in ' server ' mode @param handler
// QOS prefetch is set to unlimited (0 or -1) so we are capping it at the core pool size