Bug 1738 – Stream interface in .NET assembly (download)

https://winscp.net/tracker/1738
(cherry picked from commit 56db37fceb4f21371e4230c183daf05facb35ea7)

Source commit: 8bfb7c64b2c98cd69546a2ad027a6025bbefa7bc

dotnet/Session.cs

@@ -840,17 +840,7 @@ namespace WinSCP
         {
             using (Logger.CreateCallstack())
             {
-                if (options == null)
-                {
-                    options = new TransferOptions();
-                }
-
-                CheckOpened();
-
-                WriteCommand(
-                    string.Format(CultureInfo.InvariantCulture, "get {0} {1} {2} -- \"{3}\" \"{4}\"",
-                        BooleanSwitch(remove, "delete"), options.ToSwitches(), additionalParams,
-                        Tools.ArgumentEscape(remotePath), Tools.ArgumentEscape(localPath)));
+                StartGetCommand(remotePath, localPath, remove, options, additionalParams);
 
                 TransferOperationResult result = new TransferOperationResult();
 
@@ -887,6 +877,21 @@ namespace WinSCP
             }
         }
 
+        private void StartGetCommand(string remotePath, string localPath, bool remove, TransferOptions options, string additionalParams)
+        {
+            if (options == null)
+            {
+                options = new TransferOptions();
+            }
+
+            CheckOpened();
+
+            WriteCommand(
+                string.Format(CultureInfo.InvariantCulture, "get {0} {1} {2} -- \"{3}\" \"{4}\"",
+                    BooleanSwitch(remove, "delete"), options.ToSwitches(), additionalParams,
+                    Tools.ArgumentEscape(remotePath), Tools.ArgumentEscape(localPath)));
+        }
+
         public TransferOperationResult GetFilesToDirectory(
             string remoteDirectory, string localDirectory, string filemask = null, bool remove = false, TransferOptions options = null)
         {
@@ -941,13 +946,7 @@ namespace WinSCP
         {
             using (Logger.CreateCallstack())
             {
-                if (string.IsNullOrEmpty(remoteFilePath))
-                {
-                    throw Logger.WriteException(new ArgumentException("File to path cannot be empty", nameof(remoteFilePath)));
-                }
-
-                string remoteDirectory = RemotePath.GetDirectoryName(remoteFilePath);
-                string filemask = RemotePath.EscapeFileMask(RemotePath.GetFileName(remoteFilePath));
+                ParseRemotePath(remoteFilePath, out string remoteDirectory, out string filemask);
 
                 TransferOperationResult operationResult =
                     DoGetFilesToDirectory(remoteDirectory, localDirectory, filemask, remove, options, additionalParams ?? string.Empty);
@@ -956,6 +955,17 @@ namespace WinSCP
             }
         }
 
+        private void ParseRemotePath(string remoteFilePath, out string remoteDirectory, out string filemask)
+        {
+            if (string.IsNullOrEmpty(remoteFilePath))
+            {
+                throw Logger.WriteException(new ArgumentException("File to path cannot be empty", nameof(remoteFilePath)));
+            }
+
+            remoteDirectory = RemotePath.GetDirectoryName(remoteFilePath);
+            filemask = RemotePath.EscapeFileMask(RemotePath.GetFileName(remoteFilePath));
+        }
+
         private T GetOnlyFileOperation<T>(ICollection<T> operations)
         {
             // For "get", this should happen only when the filename is mask-like, otherwise "get" throws straight away.
@@ -971,6 +981,94 @@ namespace WinSCP
             return operations.First();
         }
 
+        public Stream GetFile(string remoteFilePath, TransferOptions options = null)
+        {
+            using (CallstackAndLock callstackAndLock = Logger.CreateCallstackAndLock())
+            {
+                const string additionalParams = "-onlyfile";
+                ParseRemotePath(remoteFilePath, out string remoteDirectory, out string filemask);
+                string remotePath = RemotePath.Combine(remoteDirectory, filemask);
+
+                StartGetCommand(remotePath, "-", false, options, additionalParams);
+
+                // only to collect failures
+                TransferOperationResult result = new TransferOperationResult();
+
+                ElementLogReader groupReader = _reader.WaitForGroupAndCreateLogReader();
+                IDisposable operationResultGuard = RegisterOperationResult(result);
+                IDisposable progressHandler = CreateProgressHandler();
+
+                void onGetEnd()
+                {
+                    try
+                    {
+                        // This can throw
+                        progressHandler.Dispose();
+                    }
+                    finally
+                    {
+                        try
+                        {
+                            groupReader.Dispose();
+                        }
+                        finally
+                        {
+                            // This should not throw, so another pair of try ... finally is not necessary.
+                            // Only after disposing the groupr reader, so when called from onGetEndWithExit, the Check() has all failures.
+                            operationResultGuard.Dispose();
+                        }
+                    }
+                }
+
+                void onGetEndWithExit()
+                {
+                    try
+                    {
+                        onGetEnd();
+                    }
+                    finally
+                    {
+                        Logger.Lock.Exit();
+                        result.Check();
+                    }
+                }
+
+                try
+                {
+                    bool downloadFound;
+                    try
+                    {
+                        // Not using WaitForNonEmptyElement only to allow throwing better exception message below.
+                        // Not using ThrowFailures as we need to return the stream, is there's <download>, even if there's a <failure> as well.
+                        downloadFound = groupReader.TryWaitForNonEmptyElement(TransferEventArgs.DownloadTag, 0);
+                    }
+                    catch (StdOutException)
+                    {
+                        downloadFound = true;
+                    }
+                    if (downloadFound)
+                    {
+                        ChunkedReadStream stream = new ChunkedReadStream(_process.StdOut, onGetEndWithExit);
+                        callstackAndLock.DisarmLock();
+                        return stream;
+                    }
+                    else
+                    {
+                        // First throw any specific error from <failure>
+                        result.Check();
+                        // And only then fallback to the generic one.
+                        // See also the comment in GetOnlyFileOperation.
+                        throw Logger.WriteException(new FileNotFoundException("File not found"));
+                    }
+                }
+                catch
+                {
+                    onGetEnd();
+                    throw;
+                }
+            }
+        }
+
         public RemovalOperationResult RemoveFiles(string path)
         {
             using (Logger.CreateCallstackAndLock())
@@ -2121,6 +2219,11 @@ namespace WinSCP
             {
                 throw Logger.WriteException(new SessionLocalException(this, "Aborted."));
             }
+
+            if (_process.StdOut.ReadAvailable(1))
+            {
+                throw new StdOutException();
+            }
         }
 
         private void RaiseFileTransferredEvent(TransferEventArgs args)
@@ -2212,6 +2315,7 @@ namespace WinSCP
 
         internal void UnregisterOperationResult(OperationResultBase operationResult)
         {
+            // GetFile relies on this not to throw
             _operationResults.Remove(operationResult);
         }
 

dotnet/internal/CallstackAndLock.cs

@@ -11,10 +11,18 @@
 
         public override void Dispose()
         {
-            _lock.Exit();
+            if (_lock != null)
+            {
+                _lock.Exit();
+            }
             base.Dispose();
         }
 
-        private readonly Lock _lock;
+        public void DisarmLock()
+        {
+            _lock = null;
+        }
+
+        private Lock _lock;
     }
 }

