public IDisposable SetupUploadTransferProgress(ITransferProgress transferProgress, TimeSpan timeout) {
            var lastTime = Tools.Generic.GetCurrentUtcDateTime;
            long lastBytes = 0;

            UploadProgressChanged +=
                (sender, args) => {
                    var bytes = args.BytesReceived;
                    var now = Tools.Generic.GetCurrentUtcDateTime;

                    transferProgress.FileSizeTransfered = bytes;
                    long? speed = null;
                    if (lastBytes != 0) {
                        var timeSpan = now - lastTime;
                        var bytesChange = bytes - lastBytes;

                        if (timeSpan.TotalMilliseconds > 0)
                            speed = (long) (bytesChange/(timeSpan.TotalMilliseconds/1000.0));
                    }
                    transferProgress.Update(speed, args.ProgressPercentage);
                    lastTime = now;
                    lastBytes = bytes;
                };

            UploadFileCompleted += (sender, args) => { transferProgress.Completed = true; };

            var timer = new TimerWithElapsedCancellation(500, () => {
                if (!transferProgress.Completed
                    && Tools.Generic.LongerAgoThan(lastTime, timeout)) {
                    CancelAsync();
                    return false;
                }
                return !transferProgress.Completed;
            });
            return timer;
        }

        public bool CheckZsyncLoop(Process process, string data, ITransferProgress progress) {
            // By not resetting the loopdata/counts we set ourselves up for a situation where it's possible to detect loops that span multiple lines..
            if (!ZsyncLoop.IsMatch(data))
                return false;

            if (progress.ZsyncLoopData == null) {
                progress.ZsyncLoopData = data;
                progress.ZsyncLoopCount = 0;
                return false;
            }

            if (progress.ZsyncLoopData.Equals(data, StringComparison.OrdinalIgnoreCase)) {
                var loopCount = ++progress.ZsyncLoopCount;
                if (loopCount < 2)
                    return false;
                if (!process.SafeHasExited())
                    process.TryKill();
                return true;
            }

            progress.ZsyncLoopData = data;
            progress.ZsyncLoopCount = 0;

            return false;
        }

 public void Pull(ITransferProgress status, string remoteSub = null, string localSub = null) {
     CreateSshFolder();
     HandleRsyncResponse(_rsyncLauncher.RunAndProcess(
         status,
         JoinPathsIfNeeded(Remote, remoteSub),
         JoinPathsIfNeeded(Local, localSub),
         BuildOptions()));
 }

        protected TransferSpec(Uri uri, IAbsoluteFilePath localFile, ITransferProgress progress) {
            Contract.Requires<ArgumentNullException>(uri != null);
            Contract.Requires<ArgumentNullException>(localFile != null);

            Uri = uri;
            LocalFile = localFile;
            Progress = progress ?? new TransferProgress();
        }

 public ProcessExitResultWithOutput RunAndProcess(ITransferProgress progress, string source, string destination,
     CancellationToken token,
     RsyncOptions options = null) {
     var processInfo = BuildProcessInfo(progress, source, destination, options);
     processInfo.CancellationToken = token;
     return ProcessExitResultWithOutput.FromProcessExitResult(_processManager.LaunchAndProcess(processInfo),
         progress.Output);
 }

 public async Task<ProcessExitResultWithOutput> RunAndProcessAsync(ITransferProgress progress, string source,
     string destination,
     RsyncOptions options = null) {
     var processInfo = BuildProcessInfo(progress, source, destination, options);
     return
         ProcessExitResultWithOutput.FromProcessExitResult(
             await _processManager.LaunchAndProcessAsync(processInfo).ConfigureAwait(false), progress.Output);
 }

 public ProcessExitResultWithOutput RunAndProcess(ITransferProgress progress, Uri url, string file) {
     TryHandleOldFiles(file);
     var processInfo = BuildProcessInfo(progress, url, file);
     var r =
         ProcessExitResultWithOutput.FromProcessExitResult(_processManager.LaunchAndProcess(processInfo),
             progress.Output);
     TryRemoveOldFiles(file);
     return r;
 }

        public override void ParseOutput(Process sender, string data, ITransferProgress progress) {
            //        3.51M  43%  177.98kB/s    0:00:25
            // sent 1.39K bytes  received 1.01K bytes  1.60K bytes/sec
            // total size is 114.55K  speedup is 47.57
            if (data == null)
                return;
            progress.UpdateOutput(data);

            var matches = RSYNC_START.Matches(data);
            if (matches.Count > 0) {
                var match = matches[0];


                var speed = match.Groups[2].Value.TryDouble();
                var speedUnit = match.Groups[3].Value;
                progress.Update(GetByteSize(speed, speedUnit), match.Groups[1].Value.TryDouble());

                TimeSpan ts;
                if (TimeSpan.TryParse(match.Groups[4].Value, out ts))
                    progress.Eta = ts;

                return;
            }

            matches = RSYNC_END.Matches(data);
            if (matches.Count > 0) {
                var match = matches[0];
                var sent = match.Groups[1].Value.TryDouble();
                var sentUnit = match.Groups[2].Value;

                var received = match.Groups[3].Value.TryDouble();
                var receivedUnit = match.Groups[4].Value;

                // Rsync final message omits B(yte) indication
                progress.FileSizeTransfered = GetByteSize(sent, sentUnit + "b") +
                                              GetByteSize(received, receivedUnit + "b");
                progress.Completed = true;
                return;
            }

            progress.Eta = null;
            progress.Update(null, 0);
        }

        public override void ParseOutput(Process sender, string data, ITransferProgress progress) {
            // ###################- 97.3% 141.5 kBps 0:00 ETA  
            // used 0 local, fetched 894
            if (data == null)
                return;
            progress.UpdateOutput(data);

            if (!VerifyZsyncCompatible(data))
                progress.ZsyncIncompatible = true;

            if (CheckZsyncLoop(sender, data, progress))
                return;

            var matches = ZSYNC_START.Matches(data);
            if (matches.Count > 0) {
                var match = matches[0];

                var speed = match.Groups[2].Value.TryDouble();
                var speedUnit = match.Groups[3].Value;
                progress.Update(GetByteSize(speed, speedUnit), match.Groups[1].Value.TryDouble());

                var eta = match.Groups[4].Value;
                if (tspan.IsMatch(eta))
                    eta = "0:" + eta;
                TimeSpan ts;
                if (TimeSpan.TryParse(eta, out ts))
                    progress.Eta = ts;

                return;
            }

            matches = ZSYNC_END.Matches(data);
            if (matches.Count > 0) {
                var match = matches[0];
                progress.FileSizeTransfered = match.Groups[2].Value.TryInt();
                progress.Completed = true;
                return;
            }

            progress.Eta = null;
            progress.Update(null, 0);
        }

		/// <summary>
		/// Gets a substream of the specified body part.
		/// </summary>
		/// <remarks>
		/// <para>Gets a substream of the specified message.</para>
		/// <para>For more information about how to construct the <paramref name="section"/>,
		/// see Section 6.4.5 of RFC3501.</para>
		/// </remarks>
		/// <returns>The stream.</returns>
		/// <param name="uid">The UID of the message.</param>
		/// <param name="section">The desired section of the message.</param>
		/// <param name="cancellationToken">The cancellation token.</param>
		/// <param name="progress">The progress reporting mechanism.</param>
		/// <exception cref="System.ArgumentException">
		/// <paramref name="uid"/> is invalid.
		/// </exception>
		/// <exception cref="System.ArgumentNullException">
		/// <paramref name="section"/> is <c>null</c>.
		/// </exception>
		/// <exception cref="System.ObjectDisposedException">
		/// The <see cref="ImapClient"/> has been disposed.
		/// </exception>
		/// <exception cref="ServiceNotConnectedException">
		/// The <see cref="ImapClient"/> is not connected.
		/// </exception>
		/// <exception cref="ServiceNotAuthenticatedException">
		/// The <see cref="ImapClient"/> is not authenticated.
		/// </exception>
		/// <exception cref="FolderNotOpenException">
		/// The <see cref="ImapFolder"/> is not currently open.
		/// </exception>
		/// <exception cref="MessageNotFoundException">
		/// The IMAP server did not return the requested message stream.
		/// </exception>
		/// <exception cref="System.OperationCanceledException">
		/// The operation was canceled via the cancellation token.
		/// </exception>
		/// <exception cref="System.IO.IOException">
		/// An I/O error occurred.
		/// </exception>
		/// <exception cref="ImapProtocolException">
		/// The server's response contained unexpected tokens.
		/// </exception>
		/// <exception cref="ImapCommandException">
		/// The server replied with a NO or BAD response.
		/// </exception>
		public override Stream GetStream (UniqueId uid, string section, CancellationToken cancellationToken = default (CancellationToken), ITransferProgress progress = null)
		{
			if (uid.Id == 0)
				throw new ArgumentException ("The uid is invalid.", "uid");

			if (section == null)
				throw new ArgumentNullException ("section");

			CheckState (true, false);

			var command = string.Format ("UID FETCH {0} (BODY.PEEK[{1}])\r\n", uid.Id, section);
			var ic = new ImapCommand (Engine, cancellationToken, this, command);
			var ctx = new FetchStreamContext (progress);
			Stream stream;

			ic.RegisterUntaggedHandler ("FETCH", FetchStream);
			ic.UserData = ctx;

			Engine.QueueCommand (ic);

			try {
				Engine.Wait (ic);

				ProcessResponseCodes (ic, null);

				if (ic.Response != ImapCommandResponse.Ok)
					throw ImapCommandException.Create ("FETCH", ic);

				if (!ctx.Sections.TryGetValue (section, out stream))
					throw new MessageNotFoundException ("The IMAP server did not return the requested stream.");

				ctx.Sections.Remove (section);
			} finally {
				ctx.Dispose ();
			}

			return stream;
		}

		/// <summary>
		/// Gets the specified body part.
		/// </summary>
		/// <remarks>
		/// Gets the specified body part.
		/// </remarks>
		/// <returns>The body part.</returns>
		/// <param name="index">The index of the message.</param>
		/// <param name="partSpecifier">The body part specifier.</param>
		/// <param name="headersOnly"><c>true</c> if only the headers should be downloaded; otherwise, <c>false</c>></param>
		/// <param name="cancellationToken">The cancellation token.</param>
		/// <param name="progress">The progress reporting mechanism.</param>
		/// <exception cref="System.ArgumentNullException">
		/// <paramref name="partSpecifier"/> is <c>null</c>.
		/// </exception>
		/// <exception cref="System.ArgumentOutOfRangeException">
		/// <paramref name="index"/> is out of range.
		/// </exception>
		/// <exception cref="System.ObjectDisposedException">
		/// The <see cref="ImapClient"/> has been disposed.
		/// </exception>
		/// <exception cref="ServiceNotConnectedException">
		/// The <see cref="ImapClient"/> is not connected.
		/// </exception>
		/// <exception cref="ServiceNotAuthenticatedException">
		/// The <see cref="ImapClient"/> is not authenticated.
		/// </exception>
		/// <exception cref="FolderNotOpenException">
		/// The <see cref="ImapFolder"/> is not currently open.
		/// </exception>
		/// <exception cref="MessageNotFoundException">
		/// The IMAP server did not return the requested message.
		/// </exception>
		/// <exception cref="System.OperationCanceledException">
		/// The operation was canceled via the cancellation token.
		/// </exception>
		/// <exception cref="System.IO.IOException">
		/// An I/O error occurred.
		/// </exception>
		/// <exception cref="ImapProtocolException">
		/// The server's response contained unexpected tokens.
		/// </exception>
		/// <exception cref="ImapCommandException">
		/// The server replied with a NO or BAD response.
		/// </exception>
		public MimeEntity GetBodyPart (int index, string partSpecifier, bool headersOnly, CancellationToken cancellationToken = default (CancellationToken), ITransferProgress progress = null)
		{
			if (index < 0 || index >= Count)
				throw new ArgumentOutOfRangeException ("index");

			if (partSpecifier == null)
				throw new ArgumentNullException ("partSpecifier");

			CheckState (true, false);

			string[] tags;

			var command = string.Format ("FETCH {0} ({1})\r\n", index + 1, GetBodyPartQuery (partSpecifier, headersOnly, out tags));
			var ic = new ImapCommand (Engine, cancellationToken, this, command);
			var ctx = new FetchStreamContext (progress);
			ChainedStream chained;
			bool dispose = false;
			Stream stream;

			ic.RegisterUntaggedHandler ("FETCH", FetchStream);
			ic.UserData = ctx;

			Engine.QueueCommand (ic);

			try {
				Engine.Wait (ic);

				ProcessResponseCodes (ic, null);

				if (ic.Response != ImapCommandResponse.Ok)
					throw ImapCommandException.Create ("FETCH", ic);

				chained = new ChainedStream ();

				foreach (var tag in tags) {
					if (!ctx.Sections.TryGetValue (tag, out stream))
						throw new MessageNotFoundException ("The IMAP server did not return the requested body part.");

					if (!(stream is MemoryStream || stream is MemoryBlockStream))
						dispose = true;

					chained.Add (stream);
				}

				foreach (var tag in tags)
					ctx.Sections.Remove (tag);
			} finally {
				ctx.Dispose ();
			}

			var entity = ParseEntity (chained, dispose, cancellationToken);

			if (partSpecifier.Length == 0) {
				for (int i = entity.Headers.Count; i > 0; i--) {
					var header = entity.Headers[i - 1];
//.........这里部分代码省略.........

			public FetchStreamContext (ITransferProgress progress)
			{
				Progress = progress;
			}

		ImapCommand QueueMultiAppend (FormatOptions options, IList<MimeMessage> messages, IList<MessageFlags> flags, IList<DateTimeOffset> dates, CancellationToken cancellationToken, ITransferProgress progress)
		{
			var args = new List<object> ();
			string format = "APPEND %F";

			args.Add (this);

			for (int i = 0; i < messages.Count; i++) {
				if ((flags[i] & SettableFlags) != 0)
					format += " " + ImapUtils.FormatFlagsList (flags[i], 0);

				if (dates != null)
					format += " \"" + ImapUtils.FormatInternalDate (dates[i]) + "\"";

				format += " %L";

				args.Add (messages[i]);
			}

			format += "\r\n";

			var ic = new ImapCommand (Engine, cancellationToken, null, options, format, args.ToArray ());
			ic.Progress = progress;

			Engine.QueueCommand (ic);

			return ic;
		}

		ImapCommand QueueAppend (FormatOptions options, MimeMessage message, MessageFlags flags, DateTimeOffset? date, CancellationToken cancellationToken, ITransferProgress progress)
		{
			string format = "APPEND %F";

			if ((flags & SettableFlags) != 0)
				format += " " + ImapUtils.FormatFlagsList (flags, 0);

			if (date.HasValue)
				format += " \"" + ImapUtils.FormatInternalDate (date.Value) + "\"";

			format += " %L\r\n";

			var ic = new ImapCommand (Engine, cancellationToken, null, options, format, this, message);
			ic.Progress = progress;

			Engine.QueueCommand (ic);

			return ic;
		}

		/// <summary>
		/// Appends the specified message to the folder.
		/// </summary>
		/// <remarks>
		/// Appends the specified message to the folder and returns the UniqueId assigned to the message.
		/// </remarks>
		/// <returns>The UID of the appended message, if available; otherwise, <c>null</c>.</returns>
		/// <param name="options">The formatting options.</param>
		/// <param name="message">The message.</param>
		/// <param name="flags">The message flags.</param>
		/// <param name="date">The received date of the message.</param>
		/// <param name="cancellationToken">The cancellation token.</param>
		/// <param name="progress">The progress reporting mechanism.</param>
		/// <exception cref="System.ArgumentNullException">
		/// <para><paramref name="options"/> is <c>null</c>.</para>
		/// <para>-or-</para>
		/// <para><paramref name="message"/> is <c>null</c>.</para>
		/// </exception>
		/// <exception cref="System.ObjectDisposedException">
		/// The <see cref="ImapClient"/> has been disposed.
		/// </exception>
		/// <exception cref="ServiceNotConnectedException">
		/// The <see cref="ImapClient"/> is not connected.
		/// </exception>
		/// <exception cref="ServiceNotAuthenticatedException">
		/// The <see cref="ImapClient"/> is not authenticated.
		/// </exception>
		/// <exception cref="System.InvalidOperationException">
		/// Internationalized formatting was requested but has not been enabled.
		/// </exception>
		/// <exception cref="FolderNotFoundException">
		/// The <see cref="ImapFolder"/> does not exist.
		/// </exception>
		/// <exception cref="System.OperationCanceledException">
		/// The operation was canceled via the cancellation token.
		/// </exception>
		/// <exception cref="System.NotSupportedException">
		/// Internationalized formatting was requested but is not supported by the server.
		/// </exception>
		/// <exception cref="System.IO.IOException">
		/// An I/O error occurred.
		/// </exception>
		/// <exception cref="ImapProtocolException">
		/// The server's response contained unexpected tokens.
		/// </exception>
		/// <exception cref="ImapCommandException">
		/// The server replied with a NO or BAD response.
		/// </exception>
		public override UniqueId? Append (FormatOptions options, MimeMessage message, MessageFlags flags, DateTimeOffset date, CancellationToken cancellationToken = default (CancellationToken), ITransferProgress progress = null)
		{
			if (options == null)
				throw new ArgumentNullException ("options");

			if (message == null)
				throw new ArgumentNullException ("message");

			CheckState (false, false);

			if (options.International && (Engine.Capabilities & ImapCapabilities.UTF8Accept) == 0)
				throw new NotSupportedException ("The IMAP server does not support the UTF8 extension.");

			var format = options.Clone ();
			format.NewLineFormat = NewLineFormat.Dos;

			if ((Engine.Capabilities & ImapCapabilities.UTF8Only) == ImapCapabilities.UTF8Only)
				format.International = true;

			if (format.International && !Engine.UTF8Enabled)
				throw new InvalidOperationException ("The UTF8 extension has not been enabled.");

			var ic = QueueAppend (format, message, flags, date, cancellationToken, progress);

			Engine.Wait (ic);

			ProcessResponseCodes (ic, this);

			if (ic.Response != ImapCommandResponse.Ok)
				throw ImapCommandException.Create ("APPEND", ic);

			var append = ic.RespCodes.OfType<AppendUidResponseCode> ().FirstOrDefault ();

			if (append != null)
				return append.UidSet[0];

			return null;
		}

		/// <summary>
		/// Asynchronously get a substream of the specified body part.
		/// </summary>
		/// <remarks>
		/// <para>Asynchronously gets a substream of the specified message. If the starting
		/// offset is beyond the end of the specified section of the message, an empty stream
		/// is returned. If the number of bytes desired extends beyond the end of the section,
		/// a truncated stream will be returned.</para>
		/// <para>For more information about how to construct the <paramref name="section"/>,
		/// see Section 6.4.5 of RFC3501.</para>
		/// </remarks>
		/// <returns>The stream.</returns>
		/// <param name="index">The index of the message.</param>
		/// <param name="section">The desired section of the message.</param>
		/// <param name="offset">The starting offset of the first desired byte.</param>
		/// <param name="count">The number of bytes desired.</param>
		/// <param name="cancellationToken">The cancellation token.</param>
		/// <param name="progress">The progress reporting mechanism.</param>
		/// <exception cref="System.ArgumentNullException">
		/// <paramref name="section"/> is <c>null</c>.
		/// </exception>
		/// <exception cref="System.ArgumentOutOfRangeException">
		/// <para><paramref name="index"/> is out of range.</para>
		/// <para>-or-</para>
		/// <para><paramref name="offset"/> is negative.</para>
		/// <para>-or-</para>
		/// <para><paramref name="count"/> is negative.</para>
		/// </exception>
		/// <exception cref="System.ObjectDisposedException">
		/// The <see cref="IMailStore"/> has been disposed.
		/// </exception>
		/// <exception cref="ServiceNotConnectedException">
		/// The <see cref="IMailStore"/> is not connected.
		/// </exception>
		/// <exception cref="ServiceNotAuthenticatedException">
		/// The <see cref="IMailStore"/> is not authenticated.
		/// </exception>
		/// <exception cref="FolderNotOpenException">
		/// The folder is not currently open.
		/// </exception>
		/// <exception cref="MessageNotFoundException">
		/// The <see cref="IMailStore"/> did not return the requested message stream.
		/// </exception>
		/// <exception cref="System.OperationCanceledException">
		/// The operation was canceled via the cancellation token.
		/// </exception>
		/// <exception cref="System.IO.IOException">
		/// An I/O error occurred.
		/// </exception>
		/// <exception cref="ProtocolException">
		/// The server's response contained unexpected tokens.
		/// </exception>
		/// <exception cref="CommandException">
		/// The command failed.
		/// </exception>
		public virtual Task<Stream> GetStreamAsync (int index, string section, int offset, int count, CancellationToken cancellationToken = default (CancellationToken), ITransferProgress progress = null)
		{
			if (index < 0 || index >= Count)
				throw new ArgumentOutOfRangeException ("index");

			if (section == null)
				throw new ArgumentNullException ("section");

			if (offset < 0)
				throw new ArgumentOutOfRangeException ("offset");

			if (count < 0)
				throw new ArgumentOutOfRangeException ("count");

			return Task.Factory.StartNew (() => {
				lock (SyncRoot) {
					return GetStream (index, section, offset, count, cancellationToken, progress);
				}
			}, cancellationToken, TaskCreationOptions.None, TaskScheduler.Default);
		}

			public SendContext (ITransferProgress progress, long? size)
			{
				this.progress = progress;
				this.size = size;
			}

		/// <summary>
		/// Gets the specified message.
		/// </summary>
		/// <remarks>
		/// Gets the specified message.
		/// </remarks>
		/// <returns>The message.</returns>
		/// <param name="index">The index of the message.</param>
		/// <param name="cancellationToken">The cancellation token.</param>
		/// <param name="progress">The progress reporting mechanism.</param>
		/// <exception cref="System.ArgumentOutOfRangeException">
		/// <paramref name="index"/> is out of range.
		/// </exception>
		/// <exception cref="System.ObjectDisposedException">
		/// The <see cref="ImapClient"/> has been disposed.
		/// </exception>
		/// <exception cref="ServiceNotConnectedException">
		/// The <see cref="ImapClient"/> is not connected.
		/// </exception>
		/// <exception cref="ServiceNotAuthenticatedException">
		/// The <see cref="ImapClient"/> is not authenticated.
		/// </exception>
		/// <exception cref="FolderNotOpenException">
		/// The <see cref="ImapFolder"/> is not currently open.
		/// </exception>
		/// <exception cref="MessageNotFoundException">
		/// The IMAP server did not return the requested message.
		/// </exception>
		/// <exception cref="System.OperationCanceledException">
		/// The operation was canceled via the cancellation token.
		/// </exception>
		/// <exception cref="System.IO.IOException">
		/// An I/O error occurred.
		/// </exception>
		/// <exception cref="ImapProtocolException">
		/// The server's response contained unexpected tokens.
		/// </exception>
		/// <exception cref="ImapCommandException">
		/// The server replied with a NO or BAD response.
		/// </exception>
		public override MimeMessage GetMessage (int index, CancellationToken cancellationToken = default (CancellationToken), ITransferProgress progress = null)
		{
			if (index < 0 || index >= Count)
				throw new ArgumentOutOfRangeException ("index");

			CheckState (true, false);

			var ic = new ImapCommand (Engine, cancellationToken, this, "FETCH %d (BODY.PEEK[])\r\n", index + 1);
			var ctx = new FetchStreamContext (progress);
			Stream stream;

			ic.RegisterUntaggedHandler ("FETCH", FetchStream);
			ic.UserData = ctx;

			Engine.QueueCommand (ic);

			try {
				Engine.Wait (ic);

				ProcessResponseCodes (ic, null);

				if (ic.Response != ImapCommandResponse.Ok)
					throw ImapCommandException.Create ("FETCH", ic);

				if (!ctx.Sections.TryGetValue (string.Empty, out stream))
					throw new MessageNotFoundException ("The IMAP server did not return the requested message.");

				ctx.Sections.Remove (string.Empty);
			} finally {
				ctx.Dispose ();
			}

			return ParseMessage (stream, cancellationToken);
		}

		void Bdat (FormatOptions options, MimeMessage message, CancellationToken cancellationToken, ITransferProgress progress)
		{
			long size;

			using (var measure = new MeasuringStream ()) {
				message.WriteTo (options, measure, cancellationToken);
				size = measure.Length;
			}

			var bytes = Encoding.UTF8.GetBytes (string.Format ("BDAT {0} LAST\r\n", size));

			Stream.Write (bytes, 0, bytes.Length, cancellationToken);

			if (progress != null) {
				var ctx = new SendContext (progress, size);

				using (var stream = new ProgressStream (Stream, ctx.Update)) {
					message.WriteTo (options, stream, cancellationToken);
					stream.Flush (cancellationToken);
				}
			} else {
				message.WriteTo (options, Stream, cancellationToken);
				Stream.Flush (cancellationToken);
			}

			var response = Stream.ReadResponse (cancellationToken);

			switch (response.StatusCode) {
			default:
				throw new SmtpCommandException (SmtpErrorCode.MessageNotAccepted, response.StatusCode, response.Response);
			case SmtpStatusCode.AuthenticationRequired:
				throw new ServiceNotAuthenticatedException (response.Response);
			case SmtpStatusCode.Ok:
				OnMessageSent (new MessageSentEventArgs (message, response.Response));
				break;
			}
		}

		void Data (FormatOptions options, MimeMessage message, CancellationToken cancellationToken, ITransferProgress progress)
		{
			var response = SendCommand ("DATA", cancellationToken);

			if (response.StatusCode != SmtpStatusCode.StartMailInput)
				throw new SmtpCommandException (SmtpErrorCode.UnexpectedStatusCode, response.StatusCode, response.Response);

			if (progress != null) {
				var ctx = new SendContext (progress, null);

				using (var stream = new ProgressStream (Stream, ctx.Update)) {
					using (var filtered = new FilteredStream (stream)) {
						filtered.Add (new SmtpDataFilter ());

						message.WriteTo (options, filtered, cancellationToken);
						filtered.Flush ();
					}
				}
			} else {
				using (var filtered = new FilteredStream (Stream)) {
					filtered.Add (new SmtpDataFilter ());

					message.WriteTo (options, filtered, cancellationToken);
					filtered.Flush ();
				}
			}

			Stream.Write (EndData, 0, EndData.Length, cancellationToken);
			Stream.Flush (cancellationToken);

			response = Stream.ReadResponse (cancellationToken);

			switch (response.StatusCode) {
			default:
				throw new SmtpCommandException (SmtpErrorCode.MessageNotAccepted, response.StatusCode, response.Response);
			case SmtpStatusCode.AuthenticationRequired:
				throw new ServiceNotAuthenticatedException (response.Response);
			case SmtpStatusCode.Ok:
				OnMessageSent (new MessageSentEventArgs (message, response.Response));
				break;
			}
		}