Added ConsumerType to Store/Get offset methods version-0.0.2

iggy-rs · Jul 27, 2023 · 2faa3a2 · 2faa3a2
1 parent 5d1393b
commit 2faa3a2
Show file tree

Hide file tree

Showing 6 changed files with 29 additions and 4 deletions.
diff --git a/Iggy_SDK.sln.DotSettings.user b/Iggy_SDK.sln.DotSettings.user
@@ -21,6 +21,8 @@
     &lt;TestId&gt;xUnit::8514D555-35CF-4538-97D3-DC42A182B694::net7.0::Iggy_SDK_Tests.MapperTests.BinaryMapper&lt;/TestId&gt;
     &lt;TestId&gt;xUnit::8514D555-35CF-4538-97D3-DC42A182B694::net7.0::Iggy_SDK_Tests.UtilityTests.ToSnakeCaseMessagePolicy&lt;/TestId&gt;
     &lt;TestId&gt;xUnit::8514D555-35CF-4538-97D3-DC42A182B694::net7.0::Iggy_SDK_Tests.ContractTests.TcpContract.TcpContracts_MessageSendRequest_HasCorrectBytes&lt;/TestId&gt;
+    &lt;TestId&gt;xUnit::8514D555-35CF-4538-97D3-DC42A182B694::net7.0::Iggy_SDK_Tests.ContractTests.TcpContract.TcpContracts_GetOffset_HasCorrectBytes&lt;/TestId&gt;
+    &lt;TestId&gt;xUnit::8514D555-35CF-4538-97D3-DC42A182B694::net7.0::Iggy_SDK_Tests.ContractTests.TcpContract&lt;/TestId&gt;
   &lt;/TestAncestor&gt;
 &lt;/SessionState&gt;</s:String>
 	<s:String x:Key="/Default/Environment/UnitTesting/UnitTestSessionStore/Sessions/=fdae582f_002D649c_002D46a6_002D8ad5_002D876f70bf8246/@EntryIndexedValue">&lt;SessionState ContinuousTestingMode="0" Name="GetTopicByIdAsync_Returns200Ok_WhenFound" xmlns="urn:schemas-jetbrains-com:jetbrains-ut-session"&gt;

diff --git a/Iggy_SDK/Contracts/Http/OffsetContract.cs b/Iggy_SDK/Contracts/Http/OffsetContract.cs
@@ -1,7 +1,10 @@
+using Iggy_SDK.Enums;
+
 namespace Iggy_SDK.Contracts.Http;
 
 public sealed class OffsetContract
 {
+	public required ConsumerType ConsumerType { get; init; }
 	public required int ConsumerId { get; init; }
 	public required int PartitionId { get; init; }
 	public required ulong Offset { get; init; }

diff --git a/Iggy_SDK/Contracts/Http/OffsetRequest.cs b/Iggy_SDK/Contracts/Http/OffsetRequest.cs
@@ -1,7 +1,10 @@
+using Iggy_SDK.Enums;
+
 namespace Iggy_SDK.Contracts.Http;
 
 public sealed class OffsetRequest
 {
+	public required ConsumerType ConsumerType { get; init; }
 	public required int StreamId { get; init; }
 	public required int TopicId { get; init; }
 	public required int ConsumerId { get; init; }

diff --git a/Iggy_SDK/Contracts/Tcp/TcpContracts.cs b/Iggy_SDK/Contracts/Tcp/TcpContracts.cs
@@ -156,7 +156,11 @@ internal static byte[] DeleteTopic(int streamId, int topicId)
     internal static byte[] UpdateOffset(int streamId, int topicId, OffsetContract contract)
     {
         Span<byte> bytes = stackalloc byte[sizeof(int) * 4 + sizeof(ulong) + 1];
-        bytes[0] = 0;
+        bytes[0] = contract.ConsumerType switch
+        {
+            ConsumerType.Consumer => 0,
+            ConsumerType.ConsumerGroup => 1,
+        };
         BinaryPrimitives.WriteInt32LittleEndian(bytes[1..5], contract.ConsumerId);
         BinaryPrimitives.WriteInt32LittleEndian(bytes[5..9], streamId);
         BinaryPrimitives.WriteInt32LittleEndian(bytes[9..13], topicId);
@@ -168,7 +172,11 @@ internal static byte[] UpdateOffset(int streamId, int topicId, OffsetContract co
     internal static byte[] GetOffset(OffsetRequest request)
     {
         Span<byte> bytes = stackalloc byte[sizeof(int) * 4 + 1];
-        bytes[0] = 0;
+        bytes[0] = request.ConsumerType switch
+        {
+            ConsumerType.Consumer => 0,
+            ConsumerType.ConsumerGroup => 1,
+        };
         BinaryPrimitives.WriteInt32LittleEndian(bytes[1..5], request.ConsumerId);
         BinaryPrimitives.WriteInt32LittleEndian(bytes[5..9], request.StreamId);
         BinaryPrimitives.WriteInt32LittleEndian(bytes[9..13], request.TopicId);

diff --git a/Iggy_SDK_Tests/Utils/Offsets/OffsetFactory.cs b/Iggy_SDK_Tests/Utils/Offsets/OffsetFactory.cs
@@ -1,5 +1,6 @@
 using Iggy_SDK.Contracts;
 using Iggy_SDK.Contracts.Http;
+using Iggy_SDK.Enums;
 
 namespace Iggy_SDK_Tests.Utils.Offset;
 
@@ -9,6 +10,7 @@ internal static OffsetRequest CreateOffsetRequest()
 	{
 		return new OffsetRequest
 		{
+			ConsumerType = ConsumerType.Consumer,
 			TopicId = Random.Shared.Next(1, 10),
 			PartitionId = Random.Shared.Next(1, 10),
 			ConsumerId = Random.Shared.Next(1, 10),
@@ -29,6 +31,7 @@ internal static OffsetContract CreateOffsetContract()
 	{
 		return new OffsetContract
 		{
+			ConsumerType = ConsumerType.Consumer,
 			Offset = (ulong)Random.Shared.Next(1, 10),
 			ConsumerId = Random.Shared.Next(1, 10),
 			PartitionId = Random.Shared.Next(1, 10),

diff --git a/Iggy_Sample_Consumer/Program.cs b/Iggy_Sample_Consumer/Program.cs
@@ -9,10 +9,16 @@
 var jsonOptions = new JsonSerializerOptions();
 jsonOptions.PropertyNamingPolicy = new ToSnakeCaseNamingPolicy();
 jsonOptions.WriteIndented = true;
-var protocol = Protocol.Tcp;
+// var protocol = Protocol.Tcp;
+// var bus = MessageStreamFactory.CreateMessageStream(options =>
+// {
+//     options.BaseAdress = "127.0.0.1:8090";
+//     options.Protocol = protocol;
+// });
+var protocol = Protocol.Http;
 var bus = MessageStreamFactory.CreateMessageStream(options =>
 {
-    options.BaseAdress = "127.0.0.1:8090";
+    options.BaseAdress = "http://127.0.0.1:3000";
     options.Protocol = protocol;
 });