Support nullable memory too

verdie-g · verdie-g · commit 5a28fb7a2a0f · 2024-09-13T19:10:53.000-04:00
diff --git a/src/Confluent.Kafka/Producer.cs b/src/Confluent.Kafka/Producer.cs
@@ -512,7 +512,10 @@ private void InitializeSerializers(
                 {
                     this.keySerializer = (ISerializer<TKey>)serializer;
                 }
-                else if (typeof(TValue) == typeof(Memory<byte>) || typeof(TValue) == typeof(ReadOnlyMemory<byte>))
+                else if (typeof(TKey) == typeof(Memory<byte>)
+                         || typeof(TKey) == typeof(ReadOnlyMemory<byte>)
+                         || typeof(TKey) == typeof(Memory<byte>?)
+                         || typeof(TKey) == typeof(ReadOnlyMemory<byte>?))
                 {
                     // Serializers are not used for Memory<byte>.
                 }
@@ -542,7 +545,10 @@ private void InitializeSerializers(
                 {
                     this.valueSerializer = (ISerializer<TValue>)serializer;
                 }
-                else if (typeof(TValue) == typeof(Memory<byte>) || typeof(TValue) == typeof(ReadOnlyMemory<byte>))
+                else if (typeof(TValue) == typeof(Memory<byte>)
+                         || typeof(TValue) == typeof(ReadOnlyMemory<byte>)
+                         || typeof(TValue) == typeof(Memory<byte>?)
+                         || typeof(TValue) == typeof(ReadOnlyMemory<byte>?))
                 {
                     // Serializers are not used for Memory<byte>.
                 }
@@ -765,24 +771,27 @@ public async Task<DeliveryResult<TKey, TValue>> ProduceAsync(
         {
             Headers headers = message.Headers ?? new Headers();
 
-            ReadOnlyMemory<byte>? keyBytes;
+            ReadOnlyMemory<byte>? keyBytes = null;
             try
             {
-                if (message.Key is Memory<byte> memory)
+                if (keySerializer != null)
+                {
+                    byte[] keyBytesArray = keySerializer.Serialize(message.Key, new SerializationContext(MessageComponentType.Key, topicPartition.Topic, headers));
+                    keyBytes = keyBytesArray == null ? (ReadOnlyMemory<byte>?)null : keyBytesArray;
+                }
+                else if (asyncKeySerializer != null)
+                {
+                    byte[] keyBytesArray = await asyncKeySerializer.SerializeAsync(message.Key, new SerializationContext(MessageComponentType.Key, topicPartition.Topic, headers)).ConfigureAwait(false);
+                    keyBytes = keyBytesArray == null ? (ReadOnlyMemory<byte>?)null : keyBytesArray;
+                }
+                else if (message.Key is Memory<byte> memory)
                 {
                     keyBytes = memory;
                 }
                 else if (message.Key is ReadOnlyMemory<byte> readOnlyMemory)
                 {
                     keyBytes = readOnlyMemory;
                 }
-                else
-                {
-                    byte[] keyBytesArray = keySerializer != null
-                        ? keySerializer.Serialize(message.Key, new SerializationContext(MessageComponentType.Key, topicPartition.Topic, headers))
-                        : await asyncKeySerializer.SerializeAsync(message.Key, new SerializationContext(MessageComponentType.Key, topicPartition.Topic, headers)).ConfigureAwait(false);
-                    keyBytes = keyBytesArray == null ? (ReadOnlyMemory<byte>?)null : keyBytesArray;
-                }
             }
             catch (Exception ex)
             {
@@ -796,24 +805,27 @@ public async Task<DeliveryResult<TKey, TValue>> ProduceAsync(
                     ex);
             }
 
-            ReadOnlyMemory<byte>? valBytes;
+            ReadOnlyMemory<byte>? valBytes = null;
             try
             {
-                if (message.Value is Memory<byte> memory)
+                if (valueSerializer != null)
+                {
+                    byte[] valueBytesArray = valueSerializer.Serialize(message.Value, new SerializationContext(MessageComponentType.Value, topicPartition.Topic, headers));
+                    valBytes = valueBytesArray == null ? (ReadOnlyMemory<byte>?)null : valueBytesArray;
+                }
+                else if (asyncValueSerializer != null)
+                {
+                    byte[] valueBytesArray = await asyncValueSerializer.SerializeAsync(message.Value, new SerializationContext(MessageComponentType.Value, topicPartition.Topic, headers)).ConfigureAwait(false);
+                    valBytes = valueBytesArray == null ? (ReadOnlyMemory<byte>?)null : valueBytesArray;
+                }
+                else if (message.Value is Memory<byte> memory)
                 {
                     valBytes = memory;
                 }
                 else if (message.Value is ReadOnlyMemory<byte> readOnlyMemory)
                 {
                     valBytes = readOnlyMemory;
                 }
-                else
-                {
-                    byte[] valBytesArray = valueSerializer != null
-                        ? valueSerializer.Serialize(message.Value, new SerializationContext(MessageComponentType.Value, topicPartition.Topic, headers))
-                        : await asyncValueSerializer.SerializeAsync(message.Value, new SerializationContext(MessageComponentType.Value, topicPartition.Topic, headers)).ConfigureAwait(false);
-                    valBytes = valBytesArray == null ? (ReadOnlyMemory<byte>?)null : valBytesArray;
-                }
             }
             catch (Exception ex)
             {
@@ -912,24 +924,26 @@ public void Produce(
 
             Headers headers = message.Headers ?? new Headers();
 
-            ReadOnlyMemory<byte>? keyBytes;
+            ReadOnlyMemory<byte>? keyBytes = null;
             try
             {
-                if (message.Key is Memory<byte> memory)
+                if (keySerializer != null)
+                {
+                    byte[] keyBytesArray = keySerializer.Serialize(message.Key, new SerializationContext(MessageComponentType.Key, topicPartition.Topic, headers));
+                    keyBytes = keyBytesArray == null ? (ReadOnlyMemory<byte>?)null : keyBytesArray;
+                }
+                else if (asyncKeySerializer != null)
+                {
+                    throw new InvalidOperationException("Produce called with an IAsyncSerializer key serializer configured but an ISerializer is required.");
+                }
+                else if (message.Key is Memory<byte> memory)
                 {
                     keyBytes = memory;
                 }
                 else if (message.Key is ReadOnlyMemory<byte> readOnlyMemory)
                 {
                     keyBytes = readOnlyMemory;
                 }
-                else
-                {
-                    byte[] keyBytesArray = keySerializer != null
-                        ? keySerializer.Serialize(message.Key, new SerializationContext(MessageComponentType.Key, topicPartition.Topic, headers))
-                        : throw new InvalidOperationException("Produce called with an IAsyncSerializer key serializer configured but an ISerializer is required.");
-                    keyBytes = keyBytesArray == null ? (ReadOnlyMemory<byte>?)null : keyBytesArray;
-                }
             }
             catch (Exception ex)
             {
@@ -943,24 +957,26 @@ public void Produce(
                     ex);
             }
 
-            ReadOnlyMemory<byte>? valBytes;
+            ReadOnlyMemory<byte>? valBytes = null;
             try
             {
-                if (message.Value is Memory<byte> memory)
+                if (valueSerializer != null)
+                {
+                    byte[] valueBytesArray = valueSerializer.Serialize(message.Value, new SerializationContext(MessageComponentType.Value, topicPartition.Topic, headers));
+                    valBytes = valueBytesArray == null ? (ReadOnlyMemory<byte>?)null : valueBytesArray;
+                }
+                else if (asyncValueSerializer != null)
+                {
+                    throw new InvalidOperationException("Produce called with an IAsyncSerializer value serializer configured but an ISerializer is required.");
+                }
+                else if (message.Value is Memory<byte> memory)
                 {
                     valBytes = memory;
                 }
                 else if (message.Value is ReadOnlyMemory<byte> readOnlyMemory)
                 {
                     valBytes = readOnlyMemory;
                 }
-                else
-                {
-                    byte[] valBytesArray = valueSerializer != null
-                        ? valueSerializer.Serialize(message.Value, new SerializationContext(MessageComponentType.Value, topicPartition.Topic, headers))
-                        : throw new InvalidOperationException("Produce called with an IAsyncSerializer value serializer configured but an ISerializer is required.");
-                    valBytes = valBytesArray == null ? (ReadOnlyMemory<byte>?)null : valBytesArray;
-                }
             }
             catch (Exception ex)
             {

Original file line number	Diff line number	Diff line change
`@@ -512,7 +512,10 @@ private void InitializeSerializers(`
`512`	`512`	`{`
`513`	`513`	`this.keySerializer = (ISerializer<TKey>)serializer;`
`514`	`514`	`}`
`515`		`- else if (typeof(TValue) == typeof(Memory<byte>) \|\| typeof(TValue) == typeof(ReadOnlyMemory<byte>))`
	`515`	`+ else if (typeof(TKey) == typeof(Memory<byte>)`
	`516`	`+ \|\| typeof(TKey) == typeof(ReadOnlyMemory<byte>)`
	`517`	`+ \|\| typeof(TKey) == typeof(Memory<byte>?)`
	`518`	`+ \|\| typeof(TKey) == typeof(ReadOnlyMemory<byte>?))`
`516`	`519`	`{`
`517`	`520`	`// Serializers are not used for Memory<byte>.`
`518`	`521`	`}`
`@@ -542,7 +545,10 @@ private void InitializeSerializers(`
`542`	`545`	`{`
`543`	`546`	`this.valueSerializer = (ISerializer<TValue>)serializer;`
`544`	`547`	`}`
`545`		`- else if (typeof(TValue) == typeof(Memory<byte>) \|\| typeof(TValue) == typeof(ReadOnlyMemory<byte>))`
	`548`	`+ else if (typeof(TValue) == typeof(Memory<byte>)`
	`549`	`+ \|\| typeof(TValue) == typeof(ReadOnlyMemory<byte>)`
	`550`	`+ \|\| typeof(TValue) == typeof(Memory<byte>?)`
	`551`	`+ \|\| typeof(TValue) == typeof(ReadOnlyMemory<byte>?))`
`546`	`552`	`{`
`547`	`553`	`// Serializers are not used for Memory<byte>.`
`548`	`554`	`}`
`@@ -765,24 +771,27 @@ public async Task<DeliveryResult<TKey, TValue>> ProduceAsync(`
`765`	`771`	`{`
`766`	`772`	`Headers headers = message.Headers ?? new Headers();`
`767`	`773`
`768`		`- ReadOnlyMemory<byte>? keyBytes;`
	`774`	`+ ReadOnlyMemory<byte>? keyBytes = null;`
`769`	`775`	`try`
`770`	`776`	`{`
`771`		`- if (message.Key is Memory<byte> memory)`
	`777`	`+ if (keySerializer != null)`
	`778`	`+ {`
	`779`	`+ byte[] keyBytesArray = keySerializer.Serialize(message.Key, new SerializationContext(MessageComponentType.Key, topicPartition.Topic, headers));`
	`780`	`+ keyBytes = keyBytesArray == null ? (ReadOnlyMemory<byte>?)null : keyBytesArray;`
	`781`	`+ }`
	`782`	`+ else if (asyncKeySerializer != null)`
	`783`	`+ {`
	`784`	`+ byte[] keyBytesArray = await asyncKeySerializer.SerializeAsync(message.Key, new SerializationContext(MessageComponentType.Key, topicPartition.Topic, headers)).ConfigureAwait(false);`
	`785`	`+ keyBytes = keyBytesArray == null ? (ReadOnlyMemory<byte>?)null : keyBytesArray;`
	`786`	`+ }`
	`787`	`+ else if (message.Key is Memory<byte> memory)`
`772`	`788`	`{`
`773`	`789`	`keyBytes = memory;`
`774`	`790`	`}`
`775`	`791`	`else if (message.Key is ReadOnlyMemory<byte> readOnlyMemory)`
`776`	`792`	`{`
`777`	`793`	`keyBytes = readOnlyMemory;`
`778`	`794`	`}`
`779`		`- else`
`780`		`- {`
`781`		`- byte[] keyBytesArray = keySerializer != null`
`782`		`- ? keySerializer.Serialize(message.Key, new SerializationContext(MessageComponentType.Key, topicPartition.Topic, headers))`
`783`		`- : await asyncKeySerializer.SerializeAsync(message.Key, new SerializationContext(MessageComponentType.Key, topicPartition.Topic, headers)).ConfigureAwait(false);`
`784`		`- keyBytes = keyBytesArray == null ? (ReadOnlyMemory<byte>?)null : keyBytesArray;`
`785`		`- }`
`786`	`795`	`}`
`787`	`796`	`catch (Exception ex)`
`788`	`797`	`{`
`@@ -796,24 +805,27 @@ public async Task<DeliveryResult<TKey, TValue>> ProduceAsync(`
`796`	`805`	`ex);`
`797`	`806`	`}`
`798`	`807`
`799`		`- ReadOnlyMemory<byte>? valBytes;`
	`808`	`+ ReadOnlyMemory<byte>? valBytes = null;`
`800`	`809`	`try`
`801`	`810`	`{`
`802`		`- if (message.Value is Memory<byte> memory)`
	`811`	`+ if (valueSerializer != null)`
	`812`	`+ {`
	`813`	`+ byte[] valueBytesArray = valueSerializer.Serialize(message.Value, new SerializationContext(MessageComponentType.Value, topicPartition.Topic, headers));`
	`814`	`+ valBytes = valueBytesArray == null ? (ReadOnlyMemory<byte>?)null : valueBytesArray;`
	`815`	`+ }`
	`816`	`+ else if (asyncValueSerializer != null)`
	`817`	`+ {`
	`818`	`+ byte[] valueBytesArray = await asyncValueSerializer.SerializeAsync(message.Value, new SerializationContext(MessageComponentType.Value, topicPartition.Topic, headers)).ConfigureAwait(false);`
	`819`	`+ valBytes = valueBytesArray == null ? (ReadOnlyMemory<byte>?)null : valueBytesArray;`
	`820`	`+ }`
	`821`	`+ else if (message.Value is Memory<byte> memory)`
`803`	`822`	`{`
`804`	`823`	`valBytes = memory;`
`805`	`824`	`}`
`806`	`825`	`else if (message.Value is ReadOnlyMemory<byte> readOnlyMemory)`
`807`	`826`	`{`
`808`	`827`	`valBytes = readOnlyMemory;`
`809`	`828`	`}`
`810`		`- else`
`811`		`- {`
`812`		`- byte[] valBytesArray = valueSerializer != null`
`813`		`- ? valueSerializer.Serialize(message.Value, new SerializationContext(MessageComponentType.Value, topicPartition.Topic, headers))`
`814`		`- : await asyncValueSerializer.SerializeAsync(message.Value, new SerializationContext(MessageComponentType.Value, topicPartition.Topic, headers)).ConfigureAwait(false);`
`815`		`- valBytes = valBytesArray == null ? (ReadOnlyMemory<byte>?)null : valBytesArray;`
`816`		`- }`
`817`	`829`	`}`
`818`	`830`	`catch (Exception ex)`
`819`	`831`	`{`
`@@ -912,24 +924,26 @@ public void Produce(`
`912`	`924`
`913`	`925`	`Headers headers = message.Headers ?? new Headers();`
`914`	`926`
`915`		`- ReadOnlyMemory<byte>? keyBytes;`
	`927`	`+ ReadOnlyMemory<byte>? keyBytes = null;`
`916`	`928`	`try`
`917`	`929`	`{`
`918`		`- if (message.Key is Memory<byte> memory)`
	`930`	`+ if (keySerializer != null)`
	`931`	`+ {`
	`932`	`+ byte[] keyBytesArray = keySerializer.Serialize(message.Key, new SerializationContext(MessageComponentType.Key, topicPartition.Topic, headers));`
	`933`	`+ keyBytes = keyBytesArray == null ? (ReadOnlyMemory<byte>?)null : keyBytesArray;`
	`934`	`+ }`
	`935`	`+ else if (asyncKeySerializer != null)`
	`936`	`+ {`
	`937`	`+ throw new InvalidOperationException("Produce called with an IAsyncSerializer key serializer configured but an ISerializer is required.");`
	`938`	`+ }`
	`939`	`+ else if (message.Key is Memory<byte> memory)`
`919`	`940`	`{`
`920`	`941`	`keyBytes = memory;`
`921`	`942`	`}`
`922`	`943`	`else if (message.Key is ReadOnlyMemory<byte> readOnlyMemory)`
`923`	`944`	`{`
`924`	`945`	`keyBytes = readOnlyMemory;`
`925`	`946`	`}`
`926`		`- else`
`927`		`- {`
`928`		`- byte[] keyBytesArray = keySerializer != null`
`929`		`- ? keySerializer.Serialize(message.Key, new SerializationContext(MessageComponentType.Key, topicPartition.Topic, headers))`
`930`		`- : throw new InvalidOperationException("Produce called with an IAsyncSerializer key serializer configured but an ISerializer is required.");`
`931`		`- keyBytes = keyBytesArray == null ? (ReadOnlyMemory<byte>?)null : keyBytesArray;`
`932`		`- }`
`933`	`947`	`}`
`934`	`948`	`catch (Exception ex)`
`935`	`949`	`{`
`@@ -943,24 +957,26 @@ public void Produce(`
`943`	`957`	`ex);`
`944`	`958`	`}`
`945`	`959`
`946`		`- ReadOnlyMemory<byte>? valBytes;`
	`960`	`+ ReadOnlyMemory<byte>? valBytes = null;`
`947`	`961`	`try`
`948`	`962`	`{`
`949`		`- if (message.Value is Memory<byte> memory)`
	`963`	`+ if (valueSerializer != null)`
	`964`	`+ {`
	`965`	`+ byte[] valueBytesArray = valueSerializer.Serialize(message.Value, new SerializationContext(MessageComponentType.Value, topicPartition.Topic, headers));`
	`966`	`+ valBytes = valueBytesArray == null ? (ReadOnlyMemory<byte>?)null : valueBytesArray;`
	`967`	`+ }`
	`968`	`+ else if (asyncValueSerializer != null)`
	`969`	`+ {`
	`970`	`+ throw new InvalidOperationException("Produce called with an IAsyncSerializer value serializer configured but an ISerializer is required.");`
	`971`	`+ }`
	`972`	`+ else if (message.Value is Memory<byte> memory)`
`950`	`973`	`{`
`951`	`974`	`valBytes = memory;`
`952`	`975`	`}`
`953`	`976`	`else if (message.Value is ReadOnlyMemory<byte> readOnlyMemory)`
`954`	`977`	`{`
`955`	`978`	`valBytes = readOnlyMemory;`
`956`	`979`	`}`
`957`		`- else`
`958`		`- {`
`959`		`- byte[] valBytesArray = valueSerializer != null`
`960`		`- ? valueSerializer.Serialize(message.Value, new SerializationContext(MessageComponentType.Value, topicPartition.Topic, headers))`
`961`		`- : throw new InvalidOperationException("Produce called with an IAsyncSerializer value serializer configured but an ISerializer is required.");`
`962`		`- valBytes = valBytesArray == null ? (ReadOnlyMemory<byte>?)null : valBytesArray;`
`963`		`- }`
`964`	`980`	`}`
`965`	`981`	`catch (Exception ex)`
`966`	`982`	`{`