fix: add bookkeeping for number of active subscriptions

.gitignore

@@ -46,3 +46,4 @@ examples/lib/.classpath
 examples/lib/.project
 examples/lib/bin
 .vscode/
+momento-sdk/bin

Makefile

@@ -42,6 +42,10 @@ test-leaderboard-service:
 test-topics-service:
 	@CONSISTENT_READS=1 ./gradlew test-topics-service
 
+## Run the topics subscription initialization tests
+test-topics-subscription-initialization:
+	@CONSISTENT_READS=1 ./gradlew test-topics-subscription-initialization
+
 ## Run the http service tests
 test-http-service:
 	@echo "No tests for http service."

momento-sdk/build.gradle.kts

@@ -109,3 +109,8 @@ registerIntegrationTestTask(
     "test-retries",
     listOf("momento.sdk.retry.*")
 )
+
+registerIntegrationTestTask(
+    "test-topics-subscription-initialization",
+    listOf("momento.sdk.retry.TopicsSubscriptionInitializationTest")
+)

momento-sdk/src/intTest/java/momento/sdk/retry/BaseMomentoLocalTestClass.java

@@ -14,6 +14,8 @@
 import momento.sdk.config.Configurations;
 import momento.sdk.config.TopicConfiguration;
 import momento.sdk.config.TopicConfigurations;
+import momento.sdk.config.transport.GrpcConfiguration;
+import momento.sdk.config.transport.StaticTransportStrategy;
 import momento.sdk.responses.cache.control.CacheCreateResponse;
 import momento.sdk.retry.utils.MomentoLocalMiddleware;
 import momento.sdk.retry.utils.MomentoLocalMiddlewareArgs;
@@ -126,6 +128,39 @@ public static void withCacheAndTopicClient(
     }
   }
 
+  public static void withCacheAndTopicClientWithNumStreamChannels(
+      int numStreamChannels,
+      MomentoLocalMiddlewareArgs testMetricsMiddlewareArgs,
+      TopicTestCallback testCallback)
+      throws Exception {
+
+    final String cacheName = testCacheName();
+    final String hostname =
+        Optional.ofNullable(System.getenv("MOMENTO_HOSTNAME")).orElse("127.0.0.1");
+    final int port =
+        Optional.ofNullable(System.getenv("MOMENTO_PORT")).map(Integer::parseInt).orElse(8080);
+    final CredentialProvider credentialProvider = new MomentoLocalProvider(hostname, port);
+
+    final GrpcConfiguration grpcConfig =
+        new GrpcConfiguration(Duration.ofMillis(15000))
+            .withNumStreamGrpcChannels(numStreamChannels);
+    final TopicConfiguration topicConfiguration =
+        new TopicConfiguration(new StaticTransportStrategy(grpcConfig))
+            .withMiddleware(new MomentoLocalMiddleware(testMetricsMiddlewareArgs));
+
+    try (final CacheClient cacheClient =
+            CacheClient.builder(
+                    credentialProvider, Configurations.Laptop.latest(), DEFAULT_TTL_SECONDS)
+                .build();
+        final TopicClient topicClient =
+            TopicClient.builder(credentialProvider, topicConfiguration).build()) {
+      if (cacheClient.createCache(cacheName).join() instanceof CacheCreateResponse.Error) {
+        throw new RuntimeException("Failed to create cache: " + cacheName);
+      }
+      testCallback.run(topicClient, cacheName);
+    }
+  }
+
   @FunctionalInterface
   public interface CacheTestCallback {
     void run(CacheClient cc, String cacheName) throws Exception;