dotnet/internal/ChunkedReadStream.cs

@@ -6,9 +6,10 @@ namespace WinSCP
 {
     internal class ChunkedReadStream : Stream
     {
-        public ChunkedReadStream(Stream baseStream)
+        public ChunkedReadStream(Stream baseStream, Action onDispose)
         {
             _baseStream = baseStream;
+            _onDispose = onDispose;
             _remaining = 0;
             _eof = false;
         }
@@ -86,6 +87,13 @@ namespace WinSCP
                         throw new Exception("Expected LF");
                     }
                 }
+
+                if (_eof)
+                {
+                    // Throw any pending exception asap, not only once the stream is closed.
+                    // Also releases the lock.
+                    Closed();
+                }
             }
 
             return result;
@@ -106,7 +114,33 @@ namespace WinSCP
             throw new NotImplementedException();
         }
 
+        protected override void Dispose(bool disposing)
+        {
+            try
+            {
+                // Have to consume the rest of the buffered download data, otherwise we could not continue with other downloads
+                while (!_eof)
+                {
+                    byte[] buf = new byte[10240];
+                    Read(buf, 0, buf.Length);
+                }
+                base.Dispose(disposing);
+            }
+            finally
+            {
+                Closed();
+            }
+        }
+
+        private void Closed()
+        {
+            Action onDispose = _onDispose;
+            _onDispose = null;
+            onDispose?.Invoke();
+        }
+
         private Stream _baseStream;
