feat: add configurability for number of gRPC channels

src/Cache/CacheClient.php

@@ -35,6 +35,7 @@
 use Momento\Cache\CacheOperationTypes\CreateCacheResponse;
 use Momento\Cache\CacheOperationTypes\DeleteCacheResponse;
 use Momento\Cache\CacheOperationTypes\ListCachesResponse;
+use Momento\Cache\Errors\InvalidArgumentError;
 use Momento\Cache\Internal\IdleDataClientWrapper;
 use Momento\Cache\Internal\ScsControlClient;
 use Momento\Cache\Internal\ScsDataClient;
@@ -54,7 +55,12 @@ class CacheClient implements LoggerAwareInterface
     protected ILoggerFactory $loggerFactory;
     protected LoggerInterface $logger;
     private ScsControlClient $controlClient;
-    private IdleDataClientWrapper $dataClientWrapper;
+
+    /**
+     * @var IdleDataClientWrapper[]
+     */
+    private array $dataClients;
+    private int $nextDataClientIndex = 0;
 
     /**
      * @param IConfiguration $configuration Configuration to use for transport.
@@ -77,7 +83,16 @@ public function __construct(
                 $defaultTtlSeconds
             );
         };
-        $this->dataClientWrapper = new IdleDataClientWrapper($dataClientFactory, $this->configuration);
+        $this->dataClients = [];
+
+        $numGrpcChannels = $configuration->getTransportStrategy()->getGrpcConfig()->getNumGrpcChannels();
+        $forceNewChannels = $configuration->getTransportStrategy()->getGrpcConfig()->getForceNewChannel();
+        if (($numGrpcChannels > 1) && (! $forceNewChannels)) {
+            throw new InvalidArgumentError("When setting NumGrpcChannels > 1, you must also set ForceNewChannel to true, or else the gRPC library will re-use the same channel.");
+        }
+        for ($i = 0; $i < $numGrpcChannels; $i++) {
+            array_push($this->dataClients, new IdleDataClientWrapper($dataClientFactory, $this->configuration));
+        }
     }
 
     /**
@@ -171,7 +186,7 @@ public function deleteCache(string $cacheName): DeleteCacheResponse
      */
     public function setAsync(string $cacheName, string $key, string $value, int $ttlSeconds = 0): ResponseFuture
     {
-        return $this->dataClientWrapper->getClient()->set($cacheName, $key, $value, $ttlSeconds);
+        return $this->getNextDataClient()->set($cacheName, $key, $value, $ttlSeconds);
     }
 
     /**
@@ -223,7 +238,7 @@ public function set(string $cacheName, string $key, string $value, int $ttlSecon
      */
     public function getAsync(string $cacheName, string $key): ResponseFuture
     {
-        return $this->dataClientWrapper->getClient()->get($cacheName, $key);
+        return $this->getNextDataClient()->get($cacheName, $key);
     }
 
     /**
@@ -282,7 +297,7 @@ public function get(string $cacheName, string $key): GetResponse
      */
     public function setIfNotExistsAsync(string $cacheName, string $key, string $value, int $ttlSeconds = 0): ResponseFuture
     {
-        return $this->dataClientWrapper->getClient()->setIfNotExists($cacheName, $key, $value, $ttlSeconds);
+        return $this->getNextDataClient()->setIfNotExists($cacheName, $key, $value, $ttlSeconds);
     }
 
     /**
@@ -336,7 +351,7 @@ public function setIfNotExists(string $cacheName, string $key, string $value, in
      */
     public function deleteAsync(string $cacheName, string $key): ResponseFuture
     {
-        return $this->dataClientWrapper->getClient()->delete($cacheName, $key);
+        return $this->getNextDataClient()->delete($cacheName, $key);
     }
 
     /**
@@ -387,7 +402,7 @@ public function delete(string $cacheName, string $key): DeleteResponse
      */
     public function keysExistAsync(string $cacheName, array $keys): ResponseFuture
     {
-        return $this->dataClientWrapper->getClient()->keysExist($cacheName, $keys);
+        return $this->getNextDataClient()->keysExist($cacheName, $keys);
     }
 
     /**
@@ -440,7 +455,7 @@ public function keysExist(string $cacheName, array $keys): KeysExistResponse
      */
     public function keyExistsAsync(string $cacheName, string $key): ResponseFuture
     {
-        return $this->dataClientWrapper->getClient()->keyExists($cacheName, $key);
+        return $this->getNextDataClient()->keyExists($cacheName, $key);
     }
 
     /**
@@ -495,7 +510,7 @@ public function incrementAsync(
         string $cacheName, string $key, int $amount = 1, ?int $ttlSeconds = null
     ): ResponseFuture
     {
-        return $this->dataClientWrapper->getClient()->increment($cacheName, $key, $amount, $ttlSeconds);
+        return $this->getNextDataClient()->increment($cacheName, $key, $amount, $ttlSeconds);
     }
 
     /**
@@ -543,7 +558,7 @@ public function increment(
      */
     public function listFetch(string $cacheName, string $listName): ListFetchResponse
     {
-        return $this->dataClientWrapper->getClient()->listFetch($cacheName, $listName);
+        return $this->getNextDataClient()->listFetch($cacheName, $listName);
     }
 
     /**
@@ -569,7 +584,7 @@ public function listPushFront(
         string $cacheName, string $listName, string $value, ?int $truncateBackToSize = null, ?CollectionTtl $ttl = null
     ): ListPushFrontResponse
     {
-        return $this->dataClientWrapper->getClient()->listPushFront($cacheName, $listName, $value, $truncateBackToSize, $ttl);
+        return $this->getNextDataClient()->listPushFront($cacheName, $listName, $value, $truncateBackToSize, $ttl);
     }
 
     /**
@@ -595,7 +610,7 @@ public function listPushBack(
         string $cacheName, string $listName, string $value, ?int $truncateFrontToSize = null, ?CollectionTtl $ttl = null
     ): ListPushBackResponse
     {
-        return $this->dataClientWrapper->getClient()->listPushBack($cacheName, $listName, $value, $truncateFrontToSize, $ttl);
+        return $this->getNextDataClient()->listPushBack($cacheName, $listName, $value, $truncateFrontToSize, $ttl);
     }
 
     /**
@@ -617,7 +632,7 @@ public function listPushBack(
      */
     public function listPopFront(string $cacheName, string $listName): ListPopFrontResponse
     {
-        return $this->dataClientWrapper->getClient()->listPopFront($cacheName, $listName);
+        return $this->getNextDataClient()->listPopFront($cacheName, $listName);
     }
 
     /**
@@ -639,7 +654,7 @@ public function listPopFront(string $cacheName, string $listName): ListPopFrontR
      */
     public function listPopBack(string $cacheName, string $listName): ListPopBackResponse
     {
-        return $this->dataClientWrapper->getClient()->listPopBack($cacheName, $listName);
+        return $this->getNextDataClient()->listPopBack($cacheName, $listName);
     }
 
     /**
@@ -659,7 +674,7 @@ public function listPopBack(string $cacheName, string $listName): ListPopBackRes
      */
     public function listRemoveValue(string $cacheName, string $listName, string $value): ListRemoveValueResponse
     {
-        return $this->dataClientWrapper->getClient()->listRemoveValue($cacheName, $listName, $value);
+        return $this->getNextDataClient()->listRemoveValue($cacheName, $listName, $value);
     }
 
     /**
@@ -680,7 +695,7 @@ public function listRemoveValue(string $cacheName, string $listName, string $val
      */
     public function listLength(string $cacheName, string $listName): ListLengthResponse
     {
-        return $this->dataClientWrapper->getClient()->listLength($cacheName, $listName);
+        return $this->getNextDataClient()->listLength($cacheName, $listName);
     }
 
     /**
@@ -702,7 +717,7 @@ public function listLength(string $cacheName, string $listName): ListLengthRespo
      */
     public function dictionarySetField(string $cacheName, string $dictionaryName, string $field, string $value, ?CollectionTtl $ttl = null): DictionarySetFieldResponse
     {
-        return $this->dataClientWrapper->getClient()->dictionarySetField($cacheName, $dictionaryName, $field, $value, $ttl);
+        return $this->getNextDataClient()->dictionarySetField($cacheName, $dictionaryName, $field, $value, $ttl);
     }
 
     /**
@@ -725,7 +740,7 @@ public function dictionarySetField(string $cacheName, string $dictionaryName, st
      */
     public function dictionaryGetField(string $cacheName, string $dictionaryName, string $field): DictionaryGetFieldResponse
     {
-        return $this->dataClientWrapper->getClient()->dictionaryGetField($cacheName, $dictionaryName, $field);
+        return $this->getNextDataClient()->dictionaryGetField($cacheName, $dictionaryName, $field);
     }
 
     /**
@@ -747,7 +762,7 @@ public function dictionaryGetField(string $cacheName, string $dictionaryName, st
      */
     public function dictionaryFetch(string $cacheName, string $dictionaryName): DictionaryFetchResponse
     {
-        return $this->dataClientWrapper->getClient()->dictionaryFetch($cacheName, $dictionaryName);
+        return $this->getNextDataClient()->dictionaryFetch($cacheName, $dictionaryName);
     }
 
     /**
@@ -768,7 +783,7 @@ public function dictionaryFetch(string $cacheName, string $dictionaryName): Dict
      */
     public function dictionarySetFields(string $cacheName, string $dictionaryName, array $elements, ?CollectionTtl $ttl = null): DictionarySetFieldsResponse
     {
-        return $this->dataClientWrapper->getClient()->dictionarySetFields($cacheName, $dictionaryName, $elements, $ttl);
+        return $this->getNextDataClient()->dictionarySetFields($cacheName, $dictionaryName, $elements, $ttl);
     }
 
     /**
@@ -799,7 +814,7 @@ public function dictionarySetFields(string $cacheName, string $dictionaryName, a
      */
     public function dictionaryGetFields(string $cacheName, string $dictionaryName, array $fields): DictionaryGetFieldsResponse
     {
-        return $this->dataClientWrapper->getClient()->dictionaryGetFields($cacheName, $dictionaryName, $fields);
+        return $this->getNextDataClient()->dictionaryGetFields($cacheName, $dictionaryName, $fields);
     }
 
     /**
@@ -828,7 +843,7 @@ public function dictionaryIncrement(
         string $cacheName, string $dictionaryName, string $field, int $amount = 1, ?CollectionTtl $ttl = null
     ): DictionaryIncrementResponse
     {
-        return $this->dataClientWrapper->getClient()->dictionaryIncrement($cacheName, $dictionaryName, $field, $amount, $ttl);
+        return $this->getNextDataClient()->dictionaryIncrement($cacheName, $dictionaryName, $field, $amount, $ttl);
     }
 
     /**
@@ -847,7 +862,7 @@ public function dictionaryIncrement(
      */
     public function dictionaryRemoveField(string $cacheName, string $dictionaryName, string $field): DictionaryRemoveFieldResponse
     {
-        return $this->dataClientWrapper->getClient()->dictionaryRemoveField($cacheName, $dictionaryName, $field);
+        return $this->getNextDataClient()->dictionaryRemoveField($cacheName, $dictionaryName, $field);
     }
 
     /**
@@ -866,7 +881,7 @@ public function dictionaryRemoveField(string $cacheName, string $dictionaryName,
      */
     public function dictionaryRemoveFields(string $cacheName, string $dictionaryName, array $fields): DictionaryRemoveFieldsResponse
     {
-        return $this->dataClientWrapper->getClient()->dictionaryRemoveFields($cacheName, $dictionaryName, $fields);
+        return $this->getNextDataClient()->dictionaryRemoveFields($cacheName, $dictionaryName, $fields);
     }
 
     /**
@@ -895,7 +910,7 @@ public function dictionaryRemoveFields(string $cacheName, string $dictionaryName
      */
     public function setAddElementAsync(string $cacheName, string $setName, string $element, ?CollectionTtl $ttl = null): ResponseFuture
     {
-        return $this->dataClientWrapper->getClient()->setAddElement($cacheName, $setName, $element, $ttl);
+        return $this->getNextDataClient()->setAddElement($cacheName, $setName, $element, $ttl);
     }
 
     /**
@@ -944,7 +959,7 @@ public function setAddElement(string $cacheName, string $setName, string $elemen
      */
     public function setAddElementsAsync(string $cacheName, string $setName, array $elements, ?CollectionTtl $ttl = null): ResponseFuture
     {
-        return $this->dataClientWrapper->getClient()->setAddElements($cacheName, $setName, $elements, $ttl);
+        return $this->getNextDataClient()->setAddElements($cacheName, $setName, $elements, $ttl);
     }
 
     /**
@@ -994,7 +1009,7 @@ public function setAddElements(string $cacheName, string $setName, array $elemen
      */
     public function setFetchAsync(string $cacheName, string $setName): ResponseFuture
     {
-        return $this->dataClientWrapper->getClient()->setFetch($cacheName, $setName);
+        return $this->getNextDataClient()->setFetch($cacheName, $setName);
     }
 
     /**
@@ -1044,7 +1059,7 @@ public function setFetch(string $cacheName, string $setName): SetFetchResponse
      */
     public function setLengthAsync(string $cacheName, string $setName): ResponseFuture
     {
-        return $this->dataClientWrapper->getClient()->setLength($cacheName, $setName);
+        return $this->getNextDataClient()->setLength($cacheName, $setName);
     }
 
     /**
@@ -1093,7 +1108,7 @@ public function setLength(string $cacheName, string $setName): SetLengthResponse
      */
     public function setRemoveElementAsync(string $cacheName, string $setName, string $element): ResponseFuture
     {
-        return $this->dataClientWrapper->getClient()->setRemoveElement($cacheName, $setName, $element);
+        return $this->getNextDataClient()->setRemoveElement($cacheName, $setName, $element);
     }
 
     /**
@@ -1114,4 +1129,11 @@ public function setRemoveElement(string $cacheName, string $setName, string $ele
     {
         return $this->setRemoveElementAsync($cacheName, $setName, $element)->wait();
     }
+
+    private function getNextDataClient(): ScsDataClient
+    {
+        $client = $this->dataClients[$this->nextDataClientIndex]->getClient();
+        $this->nextDataClientIndex = ($this->nextDataClientIndex + 1) % count($this->dataClients);
+        return $client;
+    }
 }

src/Config/Configuration.php

@@ -31,9 +31,9 @@ public function getLoggerFactory(): ILoggerFactory {
     }
 
     /**
-     * @return ITransportStrategy|null The currently active transport strategy
+     * @return ITransportStrategy The currently active transport strategy
      */
