| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349 |
- using System;
- using System.Diagnostics.CodeAnalysis;
- using System.Globalization;
- using System.IO;
- using System.Threading;
- using System.Threading.Tasks;
- using Renci.SshNet.Common;
- namespace Renci.SshNet.Sftp
- {
- /// <summary>
- /// Exposes a <see cref="Stream"/> around a remote SFTP file, supporting both synchronous and asynchronous read and write operations.
- /// </summary>
- /// <threadsafety static="true" instance="false"/>
- #pragma warning disable IDE0079 // We intentionally want to suppress the below warning.
- [SuppressMessage("Performance", "CA1844: Provide memory-based overrides of async methods when subclassing 'Stream'", Justification = "TODO: This should be addressed in the future.")]
- #pragma warning restore IDE0079
- public class SftpFileStream : Stream
- {
- private readonly Lock _lock = new Lock();
- private readonly int _readBufferSize;
- private readonly int _writeBufferSize;
- // Internal state.
- private byte[] _handle;
- private ISftpSession _session;
- // Buffer information.
- private byte[] _readBuffer;
- private byte[] _writeBuffer;
- private int _bufferPosition;
- private int _bufferLen;
- private long _position;
- private bool _bufferOwnedByWrite;
- private bool _canRead;
- private bool _canSeek;
- private bool _canWrite;
- private TimeSpan _timeout;
- /// <summary>
- /// Gets a value indicating whether the current stream supports reading.
- /// </summary>
- /// <value>
- /// <see langword="true"/> if the stream supports reading; otherwise, <see langword="false"/>.
- /// </value>
- public override bool CanRead
- {
- get { return _canRead; }
- }
- /// <summary>
- /// Gets a value indicating whether the current stream supports seeking.
- /// </summary>
- /// <value>
- /// <see langword="true"/> if the stream supports seeking; otherwise, <see langword="false"/>.
- /// </value>
- public override bool CanSeek
- {
- get { return _canSeek; }
- }
- /// <summary>
- /// Gets a value indicating whether the current stream supports writing.
- /// </summary>
- /// <value>
- /// <see langword="true"/> if the stream supports writing; otherwise, <see langword="false"/>.
- /// </value>
- public override bool CanWrite
- {
- get { return _canWrite; }
- }
- /// <summary>
- /// Gets a value indicating whether timeout properties are usable for <see cref="SftpFileStream"/>.
- /// </summary>
- /// <value>
- /// <see langword="true"/> in all cases.
- /// </value>
- public override bool CanTimeout
- {
- get { return true; }
- }
- /// <summary>
- /// Gets the length in bytes of the stream.
- /// </summary>
- /// <value>A long value representing the length of the stream in bytes.</value>
- /// <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>
- /// <exception cref="IOException">IO operation failed. </exception>
- public override long Length
- {
- get
- {
- // Lock down the file stream while we do this.
- lock (_lock)
- {
- CheckSessionIsOpen();
- if (!CanSeek)
- {
- throw new NotSupportedException("Seek operation is not supported.");
- }
- // Flush the write buffer, because it may
- // affect the length of the stream.
- if (_bufferOwnedByWrite)
- {
- FlushWriteBuffer();
- }
- // obtain file attributes
- var attributes = _session.RequestFStat(_handle, nullOnError: true);
- if (attributes != null)
- {
- return attributes.Size;
- }
- throw new IOException("Seek operation failed.");
- }
- }
- }
- /// <summary>
- /// Gets or sets the position within the current stream.
- /// </summary>
- /// <value>The current position within the stream.</value>
- /// <exception cref="IOException">An I/O error occurs. </exception>
- /// <exception cref="NotSupportedException">The stream does not support seeking. </exception>
- /// <exception cref="ObjectDisposedException">Methods were called after the stream was closed. </exception>
- public override long Position
- {
- get
- {
- CheckSessionIsOpen();
- if (!CanSeek)
- {
- throw new NotSupportedException("Seek operation not supported.");
- }
- return _position;
- }
- set
- {
- _ = Seek(value, SeekOrigin.Begin);
- }
- }
- /// <summary>
- /// Gets the name of the path that was used to construct the current <see cref="SftpFileStream"/>.
- /// </summary>
- /// <value>
- /// The name of the path that was used to construct the current <see cref="SftpFileStream"/>.
- /// </value>
- public string Name { get; private set; }
- /// <summary>
- /// Gets the operating system file handle for the file that the current <see cref="SftpFileStream"/> encapsulates.
- /// </summary>
- /// <value>
- /// The operating system file handle for the file that the current <see cref="SftpFileStream"/> encapsulates.
- /// </value>
- public virtual byte[] Handle
- {
- get
- {
- Flush();
- return _handle;
- }
- }
- /// <summary>
- /// Gets or sets the operation timeout.
- /// </summary>
- /// <value>
- /// The timeout.
- /// </value>
- public TimeSpan Timeout
- {
- get
- {
- return _timeout;
- }
- set
- {
- value.EnsureValidTimeout(nameof(Timeout));
- _timeout = value;
- }
- }
- private SftpFileStream(ISftpSession session, string path, FileAccess access, int bufferSize, byte[] handle, long position)
- {
- Timeout = TimeSpan.FromSeconds(30);
- Name = path;
- _session = session;
- _canRead = (access & FileAccess.Read) == FileAccess.Read;
- _canSeek = true;
- _canWrite = (access & FileAccess.Write) == FileAccess.Write;
- _handle = handle;
- /*
- * Instead of using the specified buffer size as is, we use it to calculate a buffer size
- * that ensures we always receive or send the max. number of bytes in a single SSH_FXP_READ
- * or SSH_FXP_WRITE message.
- */
- _readBufferSize = (int)session.CalculateOptimalReadLength((uint)bufferSize);
- _writeBufferSize = (int)session.CalculateOptimalWriteLength((uint)bufferSize, _handle);
- _position = position;
- }
- internal SftpFileStream(ISftpSession session, string path, FileMode mode, FileAccess access, int bufferSize)
- {
- if (session is null)
- {
- throw new SshConnectionException("Client not connected.");
- }
- ThrowHelper.ThrowIfNull(path);
- if (bufferSize <= 0)
- {
- throw new ArgumentOutOfRangeException(nameof(bufferSize), "Cannot be less than or equal to zero.");
- }
- Timeout = TimeSpan.FromSeconds(30);
- Name = path;
- // Initialize the object state.
- _session = session;
- _canRead = (access & FileAccess.Read) == FileAccess.Read;
- _canSeek = true;
- _canWrite = (access & FileAccess.Write) == FileAccess.Write;
- var flags = Flags.None;
- switch (access)
- {
- case FileAccess.Read:
- flags |= Flags.Read;
- break;
- case FileAccess.Write:
- flags |= Flags.Write;
- break;
- case FileAccess.ReadWrite:
- flags |= Flags.Read;
- flags |= Flags.Write;
- break;
- default:
- throw new ArgumentOutOfRangeException(nameof(access));
- }
- if ((access & FileAccess.Read) == FileAccess.Read && mode == FileMode.Append)
- {
- throw new ArgumentException(string.Format(CultureInfo.InvariantCulture,
- "{0} mode can be requested only when combined with write-only access.",
- mode.ToString("G")),
- nameof(mode));
- }
- if ((access & FileAccess.Write) != FileAccess.Write)
- {
- if (mode is FileMode.Create or FileMode.CreateNew or FileMode.Truncate or FileMode.Append)
- {
- throw new ArgumentException(string.Format(CultureInfo.InvariantCulture,
- "Combining {0}: {1} with {2}: {3} is invalid.",
- nameof(FileMode),
- mode,
- nameof(FileAccess),
- access),
- nameof(mode));
- }
- }
- switch (mode)
- {
- case FileMode.Append:
- flags |= Flags.Append | Flags.CreateNewOrOpen;
- break;
- case FileMode.Create:
- _handle = _session.RequestOpen(path, flags | Flags.Truncate, nullOnError: true);
- if (_handle is null)
- {
- flags |= Flags.CreateNew;
- }
- else
- {
- flags |= Flags.Truncate;
- }
- break;
- case FileMode.CreateNew:
- flags |= Flags.CreateNew;
- break;
- case FileMode.Open:
- break;
- case FileMode.OpenOrCreate:
- flags |= Flags.CreateNewOrOpen;
- break;
- case FileMode.Truncate:
- flags |= Flags.Truncate;
- break;
- default:
- throw new ArgumentOutOfRangeException(nameof(mode));
- }
- _handle ??= _session.RequestOpen(path, flags);
- /*
- * Instead of using the specified buffer size as is, we use it to calculate a buffer size
- * that ensures we always receive or send the max. number of bytes in a single SSH_FXP_READ
- * or SSH_FXP_WRITE message.
- */
- _readBufferSize = (int)session.CalculateOptimalReadLength((uint)bufferSize);
- _writeBufferSize = (int)session.CalculateOptimalWriteLength((uint)bufferSize, _handle);
- if (mode == FileMode.Append)
- {
- var attributes = _session.RequestFStat(_handle, nullOnError: false);
- _position = attributes.Size;
- }
- }
- internal static async Task<SftpFileStream> OpenAsync(ISftpSession session, string path, FileMode mode, FileAccess access, int bufferSize, CancellationToken cancellationToken)
- {
- if (session is null)
- {
- throw new SshConnectionException("Client not connected.");
- }
- ThrowHelper.ThrowIfNull(path);
- if (bufferSize <= 0)
- {
- throw new ArgumentOutOfRangeException(nameof(bufferSize), "Cannot be less than or equal to zero.");
- }
- var flags = Flags.None;
- switch (access)
- {
- case FileAccess.Read:
- flags |= Flags.Read;
- break;
- case FileAccess.Write:
- flags |= Flags.Write;
- break;
- case FileAccess.ReadWrite:
- flags |= Flags.Read;
- flags |= Flags.Write;
- break;
- default:
- throw new ArgumentOutOfRangeException(nameof(access));
- }
- if ((access & FileAccess.Read) == FileAccess.Read && mode == FileMode.Append)
- {
- throw new ArgumentException(string.Format(CultureInfo.InvariantCulture,
- "{0} mode can be requested only when combined with write-only access.",
- mode.ToString("G")),
- nameof(mode));
- }
- if ((access & FileAccess.Write) != FileAccess.Write)
- {
- if (mode is FileMode.Create or FileMode.CreateNew or FileMode.Truncate or FileMode.Append)
- {
- throw new ArgumentException(string.Format(CultureInfo.InvariantCulture,
- "Combining {0}: {1} with {2}: {3} is invalid.",
- nameof(FileMode),
- mode,
- nameof(FileAccess),
- access),
- nameof(mode));
- }
- }
- switch (mode)
- {
- case FileMode.Append:
- flags |= Flags.Append | Flags.CreateNewOrOpen;
- break;
- case FileMode.Create:
- flags |= Flags.CreateNewOrOpen | Flags.Truncate;
- break;
- case FileMode.CreateNew:
- flags |= Flags.CreateNew;
- break;
- case FileMode.Open:
- break;
- case FileMode.OpenOrCreate:
- flags |= Flags.CreateNewOrOpen;
- break;
- case FileMode.Truncate:
- flags |= Flags.Truncate;
- break;
- default:
- throw new ArgumentOutOfRangeException(nameof(mode));
- }
- var handle = await session.RequestOpenAsync(path, flags, cancellationToken).ConfigureAwait(false);
- long position = 0;
- if (mode == FileMode.Append)
- {
- try
- {
- var attributes = await session.RequestFStatAsync(handle, cancellationToken).ConfigureAwait(false);
- position = attributes.Size;
- }
- catch
- {
- try
- {
- await session.RequestCloseAsync(handle, cancellationToken).ConfigureAwait(false);
- }
- catch
- {
- // The original exception is presumably more informative, so we just ignore this one.
- }
- throw;
- }
- }
- return new SftpFileStream(session, path, access, bufferSize, handle, position);
- }
- /// <summary>
- /// Clears all buffers for this stream and causes any buffered data to be written to the file.
- /// </summary>
- /// <exception cref="IOException">An I/O error occurs. </exception>
- /// <exception cref="ObjectDisposedException">Stream is closed.</exception>
- public override void Flush()
- {
- lock (_lock)
- {
- CheckSessionIsOpen();
- if (_bufferOwnedByWrite)
- {
- FlushWriteBuffer();
- }
- else
- {
- FlushReadBuffer();
- }
- }
- }
- /// <summary>
- /// Asynchronously clears all buffers for this stream and causes any buffered data to be written to the file.
- /// </summary>
- /// <param name="cancellationToken">The <see cref="CancellationToken"/> to observe.</param>
- /// <returns>A <see cref="Task"/> that represents the asynchronous flush operation.</returns>
- /// <exception cref="IOException">An I/O error occurs. </exception>
- /// <exception cref="ObjectDisposedException">Stream is closed.</exception>
- public override Task FlushAsync(CancellationToken cancellationToken)
- {
- CheckSessionIsOpen();
- if (_bufferOwnedByWrite)
- {
- return FlushWriteBufferAsync(cancellationToken);
- }
- FlushReadBuffer();
- return Task.CompletedTask;
- }
- /// <summary>
- /// Reads a sequence of bytes from the current stream and advances the position within the stream by the
- /// number of bytes read.
- /// </summary>
- /// <param name="buffer">An array of bytes. When this method returns, the buffer contains the specified byte array with the values between <paramref name="offset"/> and (<paramref name="offset"/> + <paramref name="count"/> - 1) replaced by the bytes read from the current source.</param>
- /// <param name="offset">The zero-based byte offset in <paramref name="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>
- /// <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 (0) if the end of the stream has been reached.
- /// </returns>
- /// <exception cref="ArgumentException">The sum of <paramref name="offset"/> and <paramref name="count"/> is larger than the buffer length.</exception>
- /// <exception cref="ArgumentNullException"><paramref name="buffer"/> is <see langword="null"/>. </exception>
- /// <exception cref="ArgumentOutOfRangeException"><paramref name="offset"/> or <paramref name="count"/> is negative.</exception>
- /// <exception cref="IOException">An I/O error occurs. </exception>
- /// <exception cref="NotSupportedException">The stream does not support reading. </exception>
- /// <exception cref="ObjectDisposedException">Methods were called after the stream was closed. </exception>
- /// <remarks>
- /// <para>
- /// This method attempts to read up to <paramref name="count"/> bytes. This either from the buffer, from the
- /// server (using one or more <c>SSH_FXP_READ</c> requests) or using a combination of both.
- /// </para>
- /// <para>
- /// The read loop is interrupted when either <paramref name="count"/> bytes are read, the server returns zero
- /// bytes (EOF) or less bytes than the read buffer size.
- /// </para>
- /// <para>
- /// When a server returns less number of bytes than the read buffer size, this <c>may</c> indicate that EOF has
- /// been reached. A subsequent (<c>SSH_FXP_READ</c>) server request is necessary to make sure EOF has effectively
- /// been reached. Breaking out of the read loop avoids reading from the server twice to determine EOF: once in
- /// the read loop, and once upon the next <see cref="Read"/> or <see cref="ReadByte"/> invocation.
- /// </para>
- /// </remarks>
- public override int Read(byte[] buffer, int offset, int count)
- {
- var readLen = 0;
- ThrowHelper.ThrowIfNull(buffer);
- #if NET8_0_OR_GREATER
- ArgumentOutOfRangeException.ThrowIfNegative(offset);
- ArgumentOutOfRangeException.ThrowIfNegative(count);
- #else
- if (offset < 0)
- {
- throw new ArgumentOutOfRangeException(nameof(offset));
- }
- if (count < 0)
- {
- throw new ArgumentOutOfRangeException(nameof(count));
- }
- #endif
- if ((buffer.Length - offset) < count)
- {
- throw new ArgumentException("Invalid array range.");
- }
- // Lock down the file stream while we do this.
- lock (_lock)
- {
- CheckSessionIsOpen();
- // Set up for the read operation.
- SetupRead();
- // Read data into the caller's buffer.
- while (count > 0)
- {
- // How much data do we have available in the buffer?
- var bytesAvailableInBuffer = _bufferLen - _bufferPosition;
- if (bytesAvailableInBuffer <= 0)
- {
- var data = _session.RequestRead(_handle, (ulong)_position, (uint)_readBufferSize);
- if (data.Length == 0)
- {
- _bufferPosition = 0;
- _bufferLen = 0;
- break;
- }
- var bytesToWriteToCallerBuffer = count;
- if (bytesToWriteToCallerBuffer >= data.Length)
- {
- // write all data read to caller-provided buffer
- bytesToWriteToCallerBuffer = data.Length;
- // reset buffer since we will skip buffering
- _bufferPosition = 0;
- _bufferLen = 0;
- }
- else
- {
- // determine number of bytes that we should write into read buffer
- var bytesToWriteToReadBuffer = data.Length - bytesToWriteToCallerBuffer;
- // write remaining bytes to read buffer
- Buffer.BlockCopy(data, count, GetOrCreateReadBuffer(), 0, bytesToWriteToReadBuffer);
- // update position in read buffer
- _bufferPosition = 0;
- // update number of bytes in read buffer
- _bufferLen = bytesToWriteToReadBuffer;
- }
- // write bytes to caller-provided buffer
- Buffer.BlockCopy(data, 0, buffer, offset, bytesToWriteToCallerBuffer);
- // update stream position
- _position += bytesToWriteToCallerBuffer;
- // record total number of bytes read into caller-provided buffer
- readLen += bytesToWriteToCallerBuffer;
- // break out of the read loop when the server returned less than the request number of bytes
- // as that *may* indicate that we've reached EOF
- //
- // doing this avoids reading from server twice to determine EOF: once in the read loop, and
- // once upon the next Read or ReadByte invocation by the caller
- if (data.Length < _readBufferSize)
- {
- break;
- }
- // advance offset to start writing bytes into caller-provided buffer
- offset += bytesToWriteToCallerBuffer;
- // update number of bytes left to read into caller-provided buffer
- count -= bytesToWriteToCallerBuffer;
- }
- else
- {
- // limit the number of bytes to use from read buffer to the caller-request number of bytes
- if (bytesAvailableInBuffer > count)
- {
- bytesAvailableInBuffer = count;
- }
- // copy data from read buffer to the caller-provided buffer
- Buffer.BlockCopy(GetOrCreateReadBuffer(), _bufferPosition, buffer, offset, bytesAvailableInBuffer);
- // update position in read buffer
- _bufferPosition += bytesAvailableInBuffer;
- // update stream position
- _position += bytesAvailableInBuffer;
- // record total number of bytes read into caller-provided buffer
- readLen += bytesAvailableInBuffer;
- // advance offset to start writing bytes into caller-provided buffer
- offset += bytesAvailableInBuffer;
- // update number of bytes left to read
- count -= bytesAvailableInBuffer;
- }
- }
- }
- // return the number of bytes that were read to the caller.
- return readLen;
- }
- /// <summary>
- /// Asynchronously reads a sequence of bytes from the current stream and advances the position within the stream by the
- /// number of bytes read.
- /// </summary>
- /// <param name="buffer">An array of bytes. When this method returns, the buffer contains the specified byte array with the values between <paramref name="offset"/> and (<paramref name="offset"/> + <paramref name="count"/> - 1) replaced by the bytes read from the current source.</param>
- /// <param name="offset">The zero-based byte offset in <paramref name="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="cancellationToken">The <see cref="CancellationToken" /> to observe.</param>
- /// <returns>A <see cref="Task" /> that represents the asynchronous read operation.</returns>
- public override async Task<int> ReadAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
- {
- var readLen = 0;
- ThrowHelper.ThrowIfNull(buffer);
- #if NET8_0_OR_GREATER
- ArgumentOutOfRangeException.ThrowIfNegative(offset);
- ArgumentOutOfRangeException.ThrowIfNegative(count);
- #else
- if (offset < 0)
- {
- throw new ArgumentOutOfRangeException(nameof(offset));
- }
- if (count < 0)
- {
- throw new ArgumentOutOfRangeException(nameof(count));
- }
- #endif
- if ((buffer.Length - offset) < count)
- {
- throw new ArgumentException("Invalid array range.");
- }
- CheckSessionIsOpen();
- // Set up for the read operation.
- SetupRead();
- // Read data into the caller's buffer.
- while (count > 0)
- {
- // How much data do we have available in the buffer?
- var bytesAvailableInBuffer = _bufferLen - _bufferPosition;
- if (bytesAvailableInBuffer <= 0)
- {
- var data = await _session.RequestReadAsync(_handle, (ulong)_position, (uint)_readBufferSize, cancellationToken).ConfigureAwait(false);
- if (data.Length == 0)
- {
- _bufferPosition = 0;
- _bufferLen = 0;
- break;
- }
- var bytesToWriteToCallerBuffer = count;
- if (bytesToWriteToCallerBuffer >= data.Length)
- {
- // write all data read to caller-provided buffer
- bytesToWriteToCallerBuffer = data.Length;
- // reset buffer since we will skip buffering
- _bufferPosition = 0;
- _bufferLen = 0;
- }
- else
- {
- // determine number of bytes that we should write into read buffer
- var bytesToWriteToReadBuffer = data.Length - bytesToWriteToCallerBuffer;
- // write remaining bytes to read buffer
- Buffer.BlockCopy(data, count, GetOrCreateReadBuffer(), 0, bytesToWriteToReadBuffer);
- // update position in read buffer
- _bufferPosition = 0;
- // update number of bytes in read buffer
- _bufferLen = bytesToWriteToReadBuffer;
- }
- // write bytes to caller-provided buffer
- Buffer.BlockCopy(data, 0, buffer, offset, bytesToWriteToCallerBuffer);
- // update stream position
- _position += bytesToWriteToCallerBuffer;
- // record total number of bytes read into caller-provided buffer
- readLen += bytesToWriteToCallerBuffer;
- // break out of the read loop when the server returned less than the request number of bytes
- // as that *may* indicate that we've reached EOF
- //
- // doing this avoids reading from server twice to determine EOF: once in the read loop, and
- // once upon the next Read or ReadByte invocation by the caller
- if (data.Length < _readBufferSize)
- {
- break;
- }
- // advance offset to start writing bytes into caller-provided buffer
- offset += bytesToWriteToCallerBuffer;
- // update number of bytes left to read into caller-provided buffer
- count -= bytesToWriteToCallerBuffer;
- }
- else
- {
- // limit the number of bytes to use from read buffer to the caller-request number of bytes
- if (bytesAvailableInBuffer > count)
- {
- bytesAvailableInBuffer = count;
- }
- // copy data from read buffer to the caller-provided buffer
- Buffer.BlockCopy(GetOrCreateReadBuffer(), _bufferPosition, buffer, offset, bytesAvailableInBuffer);
- // update position in read buffer
- _bufferPosition += bytesAvailableInBuffer;
- // update stream position
- _position += bytesAvailableInBuffer;
- // record total number of bytes read into caller-provided buffer
- readLen += bytesAvailableInBuffer;
- // advance offset to start writing bytes into caller-provided buffer
- offset += bytesAvailableInBuffer;
- // update number of bytes left to read
- count -= bytesAvailableInBuffer;
- }
- }
- // return the number of bytes that were read to the caller.
- return readLen;
- }
- /// <summary>
- /// Reads a byte from the stream and advances the position within the stream by one byte, or returns -1 if at the end of the stream.
- /// </summary>
- /// <returns>
- /// The unsigned byte cast to an <see cref="int"/>, or -1 if at the end of the stream.
- /// </returns>
- /// <exception cref="NotSupportedException">The stream does not support reading. </exception>
- /// <exception cref="ObjectDisposedException">Methods were called after the stream was closed. </exception>
- /// <exception cref="IOException">Read operation failed.</exception>
- public override int ReadByte()
- {
- // Lock down the file stream while we do this.
- lock (_lock)
- {
- CheckSessionIsOpen();
- // Setup the object for reading.
- SetupRead();
- byte[] readBuffer;
- // Read more data into the internal buffer if necessary.
- if (_bufferPosition >= _bufferLen)
- {
- var data = _session.RequestRead(_handle, (ulong)_position, (uint)_readBufferSize);
- if (data.Length == 0)
- {
- // We've reached EOF.
- return -1;
- }
- readBuffer = GetOrCreateReadBuffer();
- Buffer.BlockCopy(data, 0, readBuffer, 0, data.Length);
- _bufferPosition = 0;
- _bufferLen = data.Length;
- }
- else
- {
- readBuffer = GetOrCreateReadBuffer();
- }
- // Extract the next byte from the buffer.
- ++_position;
- return readBuffer[_bufferPosition++];
- }
- }
- /// <summary>
- /// Sets the position within the current stream.
- /// </summary>
- /// <param name="offset">A byte offset relative to the <paramref name="origin"/> parameter.</param>
- /// <param name="origin">A value of type <see cref="SeekOrigin"/> indicating the reference point used to obtain the new position.</param>
- /// <returns>
- /// The new position within the current stream.
- /// </returns>
- /// <exception cref="IOException">An I/O error occurs. </exception>
- /// <exception cref="NotSupportedException">The stream does not support seeking, such as if the stream is constructed from a pipe or console output. </exception>
- /// <exception cref="ObjectDisposedException">Methods were called after the stream was closed. </exception>
- public override long Seek(long offset, SeekOrigin origin)
- {
- long newPosn;
- // Lock down the file stream while we do this.
- lock (_lock)
- {
- CheckSessionIsOpen();
- if (!CanSeek)
- {
- throw new NotSupportedException("Seek is not supported.");
- }
- // Don't do anything if the position won't be moving.
- if (origin == SeekOrigin.Begin && offset == _position)
- {
- return offset;
- }
- if (origin == SeekOrigin.Current && offset == 0)
- {
- return _position;
- }
- // The behaviour depends upon the read/write mode.
- if (_bufferOwnedByWrite)
- {
- // Flush the write buffer and then seek.
- FlushWriteBuffer();
- }
- else
- {
- // Determine if the seek is to somewhere inside
- // the current read buffer bounds.
- if (origin == SeekOrigin.Begin)
- {
- newPosn = _position - _bufferPosition;
- if (offset >= newPosn && offset < (newPosn + _bufferLen))
- {
- _bufferPosition = (int)(offset - newPosn);
- _position = offset;
- return _position;
- }
- }
- else if (origin == SeekOrigin.Current)
- {
- newPosn = _position + offset;
- if (newPosn >= (_position - _bufferPosition) &&
- newPosn < (_position - _bufferPosition + _bufferLen))
- {
- _bufferPosition = (int)(newPosn - (_position - _bufferPosition));
- _position = newPosn;
- return _position;
- }
- }
- // Abandon the read buffer.
- _bufferPosition = 0;
- _bufferLen = 0;
- }
- // Seek to the new position.
- switch (origin)
- {
- case SeekOrigin.Begin:
- newPosn = offset;
- break;
- case SeekOrigin.Current:
- newPosn = _position + offset;
- break;
- case SeekOrigin.End:
- var attributes = _session.RequestFStat(_handle, nullOnError: false);
- newPosn = attributes.Size + offset;
- break;
- default:
- throw new ArgumentException("Invalid seek origin.", nameof(origin));
- }
- if (newPosn < 0)
- {
- throw new EndOfStreamException();
- }
- _position = newPosn;
- return _position;
- }
- }
- /// <summary>
- /// Sets the length of the current stream.
- /// </summary>
- /// <param name="value">The desired length of the current stream in bytes.</param>
- /// <exception cref="IOException">An I/O error occurs.</exception>
- /// <exception cref="NotSupportedException">The stream does not support both writing and seeking.</exception>
- /// <exception cref="ObjectDisposedException">Methods were called after the stream was closed.</exception>
- /// <exception cref="ArgumentOutOfRangeException"><paramref name="value"/> must be greater than zero.</exception>
- /// <remarks>
- /// <para>
- /// Buffers are first flushed.
- /// </para>
- /// <para>
- /// If the specified value is less than the current length of the stream, the stream is truncated and - if the
- /// current position is greater than the new length - the current position is moved to the last byte of the stream.
- /// </para>
- /// <para>
- /// If the given value is greater than the current length of the stream, the stream is expanded and the current
- /// position remains the same.
- /// </para>
- /// </remarks>
- public override void SetLength(long value)
- {
- #if NET8_0_OR_GREATER
- ArgumentOutOfRangeException.ThrowIfNegative(value);
- #else
- if (value < 0)
- {
- throw new ArgumentOutOfRangeException(nameof(value));
- }
- #endif
- // Lock down the file stream while we do this.
- lock (_lock)
- {
- CheckSessionIsOpen();
- if (!CanSeek)
- {
- throw new NotSupportedException("Seek is not supported.");
- }
- if (_bufferOwnedByWrite)
- {
- FlushWriteBuffer();
- }
- else
- {
- SetupWrite();
- }
- var attributes = _session.RequestFStat(_handle, nullOnError: false);
- attributes.Size = value;
- _session.RequestFSetStat(_handle, attributes);
- if (_position > value)
- {
- _position = value;
- }
- }
- }
- /// <summary>
- /// 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="buffer">An array of bytes. This method copies <paramref name="count"/> bytes from <paramref name="buffer"/> to the current stream.</param>
- /// <param name="offset">The zero-based byte offset in <paramref name="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>
- /// <exception cref="ArgumentException">The sum of <paramref name="offset"/> and <paramref name="count"/> is greater than the buffer length.</exception>
- /// <exception cref="ArgumentNullException"><paramref name="buffer"/> is <see langword="null"/>.</exception>
- /// <exception cref="ArgumentOutOfRangeException"><paramref name="offset"/> or <paramref name="count"/> is negative.</exception>
- /// <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>
- public override void Write(byte[] buffer, int offset, int count)
- {
- ThrowHelper.ThrowIfNull(buffer);
- #if NET8_0_OR_GREATER
- ArgumentOutOfRangeException.ThrowIfNegative(offset);
- ArgumentOutOfRangeException.ThrowIfNegative(count);
- #else
- if (offset < 0)
- {
- throw new ArgumentOutOfRangeException(nameof(offset));
- }
- if (count < 0)
- {
- throw new ArgumentOutOfRangeException(nameof(count));
- }
- #endif
- if ((buffer.Length - offset) < count)
- {
- throw new ArgumentException("Invalid array range.");
- }
- // Lock down the file stream while we do this.
- lock (_lock)
- {
- CheckSessionIsOpen();
- // Setup this object for writing.
- SetupWrite();
- // Write data to the file stream.
- while (count > 0)
- {
- // Determine how many bytes we can write to the buffer.
- var tempLen = _writeBufferSize - _bufferPosition;
- if (tempLen <= 0)
- {
- // flush write buffer, and mark it empty
- FlushWriteBuffer();
- // we can now write or buffer the full buffer size
- tempLen = _writeBufferSize;
- }
- // limit the number of bytes to write to the actual number of bytes requested
- if (tempLen > count)
- {
- tempLen = count;
- }
- // Can we short-cut the internal buffer?
- if (_bufferPosition == 0 && tempLen == _writeBufferSize)
- {
- using (var wait = new AutoResetEvent(initialState: false))
- {
- _session.RequestWrite(_handle, (ulong)_position, buffer, offset, tempLen, wait);
- }
- }
- else
- {
- // No: copy the data to the write buffer first.
- Buffer.BlockCopy(buffer, offset, GetOrCreateWriteBuffer(), _bufferPosition, tempLen);
- _bufferPosition += tempLen;
- }
- // Advance the buffer and stream positions.
- _position += tempLen;
- offset += tempLen;
- count -= tempLen;
- }
- // If the buffer is full, then do a speculative flush now,
- // rather than waiting for the next call to this method.
- if (_bufferPosition >= _writeBufferSize)
- {
- using (var wait = new AutoResetEvent(initialState: false))
- {
- _session.RequestWrite(_handle, (ulong)(_position - _bufferPosition), GetOrCreateWriteBuffer(), 0, _bufferPosition, wait);
- }
- _bufferPosition = 0;
- }
- }
- }
- /// <summary>
- /// Asynchronously 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="buffer">An array of bytes. This method copies <paramref name="count"/> bytes from <paramref name="buffer"/> to the current stream.</param>
- /// <param name="offset">The zero-based byte offset in <paramref name="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="cancellationToken">The <see cref="CancellationToken"/> to observe.</param>
- /// <returns>A <see cref="Task"/> that represents the asynchronous write operation.</returns>
- /// <exception cref="ArgumentException">The sum of <paramref name="offset"/> and <paramref name="count"/> is greater than the buffer length.</exception>
- /// <exception cref="ArgumentNullException"><paramref name="buffer"/> is <see langword="null"/>.</exception>
- /// <exception cref="ArgumentOutOfRangeException"><paramref name="offset"/> or <paramref name="count"/> is negative.</exception>
- /// <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>
- public override async Task WriteAsync(byte[] buffer, int offset, int count, CancellationToken cancellationToken)
- {
- ThrowHelper.ThrowIfNull(buffer);
- #if NET8_0_OR_GREATER
- ArgumentOutOfRangeException.ThrowIfNegative(offset);
- ArgumentOutOfRangeException.ThrowIfNegative(count);
- #else
- if (offset < 0)
- {
- throw new ArgumentOutOfRangeException(nameof(offset));
- }
- if (count < 0)
- {
- throw new ArgumentOutOfRangeException(nameof(count));
- }
- #endif
- if ((buffer.Length - offset) < count)
- {
- throw new ArgumentException("Invalid array range.");
- }
- CheckSessionIsOpen();
- // Setup this object for writing.
- SetupWrite();
- // Write data to the file stream.
- while (count > 0)
- {
- // Determine how many bytes we can write to the buffer.
- var tempLen = _writeBufferSize - _bufferPosition;
- if (tempLen <= 0)
- {
- // flush write buffer, and mark it empty
- await FlushWriteBufferAsync(cancellationToken).ConfigureAwait(false);
- // we can now write or buffer the full buffer size
- tempLen = _writeBufferSize;
- }
- // limit the number of bytes to write to the actual number of bytes requested
- if (tempLen > count)
- {
- tempLen = count;
- }
- // Can we short-cut the internal buffer?
- if (_bufferPosition == 0 && tempLen == _writeBufferSize)
- {
- await _session.RequestWriteAsync(_handle, (ulong)_position, buffer, offset, tempLen, cancellationToken).ConfigureAwait(false);
- }
- else
- {
- // No: copy the data to the write buffer first.
- Buffer.BlockCopy(buffer, offset, GetOrCreateWriteBuffer(), _bufferPosition, tempLen);
- _bufferPosition += tempLen;
- }
- // Advance the buffer and stream positions.
- _position += tempLen;
- offset += tempLen;
- count -= tempLen;
- }
- // If the buffer is full, then do a speculative flush now,
- // rather than waiting for the next call to this method.
- if (_bufferPosition >= _writeBufferSize)
- {
- await _session.RequestWriteAsync(_handle, (ulong)(_position - _bufferPosition), GetOrCreateWriteBuffer(), 0, _bufferPosition, cancellationToken).ConfigureAwait(false);
- _bufferPosition = 0;
- }
- }
- /// <summary>
- /// Writes a byte to the current position in the stream and advances the position within the stream by one byte.
- /// </summary>
- /// <param name="value">The byte to write to the stream.</param>
- /// <exception cref="IOException">An I/O error occurs. </exception>
- /// <exception cref="NotSupportedException">The stream does not support writing, or the stream is already closed. </exception>
- /// <exception cref="ObjectDisposedException">Methods were called after the stream was closed. </exception>
- public override void WriteByte(byte value)
- {
- // Lock down the file stream while we do this.
- lock (_lock)
- {
- CheckSessionIsOpen();
- // Setup the object for writing.
- SetupWrite();
- var writeBuffer = GetOrCreateWriteBuffer();
- // Flush the current buffer if it is full.
- if (_bufferPosition >= _writeBufferSize)
- {
- using (var wait = new AutoResetEvent(initialState: false))
- {
- _session.RequestWrite(_handle, (ulong)(_position - _bufferPosition), writeBuffer, 0, _bufferPosition, wait);
- }
- _bufferPosition = 0;
- }
- // Write the byte into the buffer and advance the posn.
- writeBuffer[_bufferPosition++] = value;
- ++_position;
- }
- }
- /// <summary>
- /// Releases the unmanaged resources used by the <see cref="Stream"/> and optionally releases the managed resources.
- /// </summary>
- /// <param name="disposing"><see langword="true"/> to release both managed and unmanaged resources; <see langword="false"/> to release only unmanaged resources.</param>
- protected override void Dispose(bool disposing)
- {
- base.Dispose(disposing);
- if (_session != null)
- {
- if (disposing)
- {
- lock (_lock)
- {
- if (_session != null)
- {
- _canRead = false;
- _canSeek = false;
- _canWrite = false;
- if (_handle != null)
- {
- if (_session.IsOpen)
- {
- if (_bufferOwnedByWrite)
- {
- FlushWriteBuffer();
- }
- _session.RequestClose(_handle);
- }
- _handle = null;
- }
- _session = null;
- }
- }
- }
- }
- }
- private byte[] GetOrCreateReadBuffer()
- {
- _readBuffer ??= new byte[_readBufferSize];
- return _readBuffer;
- }
- private byte[] GetOrCreateWriteBuffer()
- {
- _writeBuffer ??= new byte[_writeBufferSize];
- return _writeBuffer;
- }
- /// <summary>
- /// Flushes the read data from the buffer.
- /// </summary>
- private void FlushReadBuffer()
- {
- _bufferPosition = 0;
- _bufferLen = 0;
- }
- /// <summary>
- /// Flush any buffered write data to the file.
- /// </summary>
- private void FlushWriteBuffer()
- {
- if (_bufferPosition > 0)
- {
- using (var wait = new AutoResetEvent(initialState: false))
- {
- _session.RequestWrite(_handle, (ulong)(_position - _bufferPosition), _writeBuffer, 0, _bufferPosition, wait);
- }
- _bufferPosition = 0;
- }
- }
- private async Task FlushWriteBufferAsync(CancellationToken cancellationToken)
- {
- if (_bufferPosition > 0)
- {
- await _session.RequestWriteAsync(_handle, (ulong)(_position - _bufferPosition), _writeBuffer, 0, _bufferPosition, cancellationToken).ConfigureAwait(false);
- _bufferPosition = 0;
- }
- }
- /// <summary>
- /// Setups the read.
- /// </summary>
- private void SetupRead()
- {
- if (!CanRead)
- {
- throw new NotSupportedException("Read not supported.");
- }
- if (_bufferOwnedByWrite)
- {
- FlushWriteBuffer();
- _bufferOwnedByWrite = false;
- }
- }
- /// <summary>
- /// Setups the write.
- /// </summary>
- private void SetupWrite()
- {
- if (!CanWrite)
- {
- throw new NotSupportedException("Write not supported.");
- }
- if (!_bufferOwnedByWrite)
- {
- FlushReadBuffer();
- _bufferOwnedByWrite = true;
- }
- }
- private void CheckSessionIsOpen()
- {
- ThrowHelper.ThrowObjectDisposedIf(_session is null, this);
- if (!_session.IsOpen)
- {
- throw new ObjectDisposedException(GetType().FullName, "Cannot access a closed SFTP session.");
- }
- }
- }
- }
|