- more "supports simulate failure" checks

- replace StreamConnection

Author: mgravell

tests/StackExchange.Redis.Tests/BasicOpTests.cs

@@ -270,13 +270,15 @@ public async Task GetWithExpiryWrongTypeSync()
     [Fact]
     public async Task TestSevered()
     {
-        SetExpectedAmbientFailureCount(2);
         await using var conn = Create(allowAdmin: true, shared: false);
         var db = conn.GetDatabase();
         string key = Me();
         db.KeyDelete(key, CommandFlags.FireAndForget);
         db.StringSet(key, key, flags: CommandFlags.FireAndForget);
         var server = GetServer(conn);
+        Assert.SkipUnless(server.CanSimulateConnectionFailure(), "Skipping because server cannot simulate connection failure");
+
+        SetExpectedAmbientFailureCount(2);
         server.SimulateConnectionFailure(SimulatedFailureType.All);
         var watch = Stopwatch.StartNew();
         await UntilConditionAsync(TimeSpan.FromSeconds(10), () => server.IsConnected);

tests/StackExchange.Redis.Tests/ConnectionFailedErrorsTests.cs

@@ -180,6 +180,7 @@ void InnerScenario()
             {
                 conn.GetDatabase();
                 var server = conn.GetServer(conn.GetEndPoints()[0]);
+                Assert.SkipUnless(server.CanSimulateConnectionFailure(), "Skipping because server cannot simulate connection failure");
 
                 conn.AllowConnect = false;
 

tests/StackExchange.Redis.Tests/DuplexStream.cs

@@ -0,0 +1,131 @@
+using System;
+using System.IO;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace StackExchange.Redis.Tests;
+
+/// <summary>
+/// Combines separate input and output streams into a single duplex stream.
+/// </summary>
+internal sealed class DuplexStream(Stream inputStream, Stream outputStream) : Stream
+{
+    private readonly Stream _inputStream = inputStream ?? throw new ArgumentNullException(nameof(inputStream));
+    private readonly Stream _outputStream = outputStream ?? throw new ArgumentNullException(nameof(outputStream));
+
+    public override bool CanRead => _inputStream.CanRead;
+    public override bool CanWrite => _outputStream.CanWrite;
+    public override bool CanSeek => false;
+    public override bool CanTimeout => _inputStream.CanTimeout || _outputStream.CanTimeout;
+
+    public override int ReadTimeout
+    {
+        get => _inputStream.ReadTimeout;
+        set => _inputStream.ReadTimeout = value;
+    }
+
+    public override int WriteTimeout
+    {
+        get => _outputStream.WriteTimeout;
+        set => _outputStream.WriteTimeout = value;
+    }
+
+    public override long Length => throw new NotSupportedException($"{nameof(DuplexStream)} does not support seeking.");
+    public override long Position
+    {
+        get => throw new NotSupportedException($"{nameof(DuplexStream)} does not support seeking.");
+        set => throw new NotSupportedException($"{nameof(DuplexStream)} does not support seeking.");
+    }
+
+    public override int Read(byte[] buffer, int offset, int count)
+        => _inputStream.Read(buffer, offset, count);
+
+    public override Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+        => _inputStream.ReadAsync(buffer, offset, count, cancellationToken);
+
+#if NET6_0_OR_GREATER
+    public override ValueTask<int> ReadAsync(Memory<byte> buffer, CancellationToken cancellationToken = default)
+        => _inputStream.ReadAsync(buffer, cancellationToken);
+
+    public override int Read(Span<byte> buffer)
+        => _inputStream.Read(buffer);
+#endif
+
+    public override int ReadByte()
+        => _inputStream.ReadByte();
+
+    public override void Write(byte[] buffer, int offset, int count)
+        => _outputStream.Write(buffer, offset, count);
+
+    public override Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
+        => _outputStream.WriteAsync(buffer, offset, count, cancellationToken);
+
+#if NET6_0_OR_GREATER
+    public override ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default)
+        => _outputStream.WriteAsync(buffer, cancellationToken);
+
+    public override void Write(ReadOnlySpan<byte> buffer)
+        => _outputStream.Write(buffer);
+#endif
+
+    public override void WriteByte(byte value)
+        => _outputStream.WriteByte(value);
+
+    public override void Flush()
+        => _outputStream.Flush();
+
+    public override Task FlushAsync(CancellationToken cancellationToken)
+        => _outputStream.FlushAsync(cancellationToken);
+
+    public override long Seek(long offset, SeekOrigin origin)
+        => throw new NotSupportedException($"{nameof(DuplexStream)} does not support seeking.");
+
+    public override void SetLength(long value)
+        => throw new NotSupportedException($"{nameof(DuplexStream)} does not support seeking.");
+
+    public override void Close()
+    {
+        _inputStream.Close();
+        _outputStream.Close();
+        base.Close();
+    }
+
+    protected override void Dispose(bool disposing)
+    {
+        if (disposing)
+        {
+            _inputStream.Dispose();
+            _outputStream.Dispose();
+        }
+        base.Dispose(disposing);
+    }
+
+#if NET6_0_OR_GREATER
+    public override async ValueTask DisposeAsync()
+    {
+        await _inputStream.DisposeAsync().ConfigureAwait(false);
+        await _outputStream.DisposeAsync().ConfigureAwait(false);
+        await base.DisposeAsync().ConfigureAwait(false);
+    }
+#endif
+
+    public override IAsyncResult BeginRead(byte[] buffer, int offset, int count, AsyncCallback? callback, object? state)
+        => _inputStream.BeginRead(buffer, offset, count, callback, state);
+
+    public override int EndRead(IAsyncResult asyncResult)
+        => _inputStream.EndRead(asyncResult);
+
+    public override IAsyncResult BeginWrite(byte[] buffer, int offset, int count, AsyncCallback? callback, object? state)
+        => _outputStream.BeginWrite(buffer, offset, count, callback, state);
+
+    public override void EndWrite(IAsyncResult asyncResult)
+        => _outputStream.EndWrite(asyncResult);
+
+#if NET6_0_OR_GREATER
+    public override void CopyTo(Stream destination, int bufferSize)
+        => _inputStream.CopyTo(destination, bufferSize);
+#endif
+
+    public override Task CopyToAsync(Stream destination, int bufferSize, CancellationToken cancellationToken)
+        => _inputStream.CopyToAsync(destination, bufferSize, cancellationToken);
+}

tests/StackExchange.Redis.Tests/ExceptionFactoryTests.cs

@@ -37,7 +37,9 @@ public async Task MultipleEndpointsThrowConnectionException()
 
             foreach (var endpoint in conn.GetEndPoints())
             {
-                conn.GetServer(endpoint).SimulateConnectionFailure(SimulatedFailureType.All);
+                var server = conn.GetServer(endpoint);
+                Assert.SkipUnless(server.CanSimulateConnectionFailure(), "Skipping because server cannot simulate connection failure");
+                server.SimulateConnectionFailure(SimulatedFailureType.All);
             }
 
             var ex = ExceptionFactory.NoConnectionAvailable(conn.UnderlyingMultiplexer, null, null);

tests/StackExchange.Redis.Tests/FailoverTests.cs

@@ -222,6 +222,7 @@ public async Task SubscriptionsSurviveConnectionFailureAsync()
         Assert.Equal(1, counter1);
 
         var server = GetServer(conn);
+        Assert.SkipUnless(server.CanSimulateConnectionFailure(), "Skipping because server cannot simulate connection failure");
         var socketCount = server.GetCounters().Subscription.SocketCount;
         Log($"Expecting 1 socket, got {socketCount}");
         Assert.Equal(1, socketCount);

tests/StackExchange.Redis.Tests/InProcessTestServer.cs

@@ -112,7 +112,10 @@ private sealed class InProcTunnel(
                 var serverToClient = new Pipe(pipeOptions ?? PipeOptions.Default);
                 var serverSide = new Duplex(clientToServer.Reader, serverToClient.Writer);
                 Task.Run(async () => await server.RunClientAsync(serverSide, node: node), cancellationToken).RedisFireAndForget();
-                var clientSide = StreamConnection.GetDuplex(serverToClient.Reader, clientToServer.Writer);
+
+                var readStream = serverToClient.Reader.AsStream();
+                var writeStream = clientToServer.Writer.AsStream();
+                var clientSide = new DuplexStream(readStream, writeStream);
                 return new(clientSide);
             }
             return base.BeforeAuthenticateAsync(endpoint, connectionType, socket, cancellationToken);