Resolved producer cache #127, and producer handle single partition st…

…ation #128
superstreamlabs · Aug 17, 2023 · 73ad457 · 73ad457
1 parent 2efbc36
commit 73ad457
Show file tree

Hide file tree

Showing 2 changed files with 23 additions and 14 deletions.
diff --git a/src/Memphis.Client/MemphisClient.cs b/src/Memphis.Client/MemphisClient.cs
@@ -121,6 +121,7 @@ public async Task<MemphisProducer> CreateProducer(MemphisProducerOptions produce
  string stationName = producerOptions.StationName;
  string producerName = producerOptions.ProducerName.ToLower();
  bool generateRandomSuffix = producerOptions.GenerateUniqueSuffix;
+ string internalStationName = MemphisUtil.GetInternalName(stationName);
 
  if (_brokerConnection.IsClosed())
  {
@@ -131,6 +132,14 @@ public async Task<MemphisProducer> CreateProducer(MemphisProducerOptions produce
  {
  producerName = $"{producerName}_{MemphisUtil.GetUniqueKey(8)}";
  }
+ else
+ {
+ var producerCacheKey = $"{internalStationName}_{producerName}";
+ if (_producerCache.TryGetValue(producerCacheKey, out MemphisProducer producer))
+ {
+ return producer;
+ }
+ }
 
  try
  {
@@ -158,14 +167,13 @@ public async Task<MemphisProducer> CreateProducer(MemphisProducerOptions produce
  throw new MemphisException(createProducerResponse.Error);
  }
 
- string internalStationName = MemphisUtil.GetInternalName(stationName);
 
  _stationSchemaVerseToDlsMap.AddOrUpdate(internalStationName, createProducerResponse.SchemaVerseToDls, (_, _) => createProducerResponse.SchemaVerseToDls);
  _clusterConfigurations.AddOrUpdate(MemphisSdkClientUpdateTypes.SEND_NOTIFICATION, createProducerResponse.SendNotification, (_, _) => createProducerResponse.SendNotification);
 
  await ListenForSchemaUpdate(internalStationName, createProducerResponse.SchemaUpdate);
 
- if(createProducerResponse.PartitionsUpdate is not null)
+ if (createProducerResponse.PartitionsUpdate is not null)
  {
  _stationPartitions.AddOrUpdate(internalStationName, createProducerResponse.PartitionsUpdate, (_, _) => createProducerResponse.PartitionsUpdate);
  }
@@ -176,14 +184,14 @@ public async Task<MemphisProducer> CreateProducer(MemphisProducerOptions produce
  if (partitionsUpdate.PartitionsList == null)
  {
  producer.PartitionResolver = new(1);
- } 
- else 
+ }
+ else
  {
  producer.PartitionResolver = new(partitionsUpdate.PartitionsList);
  }
  }
- var producerKey = $"{internalStationName}_{producerName.ToLower()}";
- _producerCache.AddOrUpdate(producerKey, producer, (_, _) => producer);
+ var producerCacheKey = $"{internalStationName}_{producerName.ToLower()}";
+ _producerCache.AddOrUpdate(producerCacheKey, producer, (_, _) => producer);
  return producer;
  }
  catch (MemphisException)

diff --git a/src/Memphis.Client/Producer/MemphisProducer.cs b/src/Memphis.Client/Producer/MemphisProducer.cs
@@ -75,17 +75,18 @@ internal async Task ProduceToBrokerAsync(byte[] message, NameValueCollection hea
  await EnsureMessageIsValid(message, headers);
 
  string streamName = _internalStationName;
- if(_memphisClient.StationPartitions.TryGetValue(_stationName, out var partitions))
+ if (_memphisClient.StationPartitions.TryGetValue(_stationName, out var partitions))
  {
- if (partitions != null)
+ if (partitions != null && partitions.PartitionsList != null)
  {
- if (partitions.PartitionsList != null)
+ if(partitions.PartitionsList.Length == 1)
  {
- if (partitions.PartitionsList.Length > 0)
- {
- var partition = PartitionResolver.Resolve();
- streamName = $"{streamName}${partition}";
- }
+ streamName = $"{_internalStationName}${partitions.PartitionsList[0]}";
+ }
+ else if (partitions.PartitionsList.Length > 1)
+ {
+ var partition = PartitionResolver.Resolve();
+ streamName = $"{streamName}${partition}";
  }
  }
  }