+        private Action _onDispose;
         private int _remaining;
         private bool _eof;
     }

dotnet/internal/ConsoleCommStruct.cs

@@ -4,7 +4,7 @@ using Microsoft.Win32.SafeHandles;
 
 namespace WinSCP
 {
-    public enum ConsoleEvent { None, Print, Input, Choice, Title, Init, Progress }
+    public enum ConsoleEvent { None, Print, Input, Choice, Title, Init, Progress, TransferOut }
 
     [StructLayout(LayoutKind.Sequential)]
     internal class ConsoleInitEventStruct
@@ -86,6 +86,16 @@ namespace WinSCP
         public bool Cancel; // since version 8
     }
 
+    // Since version 10
+    [StructLayout(LayoutKind.Sequential, CharSet = CharSet.Unicode)]
+    internal class ConsoleTransferEventStruct
+    {
+        [MarshalAs(UnmanagedType.ByValArray, SizeConst = 20480)]
+        public byte[] Data;
+        public UIntPtr Len;
+        public bool Error; // TransferIn only
+    }
+
     [StructLayout(LayoutKind.Sequential)]
     internal class ConsoleCommHeader
     {
@@ -165,6 +175,8 @@ namespace WinSCP
 
         public ConsoleProgressEventStruct ProgressEvent { get { return UnmarshalPayload<ConsoleProgressEventStruct>(ConsoleEvent.Progress); } }
 
+        public ConsoleTransferEventStruct TransferOutEvent { get { return UnmarshalPayload<ConsoleTransferEventStruct>(ConsoleEvent.TransferOut); } }
+
         private T UnmarshalPayload<T>(ConsoleEvent e)
         {
             CheckNotDisposed();
@@ -197,7 +209,8 @@ namespace WinSCP
                 Type[] types =
                     new[] {
                         typeof(ConsolePrintEventStruct), typeof(ConsoleInitEventStruct), typeof(ConsoleInputEventStruct),
-                        typeof(ConsoleChoiceEventStruct), typeof(ConsoleTitleEventStruct), typeof(ConsoleProgressEventStruct) };
+                        typeof(ConsoleChoiceEventStruct), typeof(ConsoleTitleEventStruct), typeof(ConsoleProgressEventStruct),
+                        typeof(ConsoleTransferEventStruct) };
 
                 int maxSize = 0;
                 foreach (Type type in types)

dotnet/internal/ExeSessionProcess.cs

@@ -19,6 +19,7 @@ namespace WinSCP
 
         public bool HasExited { get { return _process.HasExited; } }
         public int ExitCode { get { return _process.ExitCode; } }
+        public PipeStream StdOut { get; private set; }
 
         public static ExeSessionProcess CreateForSession(Session session)
         {
@@ -98,7 +99,7 @@ namespace WinSCP
                     string.Format(CultureInfo.InvariantCulture, "/dotnet={0} ", assemblyVersionStr);
 
                 string arguments =
-                    xmlLogSwitch + "/nointeractiveinput " + assemblyVersionSwitch +
+                    xmlLogSwitch + "/nointeractiveinput /stdout=chunked " + assemblyVersionSwitch +
                     configSwitch + logSwitch + logLevelSwitch + _session.AdditionalExecutableArguments;
 
                 Tools.AddRawParameters(ref arguments, _session.RawConfiguration, "/rawconfig", false);
@@ -140,6 +141,7 @@ namespace WinSCP
         {
             using (_logger.CreateCallstack())
             {
+                StdOut = new PipeStream();
                 InitializeConsole();
                 InitializeChild();
             }
@@ -321,6 +323,10 @@ namespace WinSCP
                             ProcessProgressEvent(commStruct.ProgressEvent);
                             break;
 
+                        case ConsoleEvent.TransferOut:
+                            ProcessTransferOutEvent(commStruct.TransferOutEvent);
+                            break;
+
                         default:
                             throw _logger.WriteException(new NotImplementedException());
                     }
@@ -520,6 +526,18 @@ namespace WinSCP
             }
         }
 
