Merge pull request #101 from memphisdev/fix-create-remove-consumer-pr…

…oducer

Fix create remove consumer producer

src/Memphis.Client.UnitTests/MemphisClientTest.cs

@@ -35,6 +35,7 @@ public MemphisClientTest(ITestOutputHelper testOutputHelper)
  _testOutputHelper = testOutputHelper;
 
  this._brokerOptions = ConnectionFactory.GetDefaultOptions();
+ this._brokerOptions.User = "root";
  this._connectionId = Guid.NewGuid().ToString();
  this._connectionMock = new Mock<IConnection>();
  this._jetStreamMock = new Mock<IJetStream>();

src/Memphis.Client/Consumer/MemphisConsumer.cs

@@ -100,6 +100,7 @@ public async Task DestroyAsync()
  StationName = _consumerOptions.StationName,
  ConnectionId = _memphisClient.ConnectionId,
  Username = _memphisClient.Username,
+ RequestVersion = 1,
  };
 
  var removeConsumerModelJson = JsonSerDes.PrepareJsonString<RemoveConsumerRequest>(removeConsumerModel);

src/Memphis.Client/MemphisClient.cs

@@ -64,6 +64,11 @@ public MemphisClient(Options brokerConnOptions, IConnection brokerConnection,
  _jetStreamContext = jetStreamContext ?? throw new ArgumentNullException(nameof(jetStreamContext));
  _connectionId = connectionId ?? throw new ArgumentNullException(nameof(connectionId));
  _userName = brokerConnOptions.User;
+ int usernameSeparatorIndex = _userName.LastIndexOf('$');
+ if (usernameSeparatorIndex >= 0)
+ {
+ _userName = _userName.Substring(0, usernameSeparatorIndex);
+ }
 
  _cancellationTokenSource = new();
 
@@ -298,6 +303,7 @@ public async Task<MemphisConsumer> CreateConsumer(MemphisConsumerOptions consume
  UserName = _userName,
  StartConsumeFromSequence = consumerOptions.StartConsumeFromSequence,
  LastMessages = consumerOptions.LastMessages,
+ RequestVersion = 1,
  };
 
  var createConsumerModelJson = JsonSerDes.PrepareJsonString<CreateConsumerRequest>(createConsumerModel);

src/Memphis.Client/Models/Request/CreateConsumerRequest.cs

@@ -36,5 +36,8 @@ internal sealed class CreateConsumerRequest
 
  [DataMember(Name = "last_messages")]
  public int LastMessages { get; set; }
+
+ [DataMember(Name = "req_version")]
+ public int RequestVersion { get; set; }
  }
 }

src/Memphis.Client/Models/Request/RemoveConsumerRequest.cs

@@ -18,5 +18,8 @@ internal sealed class RemoveConsumerRequest
 
  [DataMember(Name = "username")]
  public string Username { get; set; }
+
+ [DataMember(Name = "req_version")]
+ public int RequestVersion { get; set; }
  }
 }

src/Memphis.Client/Models/Request/RemoveProducerRequest.cs

@@ -18,5 +18,8 @@ internal sealed class RemoveProducerRequest
 
  [DataMember(Name = "username")]
  public string Username { get; set; }
+
+ [DataMember(Name = "req_version")]
+ public int RequestVersion { get; set; }
  }
 }

src/Memphis.Client/Producer/MemphisProducer.cs

@@ -132,7 +132,8 @@ public async Task DestroyAsync()
  ProducerName = _producerName,
  StationName = _stationName,
  ConnectionId = _memphisClient.ConnectionId,
- Username = _memphisClient.Username
+ Username = _memphisClient.Username,
+ RequestVersion = 1,
  };
 
  var removeProducerModelJson = JsonSerDes.PrepareJsonString<RemoveProducerRequest>(removeProducerModel);