RemoteProcess: Write EOF when StandardInputStream gets Closed/Dispose…

…d. (#262)

README.md

@@ -99,7 +99,8 @@ class RemoteProcess : IDisposable
   ValueTask WriteAsync(string value, CancellationToken cancellationToken = default);
   ValueTask WriteLineAsync(ReadOnlyMemory<char> buffer, CancellationToken cancellationToken = default);
   ValueTask WriteLineAsync(string? value, CancellationToken cancellationToken = default);
-  Stream StandardInputStream { get; }
+  void WriteEof();
+  Stream StandardInputStream { get; } // Disposing/Closing the Stream calls WriteEof.
   StreamWriter StandardInputWriter { get; }
 
   // Wait for the remote process to exit.

src/Tmds.Ssh/ISshChannel.cs

@@ -17,7 +17,7 @@ interface ISshChannel
             CancellationToken cancellationToken = default);
 
     ValueTask WriteAsync(ReadOnlyMemory<byte> data, CancellationToken cancellationToken = default);
-    void WriteEof();
+    void WriteEof(bool noThrow = false);
 
     Exception CreateCloseException();
 }

src/Tmds.Ssh/RemoteProcess.cs

@@ -248,6 +248,16 @@ private enum ReadMode
 
     private bool HasExited { get => _readMode == ReadMode.Exited; } // delays exit until it was read by the user.
 
+    private void WriteEof(bool noThrow)
+    {
+        _channel.WriteEof(noThrow);
+    }
+
+    public void WriteEof()
+    {
+        WriteEof(noThrow: false);
+    }
+
     public ValueTask WriteAsync(ReadOnlyMemory<byte> buffer, CancellationToken cancellationToken = default)
     {
         ThrowIfDisposed();
@@ -640,6 +650,14 @@ public override Task FlushAsync(CancellationToken cancellationToken = default)
             {
                 throw new IOException($"Unable to transport data: {ex.Message}.", ex);
             }
+
+        }
+
+        public override void Close()
+        {
+            // The base Stream class calls Close for implementing Dispose.
+            // We mustn't throw to avoid throwing on Dispose.
+            _process.WriteEof(noThrow: true);
         }
     }
 }

src/Tmds.Ssh/SshChannel.cs

@@ -174,14 +174,20 @@ string    data
         }
     }
 
-    public void WriteEof()
+    public void WriteEof(bool noThrow)
     {
-        ThrowIfDisposed();
-        ThrowIfAborted();
-        ThrowIfEofSent();
-
-        _eofSent = true;
-        TrySendEofMessage();
+        if (!noThrow)
+        {
+            ThrowIfDisposed();
+            ThrowIfAborted();
+            ThrowIfEofSent();
+        }
+
+        if (!_eofSent)
+        {
+            _eofSent = true;
+            TrySendEofMessage();
+        }
     }
 
     private void ThrowIfEofSent()

test/Tmds.Ssh.Tests/RemoteProcessTests.cs

@@ -119,7 +119,7 @@ public async Task WriteAndRead(WriteApi writeApi)
 
         byte[] buffer = new byte[512];
         (bool isError, int bytesRead) = await process.ReadAsync(buffer, buffer);
-        Assert.False(false);
+        Assert.False(isError);
         Assert.Equal(helloWorldBytes, buffer.AsSpan(0, bytesRead).ToArray());
     }
 
@@ -421,4 +421,43 @@ public static IEnumerable<object[]> NewlineTestData
             }
         }
     }
+
+    public enum EofApi
+    {
+        WriteEof,
+        StandardInputStreamClose,
+        StandardInputStreamDispose,
+    }
+
+    [Theory]
+    [InlineData(EofApi.WriteEof)]
+    [InlineData(EofApi.StandardInputStreamClose)]
+    [InlineData(EofApi.StandardInputStreamDispose)]
+    public async Task WriteEof(EofApi eofApi)
+    {
+        using var client = await _sshServer.CreateClientAsync();
+        using var process = await client.ExecuteAsync("cat");
+
+        switch (eofApi)
+        {
+            case EofApi.WriteEof:
+                process.WriteEof();
+                break;
+            case EofApi.StandardInputStreamClose:
+                process.StandardInputStream.Close();
+                break;
+            case EofApi.StandardInputStreamDispose:
+                process.StandardInputStream.Dispose();
+                break;
+        }
+
+        // Verify that Disposing the Stream after sending EOF does NOT throw.
+        Stream s = process.StandardInputStream;
+        s.Dispose();
+
+        byte[] buffer = new byte[512];
+        (bool isError, int bytesRead) = await process.ReadAsync(buffer, buffer);
+        Assert.False(isError);
+        Assert.Equal(0, bytesRead);
+    }
 }