+        private void ProcessTransferOutEvent(ConsoleTransferEventStruct e)
+        {
+            using (_logger.CreateCallstack())
+            {
+                _logger.WriteLine("Len [{0}]", e.Len);
+
+                StdOut.Write(e.Data, 0, (int)e.Len);
+
+                _logger.WriteLine("Data written to the buffer");
+            }
+        }
+
         private void InitializeConsole()
         {
             using (_logger.CreateCallstack())

dotnet/internal/Logger.cs

@@ -15,6 +15,7 @@ namespace WinSCP
         public string LogPath { get { return _logPath; } set { SetLogPath(value); } }
         public int LogLevel { get { return _logLevel; } set { SetLogLevel(value); } }
         public bool Logging { get { return (_writter != null) && _writter.BaseStream.CanWrite; } }
+        public Lock Lock { get; } = new Lock();
 
         public string GetAssemblyFilePath()
         {
@@ -282,9 +283,9 @@ namespace WinSCP
             return new Callstack(this, token);
         }
 
-        public Callstack CreateCallstackAndLock()
+        public CallstackAndLock CreateCallstackAndLock()
         {
-            return new CallstackAndLock(this, _lock);
+            return new CallstackAndLock(this, Lock);
         }
 
         public Exception WriteException(Exception e)
@@ -404,7 +405,6 @@ namespace WinSCP
         private string _logPath;
         private readonly Dictionary<int, int> _indents = new Dictionary<int, int>();
         private readonly object _logLock = new object();
-        private readonly Lock _lock = new Lock();
 #if !NETSTANDARD
         private List<PerformanceCounter> _performanceCounters = new List<PerformanceCounter>();
 #endif

dotnet/internal/PipeStream.cs

@@ -0,0 +1,360 @@
+using System;
+using System.Collections.Generic;
+using System.Globalization;
+using System.IO;
+using System.Threading;
+
+namespace WinSCP
+{
+    /// <summary>
+    /// PipeStream is a thread-safe read/write data stream for use between two threads in a
+    /// single-producer/single-consumer type problem.
+    /// </summary>
+    /// <version>2006/10/13 1.0</version>
+    /// <remarks>Update on 2008/10/9 1.1 - uses Monitor instead of Manual Reset events for more elegant synchronicity.</remarks>
+    /// <license>
+    ///  Copyright (c) 2006 James Kolpack (james dot kolpack at google mail)
+    ///
+    ///  Permission is hereby granted, free of charge, to any person obtaining a copy of this software and
+    ///  associated documentation files (the "Software"), to deal in the Software without restriction,
+    ///  including without limitation the rights to use, copy, modify, merge, publish, distribute,
+    ///  sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is
+    ///  furnished to do so, subject to the following conditions:
+    ///
+    ///  The above copyright notice and this permission notice shall be included in all copies or
+    ///  substantial portions of the Software.
+    ///
+    ///  THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED,
+    ///  INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR
+    ///  PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
+    ///  LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT
+    ///  OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
+    ///  OTHER DEALINGS IN THE SOFTWARE.
+    /// </license>
+    public class PipeStream : Stream
+    {
+        #region Private members
+
+        /// <summary>
+        /// Queue of bytes provides the datastructure for transmitting from an
+        /// input stream to an output stream.
+        /// </summary>
+        /// <remarks>Possible more effecient ways to accomplish this.</remarks>
+        private readonly Queue<byte> _buffer = new Queue<byte>();
+
+        /// <summary>
+        /// Indicates that the input stream has been flushed and that
+        /// all remaining data should be written to the output stream.
+        /// </summary>
+        private bool _isFlushed;
+
+        /// <summary>
+        /// Setting this to true will cause Read() to block if it appears
+        /// that it will run out of data.
+        /// </summary>
+        private bool _canBlockLastRead;
+
+        /// <summary>
+        /// Indicates whether the current <see cref="PipeStream"/> is disposed.
+        /// </summary>
+        private bool _isDisposed;
+
+        #endregion
+
+        #region Public properties
+
+        /// <summary>
+        /// Gets or sets the maximum number of bytes to store in the buffer.
+        /// </summary>
+        /// <value>The length of the max buffer.</value>
+        public long MaxBufferLength { get; set; } = 200 * 1024 * 1024;
+
+        /// <summary>
+        /// Gets or sets a value indicating whether to block last read method before the buffer is empty.
+        /// When true, Read() will block until it can fill the passed in buffer and count.
+        /// When false, Read() will not block, returning all the available buffer data.
+        /// </summary>
+        /// <remarks>
+        /// Setting to true will remove the possibility of ending a stream reader prematurely.
+        /// </remarks>
+        /// <value>
+        ///   <c>true</c> if block last read method before the buffer is empty; otherwise, <c>false</c>.
+        /// </value>
+        /// <exception cref="ObjectDisposedException">Methods were called after the stream was closed.</exception>
+        public bool BlockLastReadBuffer
+        {
+            get
+            {
+                CheckDisposed();
+
+                return _canBlockLastRead;
+            }
+            set
+            {
+                CheckDisposed();
+
+                _canBlockLastRead = value;
+
+                // when turning off the block last read, signal Read() that it may now read the rest of the buffer.
+                if (!_canBlockLastRead)
+                    lock (_buffer)
+                        Monitor.Pulse(_buffer);
+            }
+        }
+
+        #endregion
+
+        #region Stream overide methods
+
+        /// <summary>
+        /// When overridden in a derived class, clears all buffers for this stream and causes any buffered data to be written to the underlying device.
+        /// </summary>
+        /// <exception cref="IOException">An I/O error occurs.</exception>
+        /// <exception cref="ObjectDisposedException">Methods were called after the stream was closed.</exception>
+        /// <remarks>
+        /// Once flushed, any subsequent read operations no longer block until requested bytes are available. Any write operation reactivates blocking
+        /// reads.
+        /// </remarks>
+        public override void Flush()
+        {
+            CheckDisposed();
+
+            _isFlushed = true;
+            lock (_buffer)
+            {
+                // unblock read hereby allowing buffer to be partially filled
+                Monitor.Pulse(_buffer);
+            }
+        }
+
+        /// <summary>
+        /// When overridden in a derived class, sets the position within the current stream.
+        /// </summary>
+        /// <returns>
+        /// The new position within the current stream.
+        /// </returns>
+        /// <param name="offset">A byte offset relative to the origin parameter.</param>
+        /// <param name="origin">A value of type <see cref="SeekOrigin"/> indicating the reference point used to obtain the new position.</param>
+        /// <exception cref="NotSupportedException">The stream does not support seeking, such as if the stream is constructed from a pipe or console output.</exception>
+        public override long Seek(long offset, SeekOrigin origin)
+        {
+            throw new NotSupportedException();
+        }
+
+        /// <summary>
+        /// When overridden in a derived class, sets the length of the current stream.
+        /// </summary>
+        /// <param name="value">The desired length of the current stream in bytes.</param>
+        /// <exception cref="NotSupportedException">The stream does not support both writing and seeking, such as if the stream is constructed from a pipe or console output.</exception>
+        public override void SetLength(long value)
+        {
+            throw new NotSupportedException();
+        }
+
+        ///<summary>
+        ///When overridden in a derived class, reads a sequence of bytes from the current stream and advances the position within the stream by the number of bytes read.
+        ///</summary>
+        ///<returns>
+        ///The total number of bytes read into the buffer. This can be less than the number of bytes requested if that many bytes are not currently available, or zero if the stream is closed or end of the stream has been reached.
+        ///</returns>
+        ///<param name="offset">The zero-based byte offset in buffer at which to begin storing the data read from the current stream.</param>
+        ///<param name="count">The maximum number of bytes to be read from the current stream.</param>
+        ///<param name="buffer">An array of bytes. When this method returns, the buffer contains the specified byte array with the values between offset and (offset + count - 1) replaced by the bytes read from the current source.</param>
+        ///<exception cref="ArgumentException">The sum of offset and count is larger than the buffer length.</exception>
+        ///<exception cref="ObjectDisposedException">Methods were called after the stream was closed.</exception>
+        ///<exception cref="NotSupportedException">The stream does not support reading.</exception>
+        ///<exception cref="ArgumentNullException"><paramref name="buffer"/> is <c>null</c>.</exception>
+        ///<exception cref="IOException">An I/O error occurs.</exception>
+        ///<exception cref="ArgumentOutOfRangeException">offset or count is negative.</exception>
+        public override int Read(byte[] buffer, int offset, int count)
+        {
+            if (offset != 0)
+                throw new NotSupportedException("Offsets with value of non-zero are not supported");
+            if (buffer == null)
+                throw new ArgumentNullException("buffer");
+            if (offset + count > buffer.Length)
+                throw new ArgumentException("The sum of offset and count is greater than the buffer length.");
+            if (offset < 0 || count < 0)
+                throw new ArgumentOutOfRangeException("offset", "offset or count is negative.");
+            if (BlockLastReadBuffer && count >= MaxBufferLength)
+                throw new ArgumentException(string.Format(CultureInfo.CurrentCulture, "count({0}) > mMaxBufferLength({1})", count, MaxBufferLength));
+            CheckDisposed();
+            if (count == 0)
+                return 0;
+
+            var readLength = 0;
+
+            lock (_buffer)
+            {
+                while (!_isDisposed && !ReadAvailable(count))
+                {
+                    Monitor.Wait(_buffer);
+                }
+
+                // return zero when the read is interrupted by a close/dispose of the stream
+                if (_isDisposed)
+                {
+                    return 0;
+                }
+
+                // fill the read buffer
+                for (; readLength < count && _buffer.Count > 0; readLength++)
+                {
+                    buffer[readLength] = _buffer.Dequeue();
+                }
+
+                Monitor.Pulse(_buffer);
+            }
+
+            return readLength;
+        }
+
+        /// <summary>
+        /// Returns true if there are
+        /// </summary>
+        /// <param name="count">The count.</param>
+        /// <returns><c>True</c> if data available; otherwise<c>false</c>.</returns>
+        public bool ReadAvailable(int count)
+        {
+            var length = Length;
+            return (_isFlushed || length >= count) && (length >= (count + 1) || !BlockLastReadBuffer);
+        }
+
+        ///<summary>
+        ///When overridden in a derived class, writes a sequence of bytes to the current stream and advances the current position within this stream by the number of bytes written.
+        ///</summary>
+        ///<param name="offset">The zero-based byte offset in buffer at which to begin copying bytes to the current stream.</param>
+        ///<param name="count">The number of bytes to be written to the current stream.</param>
+        ///<param name="buffer">An array of bytes. This method copies count bytes from buffer to the current stream.</param>
+        ///<exception cref="IOException">An I/O error occurs.</exception>
+        ///<exception cref="NotSupportedException">The stream does not support writing.</exception>
+        ///<exception cref="ObjectDisposedException">Methods were called after the stream was closed.</exception>
+        ///<exception cref="ArgumentNullException"><paramref name="buffer"/> is <c>null</c>.</exception>
+        ///<exception cref="ArgumentException">The sum of offset and count is greater than the buffer length.</exception>
+        ///<exception cref="ArgumentOutOfRangeException">offset or count is negative.</exception>
+        public override void Write(byte[] buffer, int offset, int count)
+        {
+            if (buffer == null)
+                throw new ArgumentNullException("buffer");
+            if (offset + count > buffer.Length)
+                throw new ArgumentException("The sum of offset and count is greater than the buffer length.");
+            if (offset < 0 || count < 0)
+                throw new ArgumentOutOfRangeException("offset", "offset or count is negative.");
+            CheckDisposed();
+            if (count == 0)
+                return;
+
+            lock (_buffer)
+            {
+                // wait until the buffer isn't full
+                while (Length >= MaxBufferLength)
+                    Monitor.Wait(_buffer);
+
+                _isFlushed = false; // if it were flushed before, it soon will not be.
+
+                // queue up the buffer data
+                for (var i = offset; i < offset + count; i++)
+                {
+                    _buffer.Enqueue(buffer[i]);
+                }
+
+                Monitor.Pulse(_buffer); // signal that write has occurred
+            }
+        }
+
+        /// <summary>
+        /// Releases the unmanaged resources used by the Stream and optionally releases the managed resources.
+        /// </summary>
+        /// <param name="disposing"><c>true</c> to release both managed and unmanaged resources; <c>false</c> to release only unmanaged resources.</param>
+        /// <remarks>
+        /// Disposing a <see cref="PipeStream"/> will interrupt blocking read and write operations.
+        /// </remarks>
+        protected override void Dispose(bool disposing)
+        {
+            base.Dispose(disposing);
+
+            if (!_isDisposed)
+            {
+                lock (_buffer)
+                {
+                    _isDisposed = true;
+                    Monitor.Pulse(_buffer);
+                }
+            }
+        }
+
+        ///<summary>
+        ///When overridden in a derived class, gets a value indicating whether the current stream supports reading.
+        ///</summary>
+        ///<returns>
+        ///true if the stream supports reading; otherwise, false.
+        ///</returns>
+        public override bool CanRead
+        {
+            get { return !_isDisposed; }
+        }
+
+        /// <summary>
+        /// When overridden in a derived class, gets a value indicating whether the current stream supports seeking.
+        /// </summary>
+        /// <returns>
+        /// <c>true</c> if the stream supports seeking; otherwise, <c>false</c>.
+        ///</returns>
+        public override bool CanSeek
+        {
+            get { return false; }
+        }
+
+        /// <summary>
+        /// When overridden in a derived class, gets a value indicating whether the current stream supports writing.
+        /// </summary>
+        /// <returns>
+        /// <c>true</c> if the stream supports writing; otherwise, <c>false</c>.
+        /// </returns>
+        public override bool CanWrite
+        {
+            get { return !_isDisposed; }
+        }
+
+        /// <summary>
+        /// When overridden in a derived class, gets the length in bytes of the stream.
+        /// </summary>
+        /// <returns>
+        /// A long value representing the length of the stream in bytes.
+        /// </returns>
+        /// <exception cref="NotSupportedException">A class derived from Stream does not support seeking.</exception>
+        /// <exception cref="ObjectDisposedException">Methods were called after the stream was closed.</exception>
+        public override long Length
+        {
+            get
+            {
+                CheckDisposed();
+
+                return _buffer.Count;
+            }
+        }
+
+        /// <summary>
+        /// When overridden in a derived class, gets or sets the position within the current stream.
+        /// </summary>
+        /// <returns>
+        /// The current position within the stream.
+        /// </returns>
+        /// <exception cref="NotSupportedException">The stream does not support seeking.</exception>
+        public override long Position
+        {
+            get { return 0; }
+            set { throw new NotSupportedException(); }
+        }
+
+        #endregion
+
+        private void CheckDisposed()
+        {
+            if (_isDisposed)
+            {
+                throw new ObjectDisposedException(GetType().FullName);
+            }
+        }
+    }
+}

dotnet/internal/StdOutException.cs

@@ -0,0 +1,13 @@
+using System;
+
+namespace WinSCP
+{
+    internal class StdOutException : Exception
+    {
+        public StdOutException() :
+            // when the data are expected, the exception does not leave the API
+            base("Unexpected data")
+        {
+        }
+    }
+}

source/core/SessionInfo.cpp

@@ -460,6 +460,7 @@ private:
 //---------------------------------------------------------------------------
 __fastcall TSessionAction::TSessionAction(TActionLog * Log, TLogAction Action)
 {
+  FCancelled = false;
   if (Log->FLogging)
   {
     FRecord = new TSessionActionRecord(Log, Action);
@@ -504,10 +505,16 @@ void __fastcall TSessionAction::Cancel()
   {
     TSessionActionRecord * Record = FRecord;
     FRecord = NULL;
+    FCancelled = true;
     Record->Cancel();
   }
 }
 //---------------------------------------------------------------------------
+bool __fastcall TSessionAction::IsValid()
+{
+  return !FCancelled;
+}
+//---------------------------------------------------------------------------
 //---------------------------------------------------------------------------
 __fastcall TFileSessionAction::TFileSessionAction(TActionLog * Log, TLogAction Action) :
   TSessionAction(Log, Action)

source/core/SessionInfo.h

@@ -107,8 +107,11 @@ public:
   void __fastcall Rollback(Exception * E = NULL);
   void __fastcall Cancel();
 
+  bool __fastcall IsValid();
+
 protected:
   TSessionActionRecord * FRecord;
+  bool FCancelled;
 };
 //---------------------------------------------------------------------------
 class TFileSessionAction : public TSessionAction

source/core/SftpFileSystem.cpp

@@ -5607,11 +5607,7 @@ void __fastcall TSFTPFileSystem::Sink(
       // queue is discarded here
     }
 
-    if (CopyParam->OnTransferOut != NULL)
-    {
-      CopyParam->OnTransferOut(FTerminal, NULL, 0);
-    }
-    else
+    if (CopyParam->OnTransferOut == NULL)
     {
       DebugAssert(LocalHandle);
       if (CopyParam->PreserveTime)

source/core/Terminal.cpp

@@ -7387,59 +7387,70 @@ void __fastcall TTerminal::SinkRobust(
   const TCopyParamType * CopyParam, int Params, TFileOperationProgressType * OperationProgress, unsigned int Flags)
 {
   TDownloadSessionAction Action(ActionLog);
-  bool * AFileTransferAny = FLAGSET(Flags, tfUseFileTransferAny) ? &FFileTransferAny : NULL;
-  bool CanRetry = (CopyParam->OnTransferOut == NULL);
-  TRobustOperationLoop RobustLoop(this, OperationProgress, AFileTransferAny, CanRetry);
-  bool Sunk = false;
-
-  do
+  try
   {
-    try
+    bool * AFileTransferAny = FLAGSET(Flags, tfUseFileTransferAny) ? &FFileTransferAny : NULL;
+    bool CanRetry = (CopyParam->OnTransferOut == NULL);
+    TRobustOperationLoop RobustLoop(this, OperationProgress, AFileTransferAny, CanRetry);
+    bool Sunk = false;
+
+    do
     {
-      // If connection is lost while deleting the file, so not retry download on the next round.
-      // The file may not exist anymore and the download attempt might overwrite the (only) local copy.
-      if (!Sunk)
+      try
       {
-        Sink(FileName, File, TargetDir, CopyParam, Params, OperationProgress, Flags, Action);
-        Sunk = true;
-      }
+        // If connection is lost while deleting the file, so not retry download on the next round.
+        // The file may not exist anymore and the download attempt might overwrite the (only) local copy.
+        if (!Sunk)
+        {
+          Sink(FileName, File, TargetDir, CopyParam, Params, OperationProgress, Flags, Action);
+          Sunk = true;
+        }
 
-      if (FLAGSET(Params, cpDelete))
-      {
-        DebugAssert(FLAGCLEAR(Params, cpNoRecurse));
-        // If file is directory, do not delete it recursively, because it should be
-        // empty already. If not, it should not be deleted (some files were
-        // skipped or some new files were copied to it, while we were downloading)
-        int Params = dfNoRecursive;
-        DeleteFile(FileName, File, &Params);
+        if (FLAGSET(Params, cpDelete))
+        {
+          DebugAssert(FLAGCLEAR(Params, cpNoRecurse));
+          // If file is directory, do not delete it recursively, because it should be
+          // empty already. If not, it should not be deleted (some files were
+          // skipped or some new files were copied to it, while we were downloading)
+          int Params = dfNoRecursive;
+          DeleteFile(FileName, File, &Params);
+        }
       }
-    }
-    catch (Exception & E)
-    {
-      if (!RobustLoop.TryReopen(E))
+      catch (Exception & E)
       {
-        if (!Sunk)
+        if (!RobustLoop.TryReopen(E))
         {
-          RollbackAction(Action, OperationProgress, &E);
+          if (!Sunk)
+          {
+            RollbackAction(Action, OperationProgress, &E);
+          }
+          throw;
         }
-        throw;
       }
-    }
 
-    if (RobustLoop.ShouldRetry())
-    {
-      OperationProgress->RollbackTransfer();
-      Action.Restart();
-      DebugAssert(File != NULL);
-      if (!File->IsDirectory)
+      if (RobustLoop.ShouldRetry())
       {
-        // prevent overwrite and resume confirmations
-        Params |= cpNoConfirmation;
-        Flags |= tfAutoResume;
+        OperationProgress->RollbackTransfer();
+        Action.Restart();
+        DebugAssert(File != NULL);
+        if (!File->IsDirectory)
+        {
+          // prevent overwrite and resume confirmations
+          Params |= cpNoConfirmation;
+          Flags |= tfAutoResume;
+        }
       }
     }
+    while (RobustLoop.Retry());
+  }
+  __finally
+  {
+    // Once we issue <download> we must terminate the chunked stream
+    if (Action.IsValid() && (CopyParam->OnTransferOut != NULL) && DebugAlwaysTrue(IsCapable[fcTransferOut]))
+    {
+      CopyParam->OnTransferOut(this, NULL, 0);
+    }
   }
-  while (RobustLoop.Retry());
 }
 //---------------------------------------------------------------------------
 struct TSinkFileParams