using Orleans.Streams; using StackExchange.Redis; using System.Text.Json; namespace Universley.OrleansContrib.StreamsProvider.Redis { public class RedisStreamBatchContainer : IBatchContainer { public StreamId StreamId { get; } public StreamSequenceToken SequenceToken { get; } public StreamEntry StreamEntry { get; } public RedisStreamBatchContainer(StreamEntry streamEntry) { StreamEntry = streamEntry; var streamNamespace = StreamEntry.Values[0].Value; var steamKey = StreamEntry.Values[1].Value; StreamId = StreamId.Create(streamNamespace!, steamKey!); SequenceToken = new RedisStreamSequenceToken(StreamEntry.Id); } public IEnumerable> GetEvents() { List> events = new(); var eventType = typeof(T).Name; if (eventType == StreamEntry.Values[2].Value) { var data = StreamEntry.Values[3].Value; var @event = JsonSerializer.Deserialize(data!); events.Add(new(@event!, SequenceToken)); } return events; } public bool ImportRequestContext() { return false; } } }