pubsub Limit the number of subscriber threads how?

[fondberg]

Can someone please provide an example of limiting the number of pulling pubsub threads with the new API so we can start testing?

In the old API it was possible to limit the number of concurrent threads that pulled messages with the following code:

PubSub.MessageConsumer consumer = pubsub.pullAsync(
        subscription,
        messageProcessor,
        PubSub.PullOption.maxQueuedCallbacks(maxQueuedCallbacks),
        PubSub.PullOption.executorFactory(executorFactory)
    );

where maxQueuedCallbacks was an integer amount of msgs to pull and
executorFactory was provided from this

 /**
   * Create an ExecutorFactory instance for use with a single
   * {@link PubSub.MessageConsumer}.
   *
   * <p>The number of threads in the executor determines how many messages can be processed at one
   * time.</p>
   *
   * @see PubSub.PullOption#executorFactory(ExecutorFactory)
   */
  private static ExecutorFactory<ExecutorService> createExecutor(
      final int threads,
      final String threadNameFormat) {
    final ThreadFactory threadFactory = new ThreadFactoryBuilder()
        .setDaemon(true)
        .setNameFormat(threadNameFormat)
        .build();

    final ExecutorService executorService = Executors.newFixedThreadPool(threads, threadFactory);

    return new ExecutorFactory<ExecutorService>() {
      @Override
      public ExecutorService get() {
        return executorService;
      }

      @Override
      public void release(final ExecutorService executor) {
        executorService.shutdownNow();
      }
    };
  }

[garrettjonesgoogle]

There is not a way to do this yet.

[fondberg]

ok I see.

[garrettjonesgoogle]

I take it back, it is possible to do this. Example:

// InstantiatingExecutorProvider will create a new executor for each thing it is passed to;
// Here, Publisher and Subscriber will each get their own executor.
ExecutorProvider executorProvider =
  InstantiatingExecutorProvider.newBuilder()
    .setExecutorThreadCount(1)
    .build();
Publisher publisher =
  Publisher.newBuilder(topicName).setExecutorProvider(executorProvider).build();
Subscriber subscriber =
  Subscriber.newBuilder(topicName).setExecutorProvider(executorProvider).build();

(Note - this comment has been edited since its original version)

[garrettjonesgoogle]

I gave the wrong code in my last comment - it was setting the executor provider for the channel, not for the publishing/subscribing. I have fixed the code.

[fondberg]

Thanks!

[saturnism]

This sounds like #1683

[fondberg]

@garrettjonesgoogle
When trying the above code using 0.12.0-alpha Publisher.newBuilderdoes't exist.
Do I need another version or is the example wrong?

[garrettjonesgoogle]

The example is now out of date - we renamed newBuilder to defaultBuilder (since it returns a prepopulated Builder). New code sample:

// InstantiatingExecutorProvider will create a new executor for each thing it is passed to;
// Here, Publisher and Subscriber will each get their own executor.
ExecutorProvider executorProvider =
  InstantiatingExecutorProvider.newBuilder()
    .setExecutorThreadCount(1)
    .build();
Publisher publisher =
  Publisher.defaultBuilder(topicName).setExecutorProvider(executorProvider).build();
Subscriber subscriber =
  Subscriber.defaultBuilder(topicName).setExecutorProvider(executorProvider).build();

[fondberg]

thanks!

[fondberg]

I have now scratched my head long enough to admit failure...
@garrettjonesgoogle what is wrong with my testcode below (it exits after around 10-20 seconds without receiving any messages)?

    static void setUp() throws IOException {
        ExecutorProvider executorProvider =
                InstantiatingExecutorProvider.newBuilder()
                        .setExecutorThreadCount(1)
                        .build();

        String projectId = "gcp-project";
        String subscriptionId = "test-subscription";
        SubscriptionName subscriptionName = SubscriptionName.create(projectId, subscriptionId);

        // Instantiate an asynchronous message receiver
        MessageReceiver receiver =
                new MessageReceiver() {
                    @Override
                    public void receiveMessage(PubsubMessage message, AckReplyConsumer consumer) {
                        System.out.println("Got message:" + message.toString());
                        consumer.ack();
                    }
                };

        Subscriber subscriber =
                Subscriber.defaultBuilder(subscriptionName, receiver)
                        .setExecutorProvider(executorProvider).build();
        subscriber.addListener(new Subscriber.Listener() {
            public void failed(Subscriber.State from, Throwable failure) {
                System.out.println(failure);
            }
        }, executorProvider.getExecutor());
        subscriber.startAsync();

    }

[garrettjonesgoogle]

@pongad could you take a look?

[fondberg]

@pongad ^^