feat: update message type resolver methods to return ValueTask instea…

…d of Task
Farfetch · Sep 15, 2023 · 332fbd8 · 332fbd8
1 parent 684338e
commit 332fbd8
Show file tree

Hide file tree

Showing 6 changed files with 19 additions and 13 deletions.
diff --git a/src/KafkaFlow.SchemaRegistry/SchemaRegistryTypeResolver.cs b/src/KafkaFlow.SchemaRegistry/SchemaRegistryTypeResolver.cs
@@ -28,7 +28,7 @@ public SchemaRegistryTypeResolver(ISchemaRegistryTypeNameResolver typeNameResolv
         }
 
         /// <inheritdoc />
-        public async Task<Type> OnConsumeAsync(IMessageContext context)
+        public async ValueTask<Type> OnConsumeAsync(IMessageContext context)
         {
             var schemaId = BinaryPrimitives.ReadInt32BigEndian(
                 ((byte[]) context.Message.Value).AsSpan().Slice(1, 4));
@@ -61,6 +61,6 @@ public async Task<Type> OnConsumeAsync(IMessageContext context)
         }
 
         /// <inheritdoc />
-        public Task OnProduceAsync(IMessageContext context) => Task.CompletedTask;
+        public ValueTask OnProduceAsync(IMessageContext context) => default(ValueTask);
     }
 }
diff --git a/src/KafkaFlow.Serializer/DefaultTypeResolver.cs b/src/KafkaFlow.Serializer/DefaultTypeResolver.cs
@@ -7,18 +7,20 @@ internal class DefaultTypeResolver : IMessageTypeResolver
     {
         private const string MessageType = "Message-Type";
 
-        public Task<Type> OnConsumeAsync(IMessageContext context)
+        public ValueTask<Type> OnConsumeAsync(IMessageContext context)
         {
             var typeName = context.Headers.GetString(MessageType);
 
-            return typeName is null ? null : Task.FromResult(Type.GetType(typeName));
+            return typeName is null ?
+                new ValueTask<Type>((Type) null) :
+                new ValueTask<Type>(Type.GetType(typeName));
         }
 
-        public Task OnProduceAsync(IMessageContext context)
+        public ValueTask OnProduceAsync(IMessageContext context)
         {
             if (context.Message.Value is null)
             {
-                return Task.CompletedTask;
+                return default(ValueTask);
             }
 
             var messageType = context.Message.Value.GetType();
@@ -27,7 +29,7 @@ public Task OnProduceAsync(IMessageContext context)
                 MessageType,
                 $"{messageType.FullName}, {messageType.Assembly.GetName().Name}");
 
-            return Task.CompletedTask;
+            return default(ValueTask);
         }
     }
 }
diff --git a/src/KafkaFlow.Serializer/IMessageTypeResolver.cs b/src/KafkaFlow.Serializer/IMessageTypeResolver.cs
@@ -13,12 +13,12 @@ public interface IMessageTypeResolver
         /// </summary>
         /// <param name="context">The <see cref="IMessageContext"/> containing the message and the metadata</param>
         /// <returns></returns>
-        Task<Type> OnConsumeAsync(IMessageContext context);
+        ValueTask<Type> OnConsumeAsync(IMessageContext context);
 
         /// <summary>
         /// Stores the message type somewhere when producing
         /// </summary>
         /// <param name="context">The <see cref="IMessageContext"/> containing the message and the metadata</param>
-        Task OnProduceAsync(IMessageContext context);
+        ValueTask OnProduceAsync(IMessageContext context);
     }
 }
diff --git a/src/KafkaFlow.Serializer/KafkaFlow.Serializer.csproj b/src/KafkaFlow.Serializer/KafkaFlow.Serializer.csproj
@@ -13,6 +13,7 @@
 
     <ItemGroup>
       <PackageReference Include="Microsoft.IO.RecyclableMemoryStream" Version="2.1.3" />
+      <PackageReference Include="System.Threading.Tasks.Extensions" Version="4.5.4" />
     </ItemGroup>
 
 </Project>
diff --git a/src/KafkaFlow.Serializer/SingleMessageTypeResolver.cs b/src/KafkaFlow.Serializer/SingleMessageTypeResolver.cs
@@ -20,13 +20,13 @@ public SingleMessageTypeResolver(Type messageType)
         }
 
         /// <inheritdoc/>
-        public Task<Type> OnConsumeAsync(IMessageContext context) => Task.FromResult(this.messageType);
+        public ValueTask<Type> OnConsumeAsync(IMessageContext context) => new ValueTask<Type>(this.messageType);
 
         /// <inheritdoc/>
-        public Task OnProduceAsync(IMessageContext context)
+        public ValueTask OnProduceAsync(IMessageContext context)
         {
             // Do nothing
-            return Task.CompletedTask;
+            return default(ValueTask);
         }
     }
 }
diff --git a/src/KafkaFlow.UnitTests/Middlewares/Serialization/ConfluentProtobufTypeNameResolverTests.cs b/src/KafkaFlow.UnitTests/Middlewares/Serialization/ConfluentProtobufTypeNameResolverTests.cs
@@ -3,6 +3,9 @@ namespace KafkaFlow.UnitTests.Middlewares.Serialization
     using System;
     using System.Threading.Tasks;
     using Confluent.SchemaRegistry;
+
+    using FluentAssertions;
+
     using Google.Protobuf;
     using KafkaFlow.Serializer.SchemaRegistry;
     using Microsoft.VisualStudio.TestTools.UnitTesting;
@@ -40,7 +43,7 @@ public async Task ResolveAsync_ValidProtobufObject_ReturnsProtoFields()
             var protoFields = await this.schemaRegistryTypeResolver.ResolveAsync(schemaId);
 
             // Assert
-            // TODO fix returning empty
+            protoFields.Should().NotBeNull();
         }
     }
 }