Ensure CustomDataAdapter.cs works in cluster with multiple silos

dotnet · Jan 18, 2025 · 78f31ad · 78f31ad
1 parent 2adfa49
commit 78f31ad
Showing 1 changed file with 10 additions and 8 deletions.
diff --git a/orleans/Streaming/CustomDataAdapter/Silo/CustomDataAdapter.cs b/orleans/Streaming/CustomDataAdapter/Silo/CustomDataAdapter.cs
@@ -31,31 +31,33 @@ protected override IBatchContainer GetBatchContainer(EventHubMessage eventHubMes
         => new CustomBatchContainer(eventHubMessage);
 }
 
-public class CustomBatchContainer : IBatchContainer
+[GenerateSerializer, Immutable]
+public sealed class CustomBatchContainer : IBatchContainer
 {
-    public StreamId StreamId { get; }
+    [Id(0)]
+    private readonly EventHubMessage _eventHubMessage;
 
+    [Id(1)]
     public StreamSequenceToken SequenceToken { get; }
 
-    private readonly byte[] _payload;
+    public StreamId StreamId => _eventHubMessage.StreamId;
 
     public CustomBatchContainer(EventHubMessage eventHubMessage)
     {
-        StreamId = eventHubMessage.StreamId;
-        SequenceToken = new EventHubSequenceTokenV2(eventHubMessage.Offset, eventHubMessage.SequenceNumber, 0);
-        _payload = eventHubMessage.Payload;
+        _eventHubMessage = eventHubMessage;
+        SequenceToken = new EventHubSequenceTokenV2(_eventHubMessage.Offset, _eventHubMessage.SequenceNumber, 0);
     }
 
     public IEnumerable<Tuple<T, StreamSequenceToken>> GetEvents<T>()
     {
         try
         {
-            var evt = JsonSerializer.Deserialize<T>(_payload)!;
+            var evt = JsonSerializer.Deserialize<T>(_eventHubMessage.Payload)!;
             return new[] { Tuple.Create(evt, SequenceToken) };
         }
         catch (Exception)
         {
-            return new List<Tuple<T, StreamSequenceToken>>();
+            return Array.Empty<Tuple<T, StreamSequenceToken>>();
         }
     }