Improve end to end tests (#12)

.github/workflows/dotnet.yml

@@ -1,5 +1,3 @@
-# This workflow will build, test, and publish a .NET 8 project on push to master
-
 name: .NET
 
 on:
@@ -16,7 +14,7 @@ jobs:
       - name: Setup .NET 8
         uses: actions/setup-dotnet@v3
         with:
-          dotnet-version: '8.0.x'
+          dotnet-version: '8.0.x' 
       - name: Restore dependencies
         run: dotnet restore
       - name: Build

Iggy_SDK/Contracts/Http/ClientInfo.cs

@@ -2,7 +2,7 @@ namespace Iggy_SDK.Contracts.Http;
 
 public sealed class ClientResponse
 {
-    public required uint Id { get; init; }
+    public required uint ClientId { get; init; }
     public required string Adress { get; init; }
     public required uint UserId { get; init; }
     public required string Transport { get; init; }

Iggy_SDK/Contracts/Http/TopicResponse.cs

@@ -7,7 +7,7 @@ public sealed class TopicResponse
     public required int Id { get; init; }
     public required DateTimeOffset CreatedAt { get; init; }
     public required string Name { get; init; }
-    public required ulong SizeBytes { get; init; }
+    public required ulong Size { get; init; }
     public int MessageExpiry { get; init; }
     public required ulong MaxTopicSize { get; init; }
     public required ulong MessagesCount { get; init; }

Iggy_SDK/Extensions/Extensions.cs

@@ -36,6 +36,7 @@ internal static string ToSnakeCase(this string input)
     internal static UInt128 ToUInt128(this Guid g)
     {
         Span<byte> array = stackalloc byte[16];
+        #pragma warning disable CS9191 // The 'ref' modifier for an argument corresponding to 'in' parameter is equivalent to 'in'. Consider using 'in' instead.
         MemoryMarshal.TryWrite(array, ref g);
         var hi = BinaryPrimitives.ReadUInt64LittleEndian(array[..8]);
         var lo = BinaryPrimitives.ReadUInt64LittleEndian(array[8..16]);
@@ -56,6 +57,7 @@ internal static Int128 ToInt128(this byte[] bytes)
     internal static byte[] GetBytesFromGuid(this Guid value)
     {
         Span<byte> bytes = stackalloc byte[16];
+        #pragma warning disable CS9191 // The 'ref' modifier for an argument corresponding to 'in' parameter is equivalent to 'in'. Consider using 'in' instead.
         MemoryMarshal.TryWrite(bytes, ref value);
         return bytes.ToArray();
     }

Iggy_SDK/Headers/HeaderValue.cs

@@ -157,7 +157,7 @@ public byte[] ToBytes()
             HeaderKind.Float => BinaryPrimitives.ReadSingleLittleEndian(Value).ToString(),
             HeaderKind.Double => BinaryPrimitives.ReadDoubleLittleEndian(Value).ToString(),
             _ => throw new InvalidOperationException("Can't convert header")
-        };
+        } ?? throw new InvalidOperationException();
     }
     public bool ToBool()
     {

Iggy_SDK/IggyClient/Implementations/HttpMessageStream.cs

@@ -96,7 +96,7 @@ public async Task<IReadOnlyList<StreamResponse>> GetStreamsAsync(CancellationTok
     }
     public async Task CreateTopicAsync(Identifier streamId, TopicRequest topic, CancellationToken token = default)
     {
-        var json = JsonSerializer.Serialize(topic, JsonConverterFactory.SnakeCaseOptions);
+        var json = JsonSerializer.Serialize(topic, JsonConverterFactory.CreateTopicOptions);
         var data = new StringContent(json, Encoding.UTF8, "application/json");
 
         var response = await _httpClient.PostAsync($"/streams/{streamId}/topics", data, token);
@@ -162,16 +162,7 @@ public async Task SendMessagesAsync(MessageSendRequest request,
 
         if (_messageInvoker is not null)
         {
-            try
-            {
-                await _messageInvoker.SendMessagesAsync(request, token);
-            }
-            catch
-            {
-                var partId = BinaryPrimitives.ReadInt32LittleEndian(request.Partitioning.Value);
-                _logger.LogError("Error encountered while sending messages - Stream ID:{streamId}, Topic ID:{topicId}, Partition ID: {partitionId}",
-                    request.StreamId, request.TopicId, partId);
-            }
+            await _messageInvoker.SendMessagesAsync(request, token);
             return;
         }
         await _channel!.Writer.WriteAsync(request, token);
@@ -256,7 +247,8 @@ public async IAsyncEnumerable<MessageResponse<TMessage>> PollMessagesAsync<TMess
         {
             StoreOffset.Never => false,
             StoreOffset.WhenMessagesAreReceived => true,
-            StoreOffset.AfterProcessingEachMessage => false
+            StoreOffset.AfterProcessingEachMessage => false,
+            _ => throw new ArgumentOutOfRangeException()
         };
         var fetchRequest = new MessageFetchRequest
         {
@@ -401,7 +393,7 @@ public async Task DeleteConsumerGroupAsync(DeleteConsumerGroupRequest request, C
         var response = await _httpClient.GetAsync($"/stats", token);
         if (response.IsSuccessStatusCode)
         {
-            var result = await response.Content.ReadFromJsonAsync<StatsResponse>(JsonConverterFactory.SnakeCaseOptions, cancellationToken: token);
+            var result = await response.Content.ReadFromJsonAsync<StatsResponse>(JsonConverterFactory.StatsResponseOptions, cancellationToken: token);
             return result?.ToStats();
         }
         await HandleResponseAsync(response);

Iggy_SDK/IggyClient/Implementations/TcpMessageStream.cs

@@ -381,7 +381,8 @@ public async IAsyncEnumerable<MessageResponse<TMessage>> PollMessagesAsync<TMess
         {
             StoreOffset.Never => false,
             StoreOffset.WhenMessagesAreReceived => true,
-            StoreOffset.AfterProcessingEachMessage => false
+            StoreOffset.AfterProcessingEachMessage => false,
+            _ => throw new ArgumentOutOfRangeException()
         };
         var fetchRequest = new MessageFetchRequest
         {

Iggy_SDK/JsonConfiguration/CreateTopicConverter.cs

@@ -0,0 +1,32 @@
+using Iggy_SDK.Contracts.Http;
+using Iggy_SDK.Extensions;
+using System.Text.Json;
+using System.Text.Json.Serialization;
+
+namespace Iggy_SDK.JsonConfiguration;
+
+internal sealed class CreateTopicConverter : JsonConverter<TopicRequest>
+{
+    public override TopicRequest? Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options)
+    {
+        throw new NotImplementedException();
+    }
+
+    public override void Write(Utf8JsonWriter writer, TopicRequest value, JsonSerializerOptions options)
+    {
+        writer.WriteStartObject();
+        if (value.TopicId is not null)
+        {
+            writer.WriteNumber(nameof(value.TopicId).ToSnakeCase(), (int)value.TopicId);
+        }
+        writer.WriteString(nameof(value.Name).ToSnakeCase(), value.Name);
+        if (value.MessageExpiry is not null)
+        {
+            writer.WriteNumber(nameof(value.MessageExpiry).ToSnakeCase(), (int)value.MessageExpiry);
+        }
+        writer.WriteNumber(nameof(value.PartitionsCount).ToSnakeCase(), value.PartitionsCount);
+        writer.WriteString(nameof(value.MaxTopicSize).ToSnakeCase(), value.MaxTopicSize.ToString());
+        writer.WriteNumber(nameof(value.ReplicationFactor).ToSnakeCase(), value.ReplicationFactor);
+        writer.WriteEndObject();
+    }
+}

Iggy_SDK/JsonConfiguration/JsonConverterFactory.cs

@@ -1,4 +1,3 @@
-using Iggy_SDK.Contracts.Http;
 using System.Text.Json;
 using System.Text.Json.Serialization;
 using System.Text.Json.Serialization.Metadata;
@@ -68,6 +67,22 @@ public static JsonSerializerOptions TopicResponseOptions
                 new TopicResponseConverter()
             }
         };
+    public static JsonSerializerOptions StatsResponseOptions
+        => new()
+        {
+            Converters =
+            {
+                new StatsResponseConverter()
+            }
+        };
+    public static JsonSerializerOptions CreateTopicOptions
+        => new()
+        {
+            Converters =
+            {
+                new CreateTopicConverter()
+            }
+        };
 
     public static JsonSerializerOptions MessageResponseOptions(Func<byte[], byte[]>? decryptor)
         => new()

Iggy_SDK/JsonConfiguration/MessageResponseConverter.cs

@@ -65,10 +65,27 @@ public override PolledMessages Read(ref Utf8JsonReader reader, Type typeToConver
                             "uint32" => HeaderKind.Uint32,
                             "uint64" => HeaderKind.Uint64,
                             "uint128" => HeaderKind.Uint128,
-                            "float" => HeaderKind.Float,
-                            "double" => HeaderKind.Double,
+                            "float32" => HeaderKind.Float,
+                            "float64" => HeaderKind.Double,
                             "string" => HeaderKind.String,
                             "raw" => HeaderKind.Raw,
+                            /*
+                            "raw" => Ok(HeaderKind::Raw),
+                            "string" => Ok(HeaderKind::String),
+                            "bool" => Ok(HeaderKind::Bool),
+                            "int8" => Ok(HeaderKind::Int8),
+                            "int16" => Ok(HeaderKind::Int16),
+                            "int32" => Ok(HeaderKind::Int32),
+                            "int64" => Ok(HeaderKind::Int64),
+                            "int128" => Ok(HeaderKind::Int128),
+                            "uint8" => Ok(HeaderKind::Uint8),
+                            "uint16" => Ok(HeaderKind::Uint16),
+                            "uint32" => Ok(HeaderKind::Uint32),
+                            "uint64" => Ok(HeaderKind::Uint64),
+                            "uint128" => Ok(HeaderKind::Uint128),
+                            "float32" => Ok(HeaderKind::Float32),
+                            "float64" => Ok(HeaderKind::Float64),
+                            */
                             _ => throw new ArgumentOutOfRangeException()
                         },
                         Value = headerValue

Iggy_SDK/JsonConfiguration/PersonAccessTokenResponseConverter.cs

@@ -22,7 +22,7 @@ public sealed class PersonalAccessTokenResponseConverter : JsonConverter<Persona
         };
         return new PersonalAccessTokenResponse
         {
-            Name = name,
+            Name = name!,
             Expiry = expiry
         };
     }

Iggy_SDK/JsonConfiguration/StatsResponseConverter.cs

@@ -0,0 +1,149 @@
+using Iggy_SDK.Contracts.Http;
+using Iggy_SDK.Extensions;
+using System.ComponentModel;
+using System.Text.Json;
+using System.Text.Json.Serialization;
+
+namespace Iggy_SDK.JsonConfiguration;
+
+public class StatsResponseConverter : JsonConverter<StatsResponse>
+{
+    public override StatsResponse? Read(ref Utf8JsonReader reader, Type typeToConvert, JsonSerializerOptions options)
+    {
+        using var doc = JsonDocument.ParseValue(ref reader);
+        var root = doc.RootElement;
+
+        //int processId = root.GetProperty(nameof(Stats.ProcessId).ToSnakeCase()).GetInt32();
+        int processId = root.GetProperty(nameof(Stats.ProcessId).ToSnakeCase()).GetInt32();
+        float cpuUsage = root.GetProperty(nameof(Stats.CpuUsage).ToSnakeCase()).GetSingle();
+        float totalCpuUsage = root.GetProperty(nameof(Stats.TotalCpuUsage).ToSnakeCase()).GetSingle();
+        string? memoryUsageString = root.GetProperty(nameof(Stats.MemoryUsage).ToSnakeCase()).GetString();
+        string[] memoryUsageStringSplit = memoryUsageString.Split(' ');
+        (ulong memoryUsageBytesVal, string memoryUnit) = (ulong.Parse(memoryUsageStringSplit[0]), memoryUsageStringSplit[1]);
+        ulong memoryUsage = memoryUnit switch
+        {
+            "B" => memoryUsageBytesVal,
+            "KiB" => memoryUsageBytesVal * (ulong)1e03,
+            "MiB" => memoryUsageBytesVal * (ulong)1e06,
+            "GiB" => memoryUsageBytesVal * (ulong)1e09,
+            "TiB" => memoryUsageBytesVal * (ulong)1e12,
+            _ => throw new InvalidEnumArgumentException("Error Wrong Unit when deserializing MemoryUsage")
+        };
+        string? totalMemoryString = root.GetProperty(nameof(Stats.TotalMemory).ToSnakeCase()).GetString();
+        string[] totalMemoryStringSplit = totalMemoryString.Split(' ');
+        (ulong totalMemoryUsageBytesVal, string totalMemoryUnit) = (ulong.Parse(totalMemoryStringSplit[0]), totalMemoryStringSplit[1]);
+        ulong totalMemoryUsage = totalMemoryUnit switch
+        {
+            "B" => totalMemoryUsageBytesVal,
+            "KiB" => totalMemoryUsageBytesVal * (ulong)1e03,
+            "MiB" => totalMemoryUsageBytesVal * (ulong)1e06,
+            "GiB" => totalMemoryUsageBytesVal * (ulong)1e09,
+            "TiB" => totalMemoryUsageBytesVal * (ulong)1e12,
+            _ => throw new InvalidEnumArgumentException("Error Wrong Unit when deserializing TotalMemoryUsage")
+        };
+        string? availableMemoryString = root.GetProperty(nameof(Stats.AvailableMemory).ToSnakeCase()).GetString();
+        string[] availableMemoryStringSplit = availableMemoryString.Split(' ');
+        (ulong availableMemoryBytesVal, string availableMemoryUnit) = (ulong.Parse(availableMemoryStringSplit[0]), availableMemoryStringSplit[1]);
+        ulong availableMemory = availableMemoryUnit switch
+        {
+            "B" => availableMemoryBytesVal,
+            "KiB" => availableMemoryBytesVal * (ulong)1e03,
+            "MiB" => availableMemoryBytesVal * (ulong)1e06,
+            "GiB" => availableMemoryBytesVal * (ulong)1e09,
+            "TiB" => availableMemoryBytesVal * (ulong)1e12,
+            _ => throw new InvalidEnumArgumentException("Error Wrong Unit when deserializing AvailableMemory")
+        };
+        var runtimeRoot = root.GetProperty(nameof(Stats.RunTime).ToSnakeCase());
+        ulong runtimeNanos = 0;
+        foreach (var runtimeProp in runtimeRoot.EnumerateObject())
+        {
+            var durationProp = runtimeProp.Value;
+            var props = durationProp.EnumerateObject();
+            ulong secsValue = props.First().Value.GetUInt64();
+            ulong nanosValue = props.Last().Value.GetUInt64();
+            nanosValue += secsValue * (ulong)1e09;
+            runtimeNanos = nanosValue;
+        }
+
+        ulong startTime = root.GetProperty(nameof(Stats.StartTime).ToSnakeCase()).GetUInt64();
+        string? readBytesString = root.GetProperty(nameof(Stats.ReadBytes).ToSnakeCase()).GetString();
+        string[] readBytesStringSplit = readBytesString.Split(' ');
+        (ulong readBytesVal, string readBytesUnit) = (ulong.Parse(readBytesStringSplit[0]), readBytesStringSplit[1]);
+        ulong readBytes = readBytesUnit switch
+        {
+            "B" => readBytesVal,
+            "KB" => readBytesVal * (ulong)1e03,
+            "MB" => readBytesVal * (ulong)1e06,
+            "GB" => readBytesVal * (ulong)1e09,
+            "TB" => readBytesVal * (ulong)1e12,
+            _ => throw new InvalidEnumArgumentException("Error Wrong Unit when deserializing ReadBytes")
+        };
+        string? writtenBytesString = root.GetProperty(nameof(Stats.WrittenBytes).ToSnakeCase()).GetString();
+        string[] writtenBytesStringSplit = writtenBytesString.Split(' ');
+        (ulong writtenBytesVal, string writtenBytesUnit) = (ulong.Parse(writtenBytesStringSplit[0]), writtenBytesStringSplit[1]);
+        ulong writtenBytes = writtenBytesUnit switch
+        {
+            "B" => writtenBytesVal,
+            "KiB" => writtenBytesVal * (ulong)1e03,
+            "MiB" => writtenBytesVal * (ulong)1e06,
+            "GiB" => writtenBytesVal * (ulong)1e09,
+            "TiB" => writtenBytesVal * (ulong)1e12,
+            _ => throw new InvalidEnumArgumentException("Error Wrong Unit when deserializing WrittenBytes")
+        };
+        string? messageWrittenSizeBytesString = root.GetProperty(nameof(Stats.MessagesSizeBytes).ToSnakeCase()).GetString();
+        string[] messageWrittenSizeBytesStringSplit = messageWrittenSizeBytesString.Split(' ');
+        (ulong messageWrittenSizeBytesVal, string messageWrittenSizeBytesUnit) = (ulong.Parse(messageWrittenSizeBytesStringSplit[0]), messageWrittenSizeBytesStringSplit[1]);
+        ulong messageWrittenSizeBytes = messageWrittenSizeBytesUnit switch
+        {
+            "B" => messageWrittenSizeBytesVal,
+            "KB" => messageWrittenSizeBytesVal * (ulong)1e03,
+            "MB" => messageWrittenSizeBytesVal * (ulong)1e06,
+            "GB" => messageWrittenSizeBytesVal * (ulong)1e09,
+            "TB" => messageWrittenSizeBytesVal * (ulong)1e12,
+            _ => throw new InvalidEnumArgumentException("Error Wrong Unit when deserializing MessageWrittenBytes")
+        };
+        int streamsCount = root.GetProperty(nameof(Stats.StreamsCount).ToSnakeCase()).GetInt32();
+        int topicsCount = root.GetProperty(nameof(Stats.TopicsCount).ToSnakeCase()).GetInt32();
+        int partitionsCount = root.GetProperty(nameof(Stats.PartitionsCount).ToSnakeCase()).GetInt32();
+        int segmentsCount = root.GetProperty(nameof(Stats.SegmentsCount).ToSnakeCase()).GetInt32();
+        ulong messagesCount = root.GetProperty(nameof(Stats.MessagesCount).ToSnakeCase()).GetUInt64();
+        int clientsCount = root.GetProperty(nameof(Stats.ClientsCount).ToSnakeCase()).GetInt32();
+        int consumerGroupsCount = root.GetProperty(nameof(Stats.ConsumerGroupsCount).ToSnakeCase()).GetInt32();
+        string? hostname = root.GetProperty(nameof(Stats.Hostname).ToSnakeCase()).GetString();
+        string? osName = root.GetProperty(nameof(Stats.OsName).ToSnakeCase()).GetString();
+        string? osVersion = root.GetProperty(nameof(Stats.OsVersion).ToSnakeCase()).GetString();
+        string? kernelVersion = root.GetProperty(nameof(Stats.KernelVersion).ToSnakeCase()).GetString();
+
+        return new StatsResponse
+        {
+            AvailableMemory = availableMemory,
+            ClientsCount = clientsCount,
+            ConsumerGroupsCount = consumerGroupsCount,
+            CpuUsage = cpuUsage,
+            Hostname = hostname,
+            KernelVersion = kernelVersion,
+            MemoryUsage = memoryUsage,
+            MessagesCount = messagesCount,
+            MessagesSizeBytes = messageWrittenSizeBytes,
+            OsName = osName,
+            OsVersion = osVersion,
+            PartitionsCount = partitionsCount,
+            ProcessId = processId,
+            ReadBytes = readBytes,
+            RunTime = runtimeNanos,
+            SegmentsCount = segmentsCount,
+            StartTime = startTime,
+            StreamsCount = streamsCount,
+            TopicsCount = topicsCount,
+            TotalCpuUsage = totalCpuUsage,
+            TotalMemory = totalMemoryUsage,
+            WrittenBytes = writtenBytes
+        };
+
+    }
+
+    public override void Write(Utf8JsonWriter writer, StatsResponse value, JsonSerializerOptions options)
+    {
+        throw new NotImplementedException();
+    }
+}

Iggy_SDK/JsonConfiguration/StreamResponseConverter.cs

@@ -13,9 +13,20 @@ public sealed class StreamResponseConverter : JsonConverter<StreamResponse>
         var root = doc.RootElement;
 
         var id = root.GetProperty(nameof(StreamResponse.Id).ToSnakeCase()).GetInt32();
-        var name = root.GetProperty(nameof(StreamResponse.Name).ToSnakeCase()).GetString();
         var createdAt = root.GetProperty(nameof(StreamResponse.CreatedAt).ToSnakeCase()).GetUInt64();
-        var sizeBytes = root.GetProperty(nameof(StreamResponse.SizeBytes).ToSnakeCase()).GetUInt64();
+        var name = root.GetProperty(nameof(StreamResponse.Name).ToSnakeCase()).GetString();
+        var sizeBytesString = root.GetProperty(nameof(StreamResponse.SizeBytes).ToSnakeCase()).GetString();
+        var sizeBytesStringSplit = sizeBytesString.Split(' ');
+        var (sizeBytesVal, Unit) = (ulong.Parse(sizeBytesStringSplit[0]), sizeBytesStringSplit[1]);
+        var sizeBytes = Unit switch
+        {
+            "B" => sizeBytesVal,
+            "KB" => sizeBytesVal * (ulong)1e03,
+            "MB" => sizeBytesVal * (ulong)1e06,
+            "GB" => sizeBytesVal * (ulong)1e09,
+            "TB" => sizeBytesVal * (ulong)1e12,
+            _ => throw new InvalidEnumArgumentException("Error Wrong Unit when deserializing SizeBytes")
+        };
         var messagesCount = root.GetProperty(nameof(StreamResponse.MessagesCount).ToSnakeCase()).GetUInt64();
         var topicsCount = root.GetProperty(nameof(StreamResponse.TopicsCount).ToSnakeCase()).GetInt32();
         root.TryGetProperty(nameof(StreamResponse.Topics).ToSnakeCase(), out var topicsProperty);