diff --git a/src/Redis/src/Eventuous.Redis/Extensions.cs b/src/Redis/src/Eventuous.Redis/Extensions.cs
new file mode 100644
index 000000000..902c3fb6c
--- /dev/null
+++ b/src/Redis/src/Eventuous.Redis/Extensions.cs
@@ -0,0 +1,20 @@
+using StackExchange.Redis;
+namespace Eventuous.Redis.Extension;
+
+public static class Extensions
+{
+    public static long ToLong(this RedisValue value) {
+        var parts = ((string?)value)?.Split('-');
+        return Convert.ToInt64(parts?[0]) * 10 + Convert.ToInt64(parts?[1]);
+    }
+
+    public static ulong ToULong(this RedisValue value) {
+        var parts = ((string?)value)?.Split('-');
+        return Convert.ToUInt64(parts?[0]) * 10 + Convert.ToUInt64(parts?[1]);
+    }
+
+    public static RedisValue ToRedisValue(this long position) {
+        if (position == 0) return "0-0";
+        return new RedisValue($"{position / 10}-{position % 10}");
+    }
+}
diff --git a/src/Redis/src/Eventuous.Redis/Module.cs b/src/Redis/src/Eventuous.Redis/Module.cs
new file mode 100644
index 000000000..a6151cab8
--- /dev/null
+++ b/src/Redis/src/Eventuous.Redis/Module.cs
@@ -0,0 +1,30 @@
+using System.Reflection;
+using Eventuous.Tools;
+
+namespace Eventuous.Redis;
+
+public class Module {
+
+    static readonly Assembly Assembly = typeof(Module).Assembly;
+    public async Task LoadModule(GetRedisDatabase getDatabase) {
+        var names = Assembly.GetManifestResourceNames()
+            .Where(x => x.EndsWith(".lua"))
+            .OrderBy(x => x);
+
+        var db = getDatabase();
+
+        foreach (var name in names) {
+            await using var stream    = Assembly.GetManifestResourceStream(name);
+            using var       reader    = new StreamReader(stream!);
+            var             script    = await reader.ReadToEndAsync().NoContext();
+
+            try {
+                await db.ExecuteAsync("FUNCTION", "LOAD", "REPLACE", script);
+            }
+            catch (Exception e) {
+                Console.WriteLine(e);
+                if (!e.Message.Contains("'append_events' already exists")) throw;
+            }
+        }
+    }
+}
\ No newline at end of file
diff --git a/src/Redis/src/Eventuous.Redis/RedisStore.cs b/src/Redis/src/Eventuous.Redis/RedisStore.cs
index 3167776df..23c7ac650 100644
--- a/src/Redis/src/Eventuous.Redis/RedisStore.cs
+++ b/src/Redis/src/Eventuous.Redis/RedisStore.cs
@@ -3,26 +3,32 @@
 using System.Text;
 using Eventuous.Diagnostics;
 using StackExchange.Redis;
+using Eventuous.Redis.Extension;
 
 namespace Eventuous.Redis;
 
 public delegate IDatabase GetRedisDatabase();
 
