redis stream.cs

public static class Ext
{
    public static IObservable<string> AnonymousConsume(this IDatabase db, string topic, int prefetchCount = 100, CancellationToken? cancellationToken = null)
    {
        async Task<(string[], string lastPos)> SafeRead(string position)
        {
            try
            {
                var msgs = await db.StreamReadAsync(topic, position, prefetchCount);
                var items =  msgs.SelectMany(s => s.Values)
                    .Select(s => s.Value.ToString())
                    .ToArray();


                var lastPos = msgs.LastOrDefault().Id;

                return (items, lastPos);
            }
            catch (Exception e)
            {
                // log here
                return (Array.Empty<string>(), position);
            }
        }

        return Observable.Create(async (IObserver<string> x) =>
        {
            var cts = cancellationToken ?? CancellationToken.None;
            var pos = "0-0";

            while (!cts.IsCancellationRequested)
            {
                var (items, newPos) = await SafeRead(pos);
                foreach (var item in items)
                {
                    x.OnNext(item);
                }

                pos = newPos;
            }
        });
    }


    public static void Example(this IDatabase db)
    {
        var state = new Dictionary<string, dynamic[]>();


        db.AnonymousConsume("topic-name")
            .Synchronize(new object())
            .Select(value => JsonConvert.DeserializeObject<dynamic>(value)) // deserialize to model not dynamic :)
            .Subscribe(item =>
            {
                var gameId = item.GameId;
                var blocks = item.Blocks;

                state[gameId] = blocks;
            });


    }
}