-    public function getTransportStrategy(): ITransportStrategy|null
+    public function getTransportStrategy(): ITransportStrategy
     {
         return $this->transportStrategy;
     }

src/Config/Configurations/Laptop.php

@@ -37,7 +37,7 @@ public static function v1(?ILoggerFactory $loggerFactory = null): Laptop
     {
         $loggerFactory = $loggerFactory ?? new NullLoggerFactory();
         $grpcConfig = new StaticGrpcConfiguration(5000);
-        $transportStrategy = new StaticTransportStrategy(null, $grpcConfig, $loggerFactory, self::$maxIdleMillis);
+        $transportStrategy = new StaticTransportStrategy($grpcConfig, $loggerFactory, self::$maxIdleMillis);
         return new Laptop($loggerFactory, $transportStrategy);
     }
 }

src/Config/IConfiguration.php

@@ -16,6 +16,12 @@ interface IConfiguration
      */
     public function getLoggerFactory(): ILoggerFactory;
 
+
+    /**
+     * @return ITransportStrategy The currently active transport strategy
+     */
+    public function getTransportStrategy(): ITransportStrategy;
+
     /**
      * Creates a new instance of the Configuration object, updated to use the specified transport strategy.
      *

src/Config/Transport/IGrpcConfiguration.php

@@ -12,4 +12,8 @@ public function withDeadlineMilliseconds(int $deadlineMilliseconds): IGrpcConfig
     public function getForceNewChannel(): ?bool;
 
     public function withForceNewChannel(bool $forceNewChannel) : IGrpcConfiguration;
+
+    public function getNumGrpcChannels(): int;
+
+    public function withNumGrpcChannels(int $numGrpcChannels): IGrpcConfiguration;
 }

src/Config/Transport/ITransportStrategy.php

@@ -7,8 +7,6 @@
 
 interface ITransportStrategy
 {
-    public function getMaxConcurrentRequests(): ?int;
-
     public function getGrpcConfig(): ?IGrpcConfiguration;
 
     public function getMaxIdleMillis(): ?int;

src/Config/Transport/StaticGrpcConfiguration.php

@@ -7,12 +7,14 @@ class StaticGrpcConfiguration implements IGrpcConfiguration
 {
 
     private ?int $deadlineMilliseconds;
-    private ?bool $forceNewChannel;
+    private bool $forceNewChannel;
+    private int $numGrpcChannels;
 
-    public function __construct(?int $deadlineMilliseconds = null, ?bool $forceNewChannel = false)
+    public function __construct(?int $deadlineMilliseconds = null, bool $forceNewChannel = false, int $numGrpcChannels = 1)
     {
         $this->deadlineMilliseconds = $deadlineMilliseconds;
         $this->forceNewChannel = $forceNewChannel;
+        $this->numGrpcChannels = $numGrpcChannels;
     }
 
     public function getDeadlineMilliseconds(): int|null
@@ -22,7 +24,7 @@ public function getDeadlineMilliseconds(): int|null
 
     public function withDeadlineMilliseconds(int $deadlineMilliseconds): StaticGrpcConfiguration
     {
-        return new StaticGrpcConfiguration($deadlineMilliseconds, $this->forceNewChannel);
+        return new StaticGrpcConfiguration($deadlineMilliseconds, $this->forceNewChannel, $this->numGrpcChannels);
     }
 
     public function getForceNewChannel(): bool|null
@@ -32,6 +34,16 @@ public function getForceNewChannel(): bool|null
 
     public function withForceNewChannel(bool $forceNewChannel): StaticGrpcConfiguration
     {
-        return new StaticGrpcConfiguration($this->deadlineMilliseconds, $forceNewChannel);
+        return new StaticGrpcConfiguration($this->deadlineMilliseconds, $forceNewChannel, $this->numGrpcChannels);
+    }
+
+    public function getNumGrpcChannels(): int
+    {
+        return $this->numGrpcChannels;
+    }
+
+    public function withNumGrpcChannels(int $numGrpcChannels): IGrpcConfiguration
+    {
+        return new StaticGrpcConfiguration($this->deadlineMilliseconds, $this->forceNewChannel, $numGrpcChannels);
     }
 }