-public class RedisStore : IEventStore {
-    readonly GetRedisDatabase    _getDatabase;
-    readonly IEventSerializer    _serializer;
-    readonly IMetadataSerializer _metaSerializer;
+public record RedisStoreOptions();
+
+public class RedisStore : IEventStore
+{
+    readonly GetRedisDatabase       _getDatabase;
+    readonly IEventSerializer       _serializer;
+    readonly IMetadataSerializer    _metaSerializer;
 
     public RedisStore(
-        GetRedisDatabase     getDatabase,
-        IEventSerializer?    serializer     = null,
-        IMetadataSerializer? metaSerializer = null
-    ) {
+        GetRedisDatabase getDatabase,
+        RedisStoreOptions  options,
+        IEventSerializer?     serializer     = null,
+        IMetadataSerializer?  metaSerializer = null
+    )
+    {
         _serializer     = serializer     ?? DefaultEventSerializer.Instance;
         _metaSerializer = metaSerializer ?? DefaultMetadataSerializer.Instance;
-        _getDatabase    = Ensure.NotNull(getDatabase, "Connection factory");
+        _getDatabase  = Ensure.NotNull(getDatabase, "Connection factory");
+        
     }
-
     const string ContentType = "application/json";
 
     public async Task<StreamEvent[]> ReadEvents(
@@ -31,10 +37,10 @@ public async Task<StreamEvent[]> ReadEvents(
         int                count,
         CancellationToken  cancellationToken
     ) {
-        var result = await _getDatabase().StreamRangeAsync(stream.ToString(), ULongToStreamId(start), null, count);
-        if (result == null) throw new StreamNotFound(stream);
-
-        return result.Select(x => ToStreamEvent(x)).ToArray();
+            var result = await _getDatabase().StreamReadAsync(stream.ToString(), start.Value.ToRedisValue(), count);
+            if (result == null)
+                throw new StreamNotFound(stream);
+            return result.Select(x => ToStreamEvent(x)).ToArray();        
     }
 
     public async Task<AppendEventsResult> AppendEvents(
@@ -70,7 +76,9 @@ CancellationToken                cancellationToken
 
         try {
             var response = (string[]?)await database.ExecuteAsync("FCALL", fCallParams);
-            return new AppendEventsResult(StreamIdToULong(response?[1]), Convert.ToInt64(response?[0]));
+            var streamPosition = response?[0];
+            var globalPosition = response?[1];
+            return new AppendEventsResult(new RedisValue(globalPosition!).ToULong(), Convert.ToInt64(streamPosition));
         }
         catch (Exception e) when (e.Message.Contains("WrongExpectedVersion")) {
             PersistenceEventSource.Log.UnableToAppendEvents(stream, e);
@@ -129,33 +137,7 @@ StreamEvent ToStreamEvent(StreamEntry evt) {
             ),
             _ => throw new Exception("Unknown deserialization result")
         };
-
         StreamEvent AsStreamEvent(object payload)
-            => new(Guid.Parse(evt["message_id"].ToString()), payload, meta ?? new Metadata(), ContentType, StreamIdToLong(evt.Id));
-    }
-
-    /*
-        16761513606580 -> 1676151360658-0
-    */
-    static RedisValue ULongToStreamId(StreamReadPosition position) {
-        if (position == StreamReadPosition.Start) return "0-0";
-
-        return new RedisValue($"{position.Value / 10}-{position.Value % 10}");
-    }
-
-    /*
-        1676151360658-0 -> 16761513606580
-    */
-    static ulong StreamIdToULong(RedisValue value) {
-        var parts = ((string?)value)?.Split('-');
-        return Convert.ToUInt64(parts?[0]) * 10 + Convert.ToUInt64(parts?[1]);
-    }
-
-    /*
-        1676151360658-0 -> 16761513606580
-    */
-    static long StreamIdToLong(RedisValue value) {
-        var parts = ((string?)value)?.Split('-');
-        return Convert.ToInt64(parts?[0]) * 10 + Convert.ToInt64(parts?[1]);
+            => new(Guid.Parse(evt["message_id"].ToString()), payload, meta ?? new Metadata(), ContentType, evt.Id.ToLong());
     }
 }
diff --git a/src/Redis/src/Eventuous.Redis/Subscriptions/PersistentEvent.cs b/src/Redis/src/Eventuous.Redis/Subscriptions/PersistentEvent.cs
new file mode 100644
index 000000000..66b0d84f7
--- /dev/null
+++ b/src/Redis/src/Eventuous.Redis/Subscriptions/PersistentEvent.cs
@@ -0,0 +1,12 @@
+namespace Eventuous.Redis;
+
+public record PersistentEvent(
+    Guid     MessageId,
+    string   MessageType,
+    long      StreamPosition,
+    long     GlobalPosition,
+    string   JsonData,
+    string?  JsonMetadata,
+    DateTime Created,
+    string  StreamName
+);
\ No newline at end of file
diff --git a/src/Redis/src/Eventuous.Redis/Subscriptions/RedisAllStreamSubscription.cs b/src/Redis/src/Eventuous.Redis/Subscriptions/RedisAllStreamSubscription.cs
new file mode 100644
index 000000000..f76883a80
--- /dev/null
+++ b/src/Redis/src/Eventuous.Redis/Subscriptions/RedisAllStreamSubscription.cs
@@ -0,0 +1,59 @@
+using System.Linq;
+using Eventuous.Subscriptions;
+using Eventuous.Subscriptions.Checkpoints;
+using Eventuous.Subscriptions.Context;
+using Eventuous.Subscriptions.Filters;
+using Eventuous.Redis;
+using Microsoft.Extensions.Logging;
+using StackExchange.Redis;
+using Eventuous.Redis.Extension;
+
+namespace Eventuous.Redis.Subscriptions;
+
+public class RedisAllStreamSubscription : RedisSubscriptionBase<RedisSubscriptionBaseOptions> {
+
+    public RedisAllStreamSubscription(
+        GetRedisDatabase                    getDatabase,
+        RedisAllStreamSubscriptionOptions   options,
+        ICheckpointStore                    checkpointStore,
+        ConsumePipe                         consumePipe,
+        ILoggerFactory?                     loggerFactory
+    ) : base (getDatabase, options, checkpointStore, consumePipe, loggerFactory) {
+
+    }
+
+    protected override async Task<PersistentEvent[]> ReadEvents(IDatabase database, long position)
+    {
+        var linkedEvents = await database.StreamReadAsync("_all", position.ToRedisValue(), 100);
+        var persistentEvents = new List<PersistentEvent>();
+
+        foreach(var linkEvent in linkedEvents) {
+            var stream = linkEvent["stream"];
+            var streamPosition = linkEvent["position"];
+
+            var streamEvents = await database.StreamRangeAsync(new RedisKey(stream), streamPosition);
+            var entry = streamEvents[0];
+
+            DateTime date;
+            System.DateTime.TryParse(entry["created"]!, out date);
+            persistentEvents.Add(new PersistentEvent(
+                Guid.Parse(entry["message_id"]!),
+                entry["message_type"]!,
+                entry.Id.ToLong(),
+                entry.Id.ToLong(),
+                entry["json_data"]!,
+                entry["json_metadata"],
+                date,
+                stream!));
+        }
+        return persistentEvents.ToArray();
+    }
+
+    protected override EventPosition GetPositionFromContext(IMessageConsumeContext context) 
+        => EventPosition.FromContext(context);
+
+}
+
+public record RedisAllStreamSubscriptionOptions : RedisSubscriptionBaseOptions;
+
+
diff --git a/src/Redis/src/Eventuous.Redis/Subscriptions/RedisCheckpointStore.cs b/src/Redis/src/Eventuous.Redis/Subscriptions/RedisCheckpointStore.cs
new file mode 100644
index 000000000..d3a74e372
--- /dev/null
+++ b/src/Redis/src/Eventuous.Redis/Subscriptions/RedisCheckpointStore.cs
@@ -0,0 +1,40 @@
+using Eventuous.Subscriptions.Checkpoints;
+using Eventuous.Subscriptions.Logging;
+using Eventuous.Tools;
+using Microsoft.Extensions.Logging;
+
+namespace Eventuous.Redis.Subscriptions;
+
+public class RedisCheckpointStore : ICheckpointStore {
+
+    readonly GetRedisDatabase       _getDatabase;
+    readonly ILoggerFactory?        _loggerFactory;
+
+    public RedisCheckpointStore(GetRedisDatabase getDatabase, ILoggerFactory? loggerFactory) {
+        _getDatabase = getDatabase;
+        _loggerFactory = loggerFactory;
+    }
+
+    public async ValueTask<Checkpoint> GetLastCheckpoint(
+        string checkpointId, 
+        CancellationToken cancellationToken
+    ) { 
+        Checkpoint checkpoint; 
+        Logger.ConfigureIfNull(checkpointId, _loggerFactory);
+        var position = await _getDatabase().StringGetAsync(checkpointId);
+        checkpoint = position.IsNull ? Checkpoint.Empty(checkpointId) : new Checkpoint(checkpointId, Convert.ToUInt64(position));
+        Logger.Current.CheckpointLoaded(this, checkpoint);
+        return checkpoint;
+    }
+
+    public async ValueTask<Checkpoint> StoreCheckpoint(
+        Checkpoint        checkpoint,
+        bool              force,
+        CancellationToken cancellationToken
+    ) {
+        if (checkpoint.Position == null) return checkpoint;
+        await _getDatabase().StringSetAsync(checkpoint.Id, checkpoint.Position);
+        return checkpoint;
+    }
+}
+
diff --git a/src/Redis/src/Eventuous.Redis/Subscriptions/RedisStreamSubscription.cs b/src/Redis/src/Eventuous.Redis/Subscriptions/RedisStreamSubscription.cs
new file mode 100644
index 000000000..01af388e0
--- /dev/null
+++ b/src/Redis/src/Eventuous.Redis/Subscriptions/RedisStreamSubscription.cs
@@ -0,0 +1,61 @@
+using System.Linq;
+using System.Text;
+using Eventuous;
+using Eventuous.Subscriptions;
+using Eventuous.Subscriptions.Checkpoints;
+using Eventuous.Subscriptions.Context;
+using Eventuous.Subscriptions.Filters;
+using Microsoft.Extensions.Logging;
+using StackExchange.Redis;
+using Eventuous.Redis.Extension;
+
+namespace Eventuous.Redis.Subscriptions;
+
+public class RedisStreamSubscription : RedisSubscriptionBase<RedisSubscriptionBaseOptions> {
+
+    public RedisStreamSubscription(
+        GetRedisDatabase                    getDatabase,
+        RedisStreamSubscriptionOptions      options,
+        ICheckpointStore                    checkpointStore,
+        ConsumePipe                         consumePipe,
+        ILoggerFactory?                     loggerFactory
+    ) : base (getDatabase, options, checkpointStore, consumePipe, loggerFactory) {
+        _streamName = options.Stream.ToString();
+    }
+
+    protected override async Task BeforeSubscribe(CancellationToken cancellationToken) {
+        var info     = await GetDatabase().StreamInfoAsync(_streamName);
+        if (info.Length <= 0)
+            throw new StreamNotFound(_streamName);
+    }
+
+    protected override async Task<PersistentEvent[]> ReadEvents(IDatabase database, long position)
+    {
+        var evts = await database.StreamReadAsync(_streamName, position.ToRedisValue(), 100);
+        return evts.Select( evt =>  {
+
+            DateTime date;
+            System.DateTime.TryParse(evt["created"]!, out date);
+            return new PersistentEvent (
+                Guid.Parse(evt["message_id"]!),
+                evt["message_type"]!,
+                evt.Id.ToLong(),
+                evt.Id.ToLong(),
+                evt["json_data"]!,
+                evt["json_metadata"],
+                date,
+                _streamName);
+        }
+        ).ToArray();
+    }
+
+    readonly string _streamName;
+
+    protected override EventPosition GetPositionFromContext(IMessageConsumeContext context) 
+        => EventPosition.FromContext(context);
+
+}
+
+public record RedisStreamSubscriptionOptions(StreamName Stream) : RedisSubscriptionBaseOptions;
+
+
diff --git a/src/Redis/src/Eventuous.Redis/Subscriptions/RedisSubscriptionBase.cs b/src/Redis/src/Eventuous.Redis/Subscriptions/RedisSubscriptionBase.cs
new file mode 100644
index 000000000..ab23e3adf
--- /dev/null
+++ b/src/Redis/src/Eventuous.Redis/Subscriptions/RedisSubscriptionBase.cs
@@ -0,0 +1,126 @@
+using System.Text;
+using Eventuous;
+using Eventuous.Subscriptions;
+using Eventuous.Subscriptions.Checkpoints;
+using Eventuous.Subscriptions.Context;
+using Eventuous.Subscriptions.Filters;
+using Eventuous.Subscriptions.Logging;
+using Eventuous.Tools;
+using Eventuous.Redis;
+using Microsoft.Extensions.Logging;
+using StackExchange.Redis;
+using Eventuous.Redis.Extension;
+
+public abstract class RedisSubscriptionBase<T> : EventSubscriptionWithCheckpoint<T>
+    where T : RedisSubscriptionBaseOptions {
+    readonly IMetadataSerializer     _metaSerializer;
+    readonly CancellationTokenSource _cts = new();
+
+    protected GetRedisDatabase GetDatabase { get; }
+
+    protected RedisSubscriptionBase(
+        GetRedisDatabase      getDatabase,
+        T                     options,
+        ICheckpointStore      checkpointStore,
+        ConsumePipe           consumePipe,
+        ILoggerFactory?       loggerFactory
+    ) : base(options, checkpointStore, consumePipe, options.ConcurrencyLimit, loggerFactory) {
+        GetDatabase   = Ensure.NotNull(getDatabase, "Connection factory");
+        _metaSerializer = DefaultMetadataSerializer.Instance;
+    }
+
+    protected override async ValueTask Subscribe(CancellationToken cancellationToken) {
+        await BeforeSubscribe(cancellationToken).NoContext();
+
+        var (_, position) = await GetCheckpoint(cancellationToken).NoContext();
+
+        _runner = Task.Run(() => PollingQuery(position + 1, _cts.Token), _cts.Token);
+    }
+
+    protected override async ValueTask Unsubscribe(CancellationToken cancellationToken) {
+        try {
+            _cts.Cancel();
+            if (_runner != null) await _runner.NoContext();
+        }
+        catch (OperationCanceledException) {
+            // Nothing to do
+        }
+    }
+
+    protected const string ContentType = "application/json";
+
+    Task? _runner;
+
+    async Task PollingQuery(ulong? position, CancellationToken cancellationToken) {
+        var start = position.HasValue ? (long)position : 0;
+
+        while (!cancellationToken.IsCancellationRequested) {
+            try {
+                
+                var persistentEvents = await ReadEvents(GetDatabase(), start);
+
+                foreach (var persistentEvent in persistentEvents) {
+                    await HandleInternal(ToConsumeContext(persistentEvent, cancellationToken)).NoContext();
+                    start = persistentEvent.StreamPosition + 1;
+                }
+            }
+            catch (OperationCanceledException) {
+                // Nothing to do
+            }
+            catch (Exception e) {
+                IsDropped = true;
+                Log.WarnLog?.Log(e, "Dropped");
+                throw;
+            }
+        }
+    }
+   
+    IMessageConsumeContext ToConsumeContext(PersistentEvent evt, CancellationToken cancellationToken) {
+        Logger.Current = Log;
+        var data = DeserializeData(
+            ContentType,
+            evt.MessageType,
+            Encoding.UTF8.GetBytes(evt.JsonData),
+            evt.StreamName,
+            (ulong)evt.StreamPosition
+        );
+
+        var meta = (evt.JsonMetadata == null)
+            ? new Metadata()
+            : _metaSerializer.Deserialize(Encoding.UTF8.GetBytes(evt.JsonMetadata));
+
+        return AsContext(evt, data, meta, cancellationToken);
+    }
+
+    IMessageConsumeContext AsContext(
+        PersistentEvent     evt,
+        object?             e,
+        Metadata?           meta,
+        CancellationToken   cancellationToken
+    )
+        => new MessageConsumeContext(
+            evt.MessageId.ToString(),
+            evt.MessageType,
+            ContentType,
+            evt.StreamName,
+            (ulong)evt.StreamPosition,
+            (ulong)evt.GlobalPosition,
+            _sequence++,
+            evt.Created,
+            e,
+            meta,
+            Options.SubscriptionId,
+            cancellationToken
+        );
+
+    protected abstract Task<PersistentEvent[]> ReadEvents(IDatabase database, long position);
+    protected virtual Task BeforeSubscribe(CancellationToken cancellationToken) => Task.CompletedTask;
+    
+    ulong           _sequence;
+
+}
+
+public abstract record RedisSubscriptionBaseOptions : SubscriptionOptions {
+    public int    ConcurrencyLimit { get; set; } = 1;
+    public int    MaxPageSize      { get; set; } = 1024;
+}
diff --git a/src/Redis/test/Eventuous.Tests.Redis/Eventuous.Tests.Redis.csproj b/src/Redis/test/Eventuous.Tests.Redis/Eventuous.Tests.Redis.csproj
index 2cdddc753..9394296f8 100644
--- a/src/Redis/test/Eventuous.Tests.Redis/Eventuous.Tests.Redis.csproj
+++ b/src/Redis/test/Eventuous.Tests.Redis/Eventuous.Tests.Redis.csproj
@@ -9,7 +9,6 @@
             <PackageReference Include="MicroElements.AutoFixture.NodaTime" />
             <PackageReference Include="NodaTime.Serialization.SystemTextJson" />
             <PackageReference Include="StackExchange.Redis" />
-            <PackageVersion Include="StackExchange.Redis" Version="2.6.90" />
     </ItemGroup>
     <ItemGroup>
         <ProjectReference Include="$(LocalRoot)\Eventuous.Redis\Eventuous.Redis.csproj" />
diff --git a/src/Redis/test/Eventuous.Tests.Redis/Fixtures/DomainFixture.cs b/src/Redis/test/Eventuous.Tests.Redis/Fixtures/DomainFixture.cs
index 49714f916..e15b0f60a 100644
--- a/src/Redis/test/Eventuous.Tests.Redis/Fixtures/DomainFixture.cs
+++ b/src/Redis/test/Eventuous.Tests.Redis/Fixtures/DomainFixture.cs
@@ -1,6 +1,7 @@
 using Eventuous.Sut.App;
 using MicroElements.AutoFixture.NodaTime;
 using NodaTime;
+using static Eventuous.Tests.Redis.Fixtures.IntegrationFixture;
 
 namespace Eventuous.Tests.Redis.Fixtures;
 
diff --git a/src/Redis/test/Eventuous.Tests.Redis/Fixtures/IntegrationFixture.cs b/src/Redis/test/Eventuous.Tests.Redis/Fixtures/IntegrationFixture.cs
index 7b2add0f2..9979489a5 100644
--- a/src/Redis/test/Eventuous.Tests.Redis/Fixtures/IntegrationFixture.cs
+++ b/src/Redis/test/Eventuous.Tests.Redis/Fixtures/IntegrationFixture.cs
@@ -8,19 +8,18 @@
 
 namespace Eventuous.Tests.Redis.Fixtures;
 
-public sealed class IntegrationFixture : IDisposable {
-    public IEventStore      EventStore     { get; }
+public sealed class IntegrationFixture : IAsyncDisposable {
+    public IEventStore      EventStore { get; }
     public IAggregateStore  AggregateStore { get; }
-    public GetRedisDatabase GetDatabase    { get; }
+    public GetRedisDatabase GetDatabase { get; }
 
     readonly ActivityListener _listener = DummyActivityListener.Create();
-    readonly GetRedisDatabase  _getDb;
 
     IEventSerializer Serializer { get; } = new DefaultEventSerializer(
         new JsonSerializerOptions(JsonSerializerDefaults.Web)
             .ConfigureForNodaTime(DateTimeZoneProviders.Tzdb)
     );
-
+    public static IntegrationFixture Instance { get; } = new();
     public IntegrationFixture() {
         const string connString = "localhost";
 
@@ -28,18 +27,18 @@ IDatabase GetDb() {
             var muxer = ConnectionMultiplexer.Connect(connString);
             return muxer.GetDatabase();
         }
-
-        _getDb = GetDb;
+               
+        var module = new Module();
+        module.LoadModule(GetDb).ConfigureAwait(false).GetAwaiter().GetResult();
 
         GetDatabase = GetDb;
         DefaultEventSerializer.SetDefaultSerializer(Serializer);
-        EventStore     = new RedisStore(GetDb, Serializer);
+        EventStore = new RedisStore(GetDb, new RedisStoreOptions(), Serializer);
         AggregateStore = new AggregateStore(EventStore);
     }
 
-    public Task Initialize()
-        => RedisEventStoreModule.LoadModule(_getDb);
-
-    public void Dispose()
-        => _listener.Dispose();
+    public ValueTask DisposeAsync() {
+        _listener.Dispose();
+        return default;
+    }
 }
diff --git a/src/Redis/test/Eventuous.Tests.Redis/Fixtures/SubscriptionFixture.cs b/src/Redis/test/Eventuous.Tests.Redis/Fixtures/SubscriptionFixture.cs
new file mode 100644
index 000000000..07964991a
--- /dev/null
+++ b/src/Redis/test/Eventuous.Tests.Redis/Fixtures/SubscriptionFixture.cs
@@ -0,0 +1,96 @@
+using Eventuous.Diagnostics.Logging;
+using Eventuous.Redis;
+using Eventuous.Redis.Subscriptions;
+using Eventuous.Subscriptions;
+using Eventuous.Subscriptions.Filters;
+using Eventuous.Sut.Subs;
+
+namespace Eventuous.Tests.Redis.Fixtures;
+
+public abstract class SubscriptionFixture<T> : IAsyncLifetime
+    where T : class, IEventHandler {
+    static SubscriptionFixture()
+        => TypeMap.Instance.RegisterKnownEventTypes(typeof(TestEvent).Assembly);
+
+    protected IntegrationFixture      IntegrationFixture { get; } = new();
+    protected StreamName              Stream             { get; }
+    protected T                       Handler            { get; }
+    protected ILogger                 Log                { get; }
+    protected RedisCheckpointStore CheckpointStore    { get; }
+    IMessageSubscription              Subscription       { get; }
+    protected ILoggerFactory          LoggerFactory      { get; }
+
+    protected SubscriptionFixture(
+        ITestOutputHelper    outputHelper,
+        bool                 subscribeToAll,
+        bool                 autoStart     = true,
+        Action<ConsumePipe>? configurePipe = null,
+        LogLevel             logLevel      = LogLevel.Trace
+    ) {
+        _autoStart = autoStart;
+
+        Stream = new StreamName(SharedAutoFixture.Auto.Create<string>());
+
+        LoggerFactory  = TestHelpers.Logging.GetLoggerFactory(outputHelper, logLevel);
+        SubscriptionId = $"test-{Guid.NewGuid():N}";
+
+        Handler         = GetHandler();
+        Log             = LoggerFactory.CreateLogger(GetType());
+        CheckpointStore = new RedisCheckpointStore(IntegrationFixture.GetDatabase, LoggerFactory);
+
+        _listener = new LoggingEventListener(LoggerFactory);
+        var pipe = new ConsumePipe();
+        configurePipe?.Invoke(pipe);
+        pipe.AddDefaultConsumer(Handler);
+
+        Subscription =
+            !subscribeToAll
+                ? new RedisStreamSubscription(
+                    IntegrationFixture.GetDatabase,
+                    new RedisStreamSubscriptionOptions(Stream) {
+                        SubscriptionId = SubscriptionId
+                    },
+                    CheckpointStore,
+                    pipe,
+                    LoggerFactory
+                )
+                : new RedisAllStreamSubscription(
+                    IntegrationFixture.GetDatabase,
+                    new RedisAllStreamSubscriptionOptions {
+                        SubscriptionId = SubscriptionId
+                    },
+                    CheckpointStore,
+                    pipe,
+                    LoggerFactory
+                );
+    }
+
+    protected abstract T GetHandler();
+
+    public string SubscriptionId { get; }
+
+    protected ValueTask Start()
+        => Subscription.SubscribeWithLog(Log);
+
+    protected ValueTask Stop()
+        => Subscription.UnsubscribeWithLog(Log);
+
+    readonly bool                 _autoStart;
+    readonly LoggingEventListener _listener;
+
+    public async Task InitializeAsync() {
+        if (_autoStart) await Start();
+    }
+
+    public async Task DisposeAsync() {
+        if (_autoStart) await Stop();
+        await FlashDB();
+        _listener.Dispose();
+        await IntegrationFixture.DisposeAsync();
+    }
+
+    async Task FlashDB() {
+        var database = IntegrationFixture.GetDatabase();
+        await database.ExecuteAsync("FLUSHDB");
+    }
+}
diff --git a/src/Redis/test/Eventuous.Tests.Redis/Store/Append.cs b/src/Redis/test/Eventuous.Tests.Redis/Store/Append.cs
index 28604efad..d5e3b02ad 100644
--- a/src/Redis/test/Eventuous.Tests.Redis/Store/Append.cs
+++ b/src/Redis/test/Eventuous.Tests.Redis/Store/Append.cs
@@ -3,15 +3,15 @@
 
 namespace Eventuous.Tests.Redis.Store;
 
-public class AppendEvents : IAsyncLifetime {
+[Collection("Sequential")]
+public class AppendEvents {
     readonly IntegrationFixture _fixture = new();
 
     [Fact]
     public async Task ShouldAppendToNoStream() {
         var evt        = CreateEvent();
         var streamName = GetStreamName();
-        var result     = await _fixture.EventStore.AppendEvent(streamName, evt, ExpectedStreamVersion.NoStream);
-
+        var result     = await AppendEvent(streamName, evt, ExpectedStreamVersion.NoStream);
         result.NextExpectedVersion.Should().Be(0);
     }
 
@@ -20,13 +20,11 @@ public async Task ShouldAppendOneByOne() {
         var evt    = CreateEvent();
         var stream = GetStreamName();
 
-        var result = await _fixture.EventStore.AppendEvent(stream, evt, ExpectedStreamVersion.NoStream);
-
+        var result = await AppendEvent(stream, evt, ExpectedStreamVersion.NoStream);
         evt = CreateEvent();
 
         var version = new ExpectedStreamVersion(result.NextExpectedVersion);
-        result = await _fixture.EventStore.AppendEvent(stream, evt, version);
-
+        result = await AppendEvent(stream, evt, version);
         result.NextExpectedVersion.Should().Be(1);
     }
 
@@ -35,11 +33,11 @@ public async Task ShouldFailOnWrongVersionNoStream() {
         var evt    = CreateEvent();
         var stream = GetStreamName();
 
-        await _fixture.EventStore.AppendEvent(stream, evt, ExpectedStreamVersion.NoStream);
+        await AppendEvent(stream, evt, ExpectedStreamVersion.NoStream);
 
         evt = CreateEvent();
 
-        var task = () => _fixture.EventStore.AppendEvent(stream, evt, ExpectedStreamVersion.NoStream);
+        var task = () => AppendEvent(stream, evt, ExpectedStreamVersion.NoStream);
         await task.Should().ThrowAsync<AppendToStreamException>();
     }
 
@@ -48,19 +46,11 @@ public async Task ShouldFailOnWrongVersion() {
         var evt    = CreateEvent();
         var stream = GetStreamName();
 
-        await _fixture.EventStore.AppendEvent(stream, evt, ExpectedStreamVersion.NoStream);
+        await AppendEvent(stream, evt, ExpectedStreamVersion.NoStream);
 
         evt = CreateEvent();
 
-        var task = () => _fixture.EventStore.AppendEvent(stream, evt, new ExpectedStreamVersion(3));
+        var task = () => AppendEvent(stream, evt, new ExpectedStreamVersion(3));
         await task.Should().ThrowAsync<AppendToStreamException>();
     }
-
-    public Task InitializeAsync()
-        => _fixture.Initialize();
-
-    public Task DisposeAsync() {
-        _fixture.Dispose();
-        return Task.CompletedTask;
-    }
 }
diff --git a/src/Redis/test/Eventuous.Tests.Redis/Store/Helpers.cs b/src/Redis/test/Eventuous.Tests.Redis/Store/Helpers.cs
index 6bfecbe6a..eed30f228 100644
--- a/src/Redis/test/Eventuous.Tests.Redis/Store/Helpers.cs
+++ b/src/Redis/test/Eventuous.Tests.Redis/Store/Helpers.cs
@@ -1,6 +1,7 @@
 using Eventuous.Tests.Redis.Fixtures;
 using static Eventuous.Sut.App.Commands;
 using static Eventuous.Sut.Domain.BookingEvents;
+using static Eventuous.Tests.Redis.Fixtures.IntegrationFixture;
 
 namespace Eventuous.Tests.Redis.Store;
 
@@ -21,22 +22,16 @@ static BookingImported ToEvent(ImportBooking cmd)
         => new(cmd.RoomId, cmd.Price, cmd.CheckIn, cmd.CheckOut);
 
     public static Task<AppendEventsResult> AppendEvents(
-        this IEventStore      store,
         StreamName            stream,
         object[]              evt,
         ExpectedStreamVersion version
     ) {
         var streamEvents = evt.Select(x => new StreamEvent(Guid.NewGuid(), x, new Metadata(), "", 0));
-        return store.AppendEvents(stream, version, streamEvents.ToArray(), default);
+        return Instance.EventStore.AppendEvents(stream, version, streamEvents.ToArray(), default);
     }
 
-    public static Task<AppendEventsResult> AppendEvent(
-        this IEventStore      store,
-        StreamName            stream,
-        object                evt,
-        ExpectedStreamVersion version
-    ) {
+    public static Task<AppendEventsResult> AppendEvent(StreamName stream, object evt, ExpectedStreamVersion version) {
         var streamEvent = new StreamEvent(Guid.NewGuid(), evt, new Metadata(), "", 0);
-        return store.AppendEvents(stream, version, new[] { streamEvent }, default);
+        return Instance.EventStore.AppendEvents(stream, version, new[] { streamEvent }, default);
     }
 }
diff --git a/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs b/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs
new file mode 100644
index 000000000..a8998dced
--- /dev/null
+++ b/src/Redis/test/Eventuous.Tests.Redis/Store/Read.cs
@@ -0,0 +1,87 @@
+using static Eventuous.Tests.Redis.Fixtures.IntegrationFixture;
+using static Eventuous.Tests.Redis.Store.Helpers;
+
+namespace Eventuous.Tests.Redis.Store;
+
+[Collection("Sequential")]
+public class ReadEvents {
+
+    [Fact]
+    public async Task ShouldReadOne() {
+        var evt        = CreateEvent();
+        var streamName = GetStreamName();
+        await AppendEvent(streamName, evt, ExpectedStreamVersion.NoStream);
+
+        var result = await Instance.EventStore.ReadEvents(
+            streamName,
+            StreamReadPosition.Start,
+            100,
+            default
+        );
+
+        result.Length.Should().Be(1);
+        result[0].Payload.Should().BeEquivalentTo(evt);
+    }
+
+    [Fact]
+    public async Task ShouldReadMany() {
+        // ReSharper disable once CoVariantArrayConversion
+        object[] events = CreateEvents(20).ToArray();
+        var streamName = GetStreamName();
+        await AppendEvents(streamName, events, ExpectedStreamVersion.NoStream);
+        
+        var result = await Instance.EventStore.ReadEvents(
+            streamName,
+            StreamReadPosition.Start,
+            100,
+            default
+        );
+
+        var actual = result.Select(x => x.Payload);
+        actual.Should().BeEquivalentTo(events);
+    }
+    
+    [Fact]
+    public async Task ShouldReadTail() {
+        // ReSharper disable once CoVariantArrayConversion
+        var streamName = GetStreamName();
+
+        object[] events1 = CreateEvents(10).ToArray();
+        var appended = await AppendEvents(streamName, events1, ExpectedStreamVersion.NoStream);
+        var position = appended.GlobalPosition;
+
+        object[] events2 = CreateEvents(10).ToArray();
+        await AppendEvents(streamName, events2, ExpectedStreamVersion.Any);        
+        
+        var result = await Instance.EventStore.ReadEvents(
+            streamName,
+            new StreamReadPosition((long)position),
+            100,
+            default
+        );
+
+        var expected = events2;
+        var actual   = result.Select(x => x.Payload);
+        actual.Should().BeEquivalentTo(expected);
+    }
+    
+    [Fact]
+    public async Task ShouldReadHead() {
+        // ReSharper disable once CoVariantArrayConversion
+        object[] events = CreateEvents(20).ToArray();
+        var streamName = GetStreamName();
+        await AppendEvents(streamName, events, ExpectedStreamVersion.NoStream);
+        
+        var result = await Instance.EventStore.ReadEvents(
+            streamName,
+            StreamReadPosition.Start, 
+            10,
+            default
+        );
+
+        var expected = events.Take(10);
+        var actual   = result.Select(x => x.Payload);
+        actual.Should().BeEquivalentTo(expected);
+    }
+
+}
\ No newline at end of file
diff --git a/src/Redis/test/Eventuous.Tests.Redis/Subscriptions/SubscribeToAll.cs b/src/Redis/test/Eventuous.Tests.Redis/Subscriptions/SubscribeToAll.cs
new file mode 100644
index 000000000..19c388d94
--- /dev/null
+++ b/src/Redis/test/Eventuous.Tests.Redis/Subscriptions/SubscribeToAll.cs
@@ -0,0 +1,73 @@
+using Eventuous.Subscriptions.Checkpoints;
+using Eventuous.Subscriptions.Logging;
+using Eventuous.Sut.Subs;
+using Eventuous.Tests.Redis.Fixtures;
+using Hypothesist;
+using static Eventuous.Sut.App.Commands;
+using static Eventuous.Sut.Domain.BookingEvents;
+
+namespace Eventuous.Tests.Redis.Subscriptions;
+
+[Collection("Sequential")]
+public class SubscribeToAll : SubscriptionFixture<TestEventHandler> {
+    public SubscribeToAll(ITestOutputHelper outputHelper)
+        : base(outputHelper, true, false) {
+    }
+
+    [Fact]
+    public async Task ShouldConsumeProducedEvents() {
+        const int count = 10;
+
+        var (testEvents, _) = await GenerateAndProduceEvents(count);
+        Handler.AssertThat().Exactly(count, x => testEvents.Contains(x));
+
+        await Start();
+        await Handler.Validate(2.Seconds());
+        await Stop();
+        
+        Handler.Count.Should().Be(10);
+    }
+
+    [Fact]
+    public async Task ShouldUseExistingCheckpoint() {
+        const int count = 10;
+
+        var (_, result) = await GenerateAndProduceEvents(count);
+        Handler.AssertThat().Any(_ => true);
+
+        var checkpoint = await CheckpointStore.GetLastCheckpoint(SubscriptionId, default);
+        Logger.ConfigureIfNull(SubscriptionId, LoggerFactory);
+        await CheckpointStore.StoreCheckpoint(new Checkpoint(SubscriptionId, (ulong)result.GlobalPosition), true, default);
+
+        await Start();
+        await Task.Delay(TimeSpan.FromSeconds(1));
+        await Stop();
+        Handler.Count.Should().Be(0);
+    }
+
+    static BookingImported ToEvent(ImportBooking cmd)
+        => new(cmd.RoomId, cmd.Price, cmd.CheckIn, cmd.CheckOut);
+
+    async Task<(List<BookingImported>, AppendEventsResult)> GenerateAndProduceEvents(int count) {
+        var commands = Enumerable
+            .Range(0, count)
+            .Select(_ => DomainFixture.CreateImportBooking())
+            .ToList();
+
+        var events = commands.Select(ToEvent).ToList();
+
+        var streamEvents = events.Select(x => new StreamEvent(Guid.NewGuid(), x, new Metadata(), "", 0));
+
+        var result = await IntegrationFixture.EventStore.AppendEvents(
+            Stream,
+            ExpectedStreamVersion.Any,
+            streamEvents.ToArray(),
+            default
+        );
+
+        return (events, result);
+    }
+
+    protected override TestEventHandler GetHandler()
+        => new();
+}
diff --git a/src/Redis/test/Eventuous.Tests.Redis/Subscriptions/SubscribeToStream.cs b/src/Redis/test/Eventuous.Tests.Redis/Subscriptions/SubscribeToStream.cs
new file mode 100644
index 000000000..31bfbcb20
--- /dev/null
+++ b/src/Redis/test/Eventuous.Tests.Redis/Subscriptions/SubscribeToStream.cs
@@ -0,0 +1,85 @@
+using System.Linq;
+using Eventuous.Subscriptions.Checkpoints;
+using Eventuous.Subscriptions.Logging;
+using Eventuous.Sut.Subs;
+using Eventuous.Tests.Redis.Fixtures;
+using Hypothesist;
+using static Eventuous.Sut.App.Commands;
+using static Eventuous.Sut.Domain.BookingEvents;
+
+namespace Eventuous.Tests.Redis.Subscriptions;
+
+[Collection("Sequential")]
+public class SubscribeToStream : SubscriptionFixture<TestEventHandler> {
+    public SubscribeToStream(ITestOutputHelper outputHelper)
+        : base(outputHelper, false, false) {
+    }
+
+    [Fact]
+    public async Task ShouldConsumeProducedEvents() {
+        const int count = 10;
+
+        var testEvents = await GenerateAndProduceEvents(count);
+        Handler.AssertThat().Exactly(count, x => testEvents.Contains(x));
+
+        await Start();
+        await Handler.Validate(2.Seconds());
+        await Stop();
+        
+        Handler.Count.Should().Be(10);
+    }
+
+    [Fact]
+    public async Task ShouldUseExistingCheckpoint() {
+        const int count = 10;
+
+        await GenerateAndProduceEvents(count);
+        Handler.AssertThat().Any(_ => true);
+
+        var checkpoint = await CheckpointStore.GetLastCheckpoint(SubscriptionId, default);
+        var streamPosition = await GetStreamPosition(count);
+        Logger.ConfigureIfNull(SubscriptionId, LoggerFactory);
+        await CheckpointStore.StoreCheckpoint(new Checkpoint(SubscriptionId, (ulong)streamPosition), true, default);
+
+        await Start();
+        await Task.Delay(TimeSpan.FromSeconds(1));
+        await Stop();
+        Handler.Count.Should().Be(0);
+    }
+
+    static BookingImported ToEvent(ImportBooking cmd)
+        => new(cmd.RoomId, cmd.Price, cmd.CheckIn, cmd.CheckOut);
+
+    async Task<List<BookingImported>> GenerateAndProduceEvents(int count) {
+        var commands = Enumerable
+            .Range(0, count)
+            .Select(_ => DomainFixture.CreateImportBooking())
+            .ToList();
+
+        var events = commands.Select(ToEvent).ToList();
+
+        var streamEvents = events.Select(x => new StreamEvent(Guid.NewGuid(), x, new Metadata(), "", 0));
+
+        await IntegrationFixture.EventStore.AppendEvents(
+            Stream,
+            ExpectedStreamVersion.Any,
+            streamEvents.ToArray(),
+            default
+        );
+
+        return events;
+    }
+
+    async Task<long> GetStreamPosition(int count) {
+        var readEvents = await IntegrationFixture.EventStore.ReadEvents(
+            Stream,
+            StreamReadPosition.Start,
+            count,
+            default
+        );
+        return readEvents.Last().Position;
+    }
+
+    protected override TestEventHandler GetHandler()
+        => new();
+}
