Parallel FTP client

Question

I was given the task to download files for a FTP server. The download should be fairly fast. 15 parallel connections can be used.

I have used FluentFTP as the underlying library taking care of the technical details of FTP.

Here is the FTP client and tests below.

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Threading.Tasks;
using FluentFTP;
using log4net;

namespace FTPSynchronizator
{
    public class ParallelFtpClient : IDisposable
    {
        private readonly ILog _logger = LogManager.GetLogger(typeof(MeteologicaFtpSyncer).FullName);
        private readonly int _maxConcurentConnections;
        private readonly Func<IFtpClient> _factory;
        private readonly List<IFtpClient> _ftpClients;

        public ParallelFtpClient(int maxConcurentConnections, Func<IFtpClient> factory)
        {
            _maxConcurentConnections = maxConcurentConnections;
            _factory = factory;
            _ftpClients = new List<IFtpClient>();
        }

        public void Connect()
        {
            var tasks = new List<Task>();
            for (int i = 0; i < _maxConcurentConnections; i++)
            {
                var ftpClient = _factory();
                _ftpClients.Add(ftpClient);
                var connectTask = ftpClient.ConnectAsync();
                tasks.Add(connectTask);
            }
            Task.WaitAll(tasks.ToArray());
        }

        public void Disconnect()
        {
            var tasks = _ftpClients
                .Select(x => x.DisconnectAsync())
                .ToArray();
            Task.WaitAll(tasks);
        }

        public List<string> ListFilesInDirectories(IEnumerable<string> directories)
        {
            var items = new ConcurrentBag<FtpListItem>();
            var clients = new ConcurrentQueue<IFtpClient>(_ftpClients);

            Parallel.ForEach(directories,
                new ParallelOptions() {MaxDegreeOfParallelism = _maxConcurentConnections},
                (path) =>
                {
                    IFtpClient client;
                    clients.TryDequeue(out client);
                    var listing = client.GetListing(path);
                    foreach (var item in listing)
                    {
                        items.Add(item);
                    }
                    clients.Enqueue(client);
                });

            var list = items
                .Where(x => x.Type == FtpFileSystemObjectType.File)
                .Select(x => x.FullName)
                .ToList();
            return list;
        }

        public List<string> DownloadFilesParallel(IEnumerable<(string,string)> ftpPathLocalPathPairs, out List<(string, string)> failedDownloads)
        {
            var clients = new ConcurrentQueue<IFtpClient>(_ftpClients);
            var downloadedFiles = new ConcurrentBag<string>();
            var failedDownloadsBag = new ConcurrentBag<(string,string)>();

            Parallel.ForEach(ftpPathLocalPathPairs,
                new ParallelOptions() {MaxDegreeOfParallelism = _maxConcurentConnections},
                (ftpPathLocalPathPair) =>
                {
                    IFtpClient client;
                    clients.TryDequeue(out client);
                    var ftpFilePath = ftpPathLocalPathPair.Item1;
                    var destinationFilePath = ftpPathLocalPathPair.Item2;
                    _logger.Debug($"Downloading {ftpFilePath} to {destinationFilePath}...");
                    try
                    {
                        var memoryStream = new MemoryStream();
                        client.Download(memoryStream, ftpFilePath);
                        memoryStream.Position = 0;
                        string destinationDirectory = Path.GetDirectoryName(destinationFilePath);
                        if (!Directory.Exists(destinationDirectory))
                        {
                            _logger.Info($"Creating new directory {destinationDirectory}");
                            Directory.CreateDirectory(destinationDirectory);
                        }
                        File.WriteAllBytes(destinationFilePath, memoryStream.ToArray());
                        downloadedFiles.Add(destinationFilePath);
                    }
                    catch (Exception e)
                    {
                        _logger.Warn($"An unhandled excetpion occured while downloading file {ftpFilePath}", e);
                        failedDownloadsBag.Add(ftpPathLocalPathPair);
                    }

                    clients.Enqueue(client);
                }
            );

            failedDownloads = failedDownloadsBag.ToList();

            return downloadedFiles.ToList();
        }

        public List<string> ListNonArchiveDirectoresWithFiles(string root)
        {
            var directoriesWithFiles = new List<string>();
            var directoiresToQuery = new Queue<string>();
            var clients = new ConcurrentQueue<IFtpClient>(_ftpClients);

            directoiresToQuery.Enqueue(root);
            var tasks = new List<Task<Result>>();
            while (directoiresToQuery.Any() || tasks.Any())
            {
                while (clients.Any() && directoiresToQuery.Any())
                {
                    var path = directoiresToQuery.Dequeue();
                    IFtpClient client;
                    clients.TryDequeue(out client);
                    var task = Task.Run(() =>
                    {
                        var childItems = client.GetListing(path);
                        clients.Enqueue(client);
                        return new Result()
                        {
                            FullName = path,
                            Children = childItems
                        };
                    });
                    tasks.Add(task);
                }

                var array = tasks.ToArray();
                var finishedTaskIndex = Task.WaitAny(array);
                var finishedTask = array[finishedTaskIndex];
                var finishedTaskResult = finishedTask.Result;
                if (finishedTaskResult.Children.Any(i => i.Type == FtpFileSystemObjectType.File))
                {
                    directoriesWithFiles.Add(finishedTaskResult.FullName);
                }
                var subdirectoriesToQuery = finishedTaskResult
                    .Children
                    .Where(x => x.Type == FtpFileSystemObjectType.Directory)
                    .Where(x => x.Name != "Archive")
                    .Select(x => x.FullName);
                foreach (var subdirectoryToQuery in subdirectoriesToQuery)
                {
                    directoiresToQuery.Enqueue(subdirectoryToQuery);
                }

                tasks = array
                    .Except(new[] {finishedTask})
                    .ToList();
            }

            return directoriesWithFiles;
        }

        public void Dispose()
        {
            foreach (var ftpClient in _ftpClients)
            {
                ftpClient.Dispose();
            }
        }
    }

    public class Result
    {
        public string FullName { get; set; }
        public FtpListItem[] Children { get; set; }
    }
}

Tests and stubs:

using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using FluentFTP;
using NUnit.Framework;
using System.Globalization;
using System.Net;
using System.Security.Authentication;
using System.Security.Cryptography.X509Certificates;
using System.Text;
using System.Threading;
using System.Threading.Tasks;


namespace Tests
{
    [TestFixture]
    public class ParallelFtpClientTests
    {
        [Test]
        public static void Connect_ConstructsCorrectNumberOfClientsAndConnectsThem()
        {
            // arrage
            int factoryInvokedCount = 0;
            List<IFtpClient> clientsCreatedByFatory = new List<IFtpClient>();
            Func<IFtpClient> factory = () =>
            {
                factoryInvokedCount++;
                var client = new StubFtpClient();
                clientsCreatedByFatory.Add(client);
                return client;
            };

            var sut = new ParallelFtpClient(10, factory);

            // act
            sut.Connect();

            // assert
            Assert.AreEqual(10, factoryInvokedCount);
            Assert.IsTrue(clientsCreatedByFatory.All(x => x.IsConnected));
        }

        [Test]
        public static void Disconnect_DisconnectsAllClients()
        {
            // arrage
            List<StubFtpClient> clientsCreatedByFatory = new List<StubFtpClient>();
            Func<IFtpClient> factory = () =>
            {
                var client = new StubFtpClient();
                clientsCreatedByFatory.Add(client);
                return client;
            };

            var sut = new ParallelFtpClient(10, factory);

            // act
            sut.Connect();
            sut.Disconnect();

            // assert
            Assert.IsTrue(clientsCreatedByFatory.All(x => x.IsDisconected));
        }

        [Test]
        public static void Dispose_DisconnectsAllClients()
        {
            // arrage
            List<StubFtpClient> clientsCreatedByFatory = new List<StubFtpClient>();
            Func<IFtpClient> factory = () =>
            {
                var client = new StubFtpClient();
                clientsCreatedByFatory.Add(client);
                return client;
            };

            var sut = new ParallelFtpClient(10, factory);

            // act
            sut.Connect();
            sut.Dispose();

            // assert
            Assert.IsTrue(clientsCreatedByFatory.All(x => x.IsDisposed));
        }

        [Test]
        public static void ListFilesInDirectories_EmptyInput_ReturnsEmptyList()
        {
            // arrange
            var directories = new string[0];
            Func<IFtpClient> factory = () => new StubFtpClient();
            var sut = new ParallelFtpClient(10, factory);
            sut.Connect();

            // act
            var result = sut.ListFilesInDirectories(directories);

            // assert
            Assert.IsNotNull(result);
            Assert.IsEmpty(result);
        }

        [Test]
        public static void ListFilesInDirectories_SingleDirectory_ReturnsFilesForDirectory()
        {
            // arrange
            var directories = new string[]{"dir1"};
            DateTime lastModifiedTime = DateTime.Now;
            var map = new Dictionary<string, FtpListItem[]>()
            {
                {
                    "dir1",
                    new []
                    { 
                        new FtpListItem("dummy", "file1", 0, false, ref lastModifiedTime){FullName = "/root/file1"},
                        new FtpListItem("dummy", "file2", 0, false, ref lastModifiedTime){FullName = "/root/file2"},
                    }
                }
            };
            Func<IFtpClient> factory = () => new StubFtpClient(map);
            var sut = new ParallelFtpClient(10, factory);
            sut.Connect();

            // act
            var result = sut.ListFilesInDirectories(directories);

            // assert
            Assert.AreEqual(2, result.Count);
            CollectionAssert.Contains(result, "/root/file1");
            CollectionAssert.Contains(result, "/root/file2");
        }

        [Test]
        public static void ListFilesInDirectories_MultipleDirectories_ReturnsFilesForAllDirectories()
        {
            // arrange
            var directories = new string[]{"dir1", "dir2"};
            DateTime lastModifiedTime = DateTime.Now;
            var map = new Dictionary<string, FtpListItem[]>()
            {
                {
                    "dir1",
                    new []
                    { 
                        new FtpListItem("dummy", "file1", 0, false, ref lastModifiedTime){FullName = "/root/file1"},
                        new FtpListItem("dummy", "file2", 0, false, ref lastModifiedTime){FullName = "/root/file2"},
                    }
                },
                {
                    "dir2",
                    new []
                    { 
                        new FtpListItem("dummy", "file3", 0, false, ref lastModifiedTime){FullName = "/root/file3"},
                        new FtpListItem("dummy", "file4", 0, false, ref lastModifiedTime){FullName = "/root/file4"},
                    }
                },
            };
            Func<IFtpClient> factory = () => new StubFtpClient(map);
            var sut = new ParallelFtpClient(10, factory);
            sut.Connect();

            // act
            var result = sut.ListFilesInDirectories(directories);

            // assert
            Assert.AreEqual(4, result.Count);
            CollectionAssert.Contains(result, "/root/file1");
            CollectionAssert.Contains(result, "/root/file2");
            CollectionAssert.Contains(result, "/root/file3");
            CollectionAssert.Contains(result, "/root/file4");
        }

        [Test]
        public static void ListFilesInDirectories_FilesCanNotBeListed_ExceptionIsThrown()
        {
            // arrange
            var directories = new string[]{"this_dir_can_not_be_found_by_ftp_server"};
            var map = new Dictionary<string, FtpListItem[]>(); // passing empty map to stub so no directorise can be listed
            Func<IFtpClient> factory = () => new StubFtpClient(map);
            var sut = new ParallelFtpClient(10, factory);
            sut.Connect();

            // act
            Assert.Throws(typeof(AggregateException), () => sut.ListFilesInDirectories(directories));
        }

        [Test]
        public static void DownloadFilesParallel_EmptyInputList_DoesNothing()
        {
            // arrange
            var files = new (string, string)[]{};
            List<(string, string)> failedDownloads;

            Func<IFtpClient> factory = () => new StubFtpClient();
            var sut = new ParallelFtpClient(10, factory);
            sut.Connect();

            // act
            var downloadedFiles = sut.DownloadFilesParallel(files, out failedDownloads);

            // assert
            Assert.IsNotNull(downloadedFiles);
            Assert.IsEmpty(downloadedFiles);
        }

        [Test]
        public static void DownloadFilesParallel_SingleFile_DownloadsFile()
        {
            // arrange
            Directory.CreateDirectory("temp");
            var files = new []{("/ftp/file1", "temp/file1")};
            List<(string, string)> failedDownloads;

            Func<IFtpClient> factory = () => new StubFtpClient();
            var sut = new ParallelFtpClient(10, factory);
            sut.Connect();

            // act
            var downloadedFiles = sut.DownloadFilesParallel(files, out failedDownloads);

            // assert
            FileAssert.Exists("temp/file1");
            Assert.IsTrue(downloadedFiles.Single() == @"temp/file1");
        }

        [Test]
        public static void DownloadFilesParallel_MultipleFiles_DownloadsAllFiles()
        {
            // arrange
            Directory.CreateDirectory("temp");
            var files = new []{("/ftp/file1", "temp/file1"), ("/ftp/file2", "temp/file2")};
            List<(string, string)> failedDownloads;

            Func<IFtpClient> factory = () => new StubFtpClient();
            var sut = new ParallelFtpClient(10, factory);
            sut.Connect();

            // act
            var downloadedFiles = sut.DownloadFilesParallel(files, out failedDownloads);

            // assert
            FileAssert.Exists("temp/file1");
            FileAssert.Exists("temp/file2");
            Assert.AreEqual(2, downloadedFiles.Count);
            Assert.IsTrue(downloadedFiles.Any(x => x == @"temp/file1"));
            Assert.IsTrue(downloadedFiles.Any(x => x == @"temp/file2"));
        }

        [Test]
        public static void DownloadFilesParallel_ExceptionDuringDownload_LocalFileShouldNotBeCreatedAndFailedDownloadsAreReturned()
        {
            // arrange
            Directory.CreateDirectory("temp");
            var files = new []{("/ftp/file1", "temp/file1")};
            List<(string, string)> failedDownloads;

            Func<IFtpClient> factory = () => new StubFtpClientThrowsAfterCreatingFile();
            var sut = new ParallelFtpClient(10, factory);
            sut.Connect();

            // act
            var downloadedFiles = sut.DownloadFilesParallel(files, out failedDownloads);

            // assert
            Assert.IsFalse(File.Exists("temp/file1"));
            Assert.AreEqual(0, downloadedFiles.Count);
            Assert.AreEqual(1, failedDownloads.Count);
            Assert.AreEqual(("/ftp/file1", "temp/file1"), failedDownloads.First());
        }

        [Test]
        public static void DownloadFilesParallel_DownloadingToNonExistingDirectory_NewDirectoryIsCreatedAndFileIsDownloaded()
        {
            // arrange
            Directory.CreateDirectory("temp");
            var files = new []{("/ftp/newdir/file1", "temp/newdir/file1")};
            List<(string, string)> failedDownloads;

            Func<IFtpClient> factory = () => new StubFtpClient();
            var sut = new ParallelFtpClient(10, factory);
            sut.Connect();

            // act
            var downloadedFiles = sut.DownloadFilesParallel(files, out failedDownloads);

            // assert
            FileAssert.Exists("temp/newdir/file1");
            Assert.AreEqual(1, downloadedFiles.Count);
        }

        [Test]
        public static void ListNonArchiveDirectoresWithFiles_ThereAreNoDirectoreis_ReturnsEmptyList()
        {
            // arange
            var map = new Dictionary<string, FtpListItem[]>()
            {
                {"/root", new FtpListItem[]{}},
            };
            Func<IFtpClient> factory = () => new StubFtpClient(map);
            var sut = new ParallelFtpClient(10, factory);
            sut.Connect();

            // act
            var result = sut.ListNonArchiveDirectoresWithFiles("/root");

            // assert
            Assert.IsEmpty(result);
        }

        [Test]
        public static void ListNonArchiveDirectoresWithFiles_RootDirectoryContainsFiles_ResultContainsRoot()
        {
            // arange
            DateTime lastModifiedTime = DateTime.Now;
            var map = new Dictionary<string, FtpListItem[]>()
            {
                {
                    "/root",
                    new []
                    { 
                        new FtpListItem("dummy", "file", 0, false, ref lastModifiedTime){FullName = "/root/file"}
                    }
                },
            };
            Func<IFtpClient> factory = () => new StubFtpClient(map);
            var sut = new ParallelFtpClient(10, factory);
            sut.Connect();

            // act
            var result = sut.ListNonArchiveDirectoresWithFiles("/root");

            // assert
            Assert.AreEqual(1, result.Count);
            Assert.IsTrue(result.Any(x => x == "/root"));
        }

        [Test]
        public static void ListNonArchiveDirectoresWithFiles_RootDirectoryContainsSubdirectoriesWithFiles_ResultContainsAllDirsWithFiles()
        {
            // arange
            DateTime lastModifiedTime = DateTime.Now;
            var map = new Dictionary<string, FtpListItem[]>()
            {
                {
                    "/root",
                    new []
                    { 
                        new FtpListItem("dummy", "file", 0, false, ref lastModifiedTime){FullName = "/root/file"},
                        new FtpListItem("", "subdir", 0, true, ref lastModifiedTime){FullName = "/root/subdir"},
                    }
                },
                {
                    "/root/subdir",
                    new []
                    { 
                        new FtpListItem("dummy", "file2", 0, false, ref lastModifiedTime){FullName = "/root/subdir/file2"},
                    }
                }
            };
            Func<IFtpClient> factory = () => new StubFtpClient(map);
            var sut = new ParallelFtpClient(10, factory);
            sut.Connect();

            // act
            var result = sut.ListNonArchiveDirectoresWithFiles("/root");

            // assert
            Assert.AreEqual(2, result.Count);
            Assert.IsTrue(result.Any(x => x == "/root"));
            Assert.IsTrue(result.Any(x => x == "/root/subdir"));
        }

        [Test]
        public static void ListNonArchiveDirectoresWithFiles_ArchiveDirectoryIsPresent_ArchiveDirectoryIsNotQueried()
        {
            // arange
            DateTime lastModifiedTime = DateTime.Now;
            var map = new Dictionary<string, FtpListItem[]>()
            {
                {
                    "/root",
                    new []
                    { 
                        new FtpListItem("dummy", "file", 0, false, ref lastModifiedTime){FullName = "/root/file"},
                        new FtpListItem("", "Archive", 0, true, ref lastModifiedTime){FullName = "/root/Archive"},
                    }
                },
                {
                    "/root/Archive",
                    new []
                    { 
                        new FtpListItem("dummy", "file2", 0, false, ref lastModifiedTime){FullName = "/root/subdir/file2"},
                        null
                    }
                }
            };
            Func<IFtpClient> factory = () => new StubFtpClient(map);
            var sut = new ParallelFtpClient(10, factory);
            sut.Connect();

            // act
            var result = sut.ListNonArchiveDirectoresWithFiles("/root");

            // assert
            Assert.AreEqual(1, result.Count);
            Assert.IsTrue(result.Any(x => x == "/root"));
        }

        [TearDown]
        public static void RemoveTempFolder()
        {
            if (Directory.Exists("temp"))
            {
                Directory.Delete("temp", true);
            }
        }
    }

    public class StubFtpClient : IFtpClient
    {
        private Dictionary<string, FtpListItem[]> _dirFilesMap;
        private bool _connected = false;
        private bool _wasDisconected = false;
        private bool _isDisposed = false;

        public StubFtpClient()
        {
            _dirFilesMap = new Dictionary<string, FtpListItem[]>();
        }

        public StubFtpClient(Dictionary<string, FtpListItem[]> dirFilesMap)
        {
            _dirFilesMap = dirFilesMap;
        }

        public void Dispose()
        {
            _isDisposed = true;
        }

        public FtpReply Execute(string command)
        {
            throw new NotImplementedException();
        }

        public FtpReply GetReply()
        {
            throw new NotImplementedException();
        }

        public virtual void Connect()
        {
            _connected = true;
        }

        public void Disconnect()
        {
            _wasDisconected = true;
        }

        public bool HasFeature(FtpCapability cap)
        {
            throw new NotImplementedException();
        }

        public void DisableUTF8()
        {
            throw new NotImplementedException();
        }

        public Task<FtpReply> ExecuteAsync(string command)
        {
            throw new NotImplementedException();
        }

        public Task<FtpReply> GetReplyAsync()
        {
            throw new NotImplementedException();
        }

        public Task ConnectAsync()
        {
            Connect();
            return Task.CompletedTask;
        }

        public Task DisconnectAsync()
        {
            _wasDisconected = true;
            return Task.CompletedTask;
        }

        public void DeleteFile(string path)
        {
            throw new NotImplementedException();
        }

        public void DeleteDirectory(string path)
        {
            throw new NotImplementedException();
        }

        public void DeleteDirectory(string path, FtpListOption options)
        {
            throw new NotImplementedException();
        }

        public bool DirectoryExists(string path)
        {
            throw new NotImplementedException();
        }

        public bool FileExists(string path)
        {
            throw new NotImplementedException();
        }

        public void CreateDirectory(string path)
        {
            throw new NotImplementedException();
        }

        public void CreateDirectory(string path, bool force)
        {
            throw new NotImplementedException();
        }

        public void Rename(string path, string dest)
        {
            throw new NotImplementedException();
        }

        public bool MoveFile(string path, string dest, FtpExists existsMode = FtpExists.Overwrite)
        {
            throw new NotImplementedException();
        }

        public bool MoveDirectory(string path, string dest, FtpExists existsMode = FtpExists.Overwrite)
        {
            throw new NotImplementedException();
        }

        public void SetFilePermissions(string path, int permissions)
        {
            throw new NotImplementedException();
        }

        public void Chmod(string path, int permissions)
        {
            throw new NotImplementedException();
        }

        public void SetFilePermissions(string path, FtpPermission owner, FtpPermission @group, FtpPermission other)
        {
            throw new NotImplementedException();
        }

        public void Chmod(string path, FtpPermission owner, FtpPermission @group, FtpPermission other)
        {
            throw new NotImplementedException();
        }

        public FtpListItem GetFilePermissions(string path)
        {
            throw new NotImplementedException();
        }

        public int GetChmod(string path)
        {
            throw new NotImplementedException();
        }

        public FtpListItem DereferenceLink(FtpListItem item)
        {
            throw new NotImplementedException();
        }

        public FtpListItem DereferenceLink(FtpListItem item, int recMax)
        {
            throw new NotImplementedException();
        }

        public void SetWorkingDirectory(string path)
        {
            throw new NotImplementedException();
        }

        public string GetWorkingDirectory()
        {
            throw new NotImplementedException();
        }

        public long GetFileSize(string path)
        {
            throw new NotImplementedException();
        }

        public DateTime GetModifiedTime(string path, FtpDate type = FtpDate.Original)
        {
            throw new NotImplementedException();
        }

        public void SetModifiedTime(string path, DateTime date, FtpDate type = FtpDate.Original)
        {
            throw new NotImplementedException();
        }

        public Task DeleteFileAsync(string path)
        {
            throw new NotImplementedException();
        }

        public Task DeleteDirectoryAsync(string path)
        {
            throw new NotImplementedException();
        }

        public Task DeleteDirectoryAsync(string path, FtpListOption options)
        {
            throw new NotImplementedException();
        }

        public Task<bool> DirectoryExistsAsync(string path)
        {
            throw new NotImplementedException();
        }

        public Task<bool> FileExistsAsync(string path)
        {
            throw new NotImplementedException();
        }

        public Task CreateDirectoryAsync(string path, bool force)
        {
            throw new NotImplementedException();
        }

        public Task CreateDirectoryAsync(string path)
        {
            throw new NotImplementedException();
        }

        public Task RenameAsync(string path, string dest)
        {
            throw new NotImplementedException();
        }

        public Task<bool> MoveFileAsync(string path, string dest, FtpExists existsMode = FtpExists.Overwrite)
        {
            throw new NotImplementedException();
        }

        public Task<bool> MoveDirectoryAsync(string path, string dest, FtpExists existsMode = FtpExists.Overwrite)
        {
            throw new NotImplementedException();
        }

        public Task SetFilePermissionsAsync(string path, int permissions)
        {
            throw new NotImplementedException();
        }

        public Task ChmodAsync(string path, int permissions)
        {
            throw new NotImplementedException();
        }

        public Task SetFilePermissionsAsync(string path, FtpPermission owner, FtpPermission @group, FtpPermission other)
        {
            throw new NotImplementedException();
        }

        public Task ChmodAsync(string path, FtpPermission owner, FtpPermission @group, FtpPermission other)
        {
            throw new NotImplementedException();
        }

        public Task<FtpListItem> GetFilePermissionsAsync(string path)
        {
            throw new NotImplementedException();
        }

        public Task<int> GetChmodAsync(string path)
        {
            throw new NotImplementedException();
        }

        public Task<FtpListItem> DereferenceLinkAsync(FtpListItem item, int recMax)
        {
            throw new NotImplementedException();
        }

        public Task<FtpListItem> DereferenceLinkAsync(FtpListItem item)
        {
            throw new NotImplementedException();
        }

        public Task SetWorkingDirectoryAsync(string path)
        {
            throw new NotImplementedException();
        }

        public Task<string> GetWorkingDirectoryAsync()
        {
            throw new NotImplementedException();
        }

        public Task<long> GetFileSizeAsync(string path)
        {
            throw new NotImplementedException();
        }

        public Task<DateTime> GetModifiedTimeAsync(string path, FtpDate type = FtpDate.Original)
        {
            throw new NotImplementedException();
        }

        public Task SetModifiedTimeAsync(string path, DateTime date, FtpDate type = FtpDate.Original)
        {
            throw new NotImplementedException();
        }

        public FtpListItem GetObjectInfo(string path, bool dateModified = false)
        {
            throw new NotImplementedException();
        }

        public FtpListItem[] GetListing()
        {
            throw new NotImplementedException();
        }

        public FtpListItem[] GetListing(string path)
        {
            if (!_connected)
            {
                throw new InvalidOperationException("you didn't connect yet!");
            }

            return _dirFilesMap[path];
        }

        public FtpListItem[] GetListing(string path, FtpListOption options)
        {
            throw new NotImplementedException();
        }

        public string[] GetNameListing()
        {
            throw new NotImplementedException();
        }

        public string[] GetNameListing(string path)
        {
            throw new NotImplementedException();
        }

        public Task<FtpListItem> GetObjectInfoAsync(string path, bool dateModified = false)
        {
            throw new NotImplementedException();
        }

        public Task<FtpListItem[]> GetListingAsync(string path, FtpListOption options)
        {
            throw new NotImplementedException();
        }

        public Task<FtpListItem[]> GetListingAsync(string path)
        {
            throw new NotImplementedException();
        }

        public Task<FtpListItem[]> GetListingAsync()
        {
            throw new NotImplementedException();
        }

        public Task<string[]> GetNameListingAsync(string path)
        {
            throw new NotImplementedException();
        }

        public Task<string[]> GetNameListingAsync()
        {
            throw new NotImplementedException();
        }

        public Stream OpenRead(string path)
        {
            throw new NotImplementedException();
        }

        public Stream OpenRead(string path, FtpDataType type)
        {
            throw new NotImplementedException();
        }

        public Stream OpenRead(string path, FtpDataType type, bool checkIfFileExists)
        {
            throw new NotImplementedException();
        }

        public Stream OpenRead(string path, FtpDataType type, long restart)
        {
            throw new NotImplementedException();
        }

        public Stream OpenRead(string path, long restart)
        {
            throw new NotImplementedException();
        }

        public Stream OpenRead(string path, long restart, bool checkIfFileExists)
        {
            throw new NotImplementedException();
        }

        public Stream OpenRead(string path, FtpDataType type, long restart, bool checkIfFileExists)
        {
            throw new NotImplementedException();
        }

        public Stream OpenWrite(string path)
        {
            throw new NotImplementedException();
        }

        public Stream OpenWrite(string path, FtpDataType type)
        {
            throw new NotImplementedException();
        }

        public Stream OpenWrite(string path, FtpDataType type, bool checkIfFileExists)
        {
            throw new NotImplementedException();
        }

        public Stream OpenAppend(string path)
        {
            throw new NotImplementedException();
        }

        public Stream OpenAppend(string path, FtpDataType type)
        {
            throw new NotImplementedException();
        }

        public Stream OpenAppend(string path, FtpDataType type, bool checkIfFileExists)
        {
            throw new NotImplementedException();
        }

        public Task<Stream> OpenReadAsync(string path, FtpDataType type, long restart, bool checkIfFileExists)
        {
            throw new NotImplementedException();
        }

        public Task<Stream> OpenReadAsync(string path, FtpDataType type, long restart)
        {
            throw new NotImplementedException();
        }

        public Task<Stream> OpenReadAsync(string path, FtpDataType type)
        {
            throw new NotImplementedException();
        }

        public Task<Stream> OpenReadAsync(string path, long restart)
        {
            throw new NotImplementedException();
        }

        public Task<Stream> OpenReadAsync(string path)
        {
            throw new NotImplementedException();
        }

        public Task<Stream> OpenWriteAsync(string path, FtpDataType type, bool checkIfFileExists)
        {
            throw new NotImplementedException();
        }

        public Task<Stream> OpenWriteAsync(string path, FtpDataType type)
        {
            throw new NotImplementedException();
        }

        public Task<Stream> OpenWriteAsync(string path)
        {
            throw new NotImplementedException();
        }

        public Task<Stream> OpenAppendAsync(string path, FtpDataType type, bool checkIfFileExists)
        {
            throw new NotImplementedException();
        }

        public Task<Stream> OpenAppendAsync(string path, FtpDataType type)
        {
            throw new NotImplementedException();
        }

        public Task<Stream> OpenAppendAsync(string path)
        {
            throw new NotImplementedException();
        }

        public int UploadFiles(IEnumerable<string> localPaths, string remoteDir, FtpExists existsMode = FtpExists.Overwrite, bool createRemoteDir = true,
            FtpVerify verifyOptions = FtpVerify.None, FtpError errorHandling = FtpError.None)
        {
            throw new NotImplementedException();
        }

        public int UploadFiles(IEnumerable<FileInfo> localFiles, string remoteDir, FtpExists existsMode = FtpExists.Overwrite, bool createRemoteDir = true,
            FtpVerify verifyOptions = FtpVerify.None, FtpError errorHandling = FtpError.None)
        {
            throw new NotImplementedException();
        }

        public int DownloadFiles(string localDir, IEnumerable<string> remotePaths, bool overwrite = true, FtpVerify verifyOptions = FtpVerify.None,
            FtpError errorHandling = FtpError.None)
        {
            throw new NotImplementedException();
        }

        public bool UploadFile(string localPath, string remotePath, FtpExists existsMode = FtpExists.Overwrite, bool createRemoteDir = false,
            FtpVerify verifyOptions = FtpVerify.None, IProgress<double> progress = null)
        {
            throw new NotImplementedException();
        }

        public bool Upload(Stream fileStream, string remotePath, FtpExists existsMode = FtpExists.Overwrite, bool createRemoteDir = false,
            IProgress<double> progress = null)
        {
            throw new NotImplementedException();
        }

        public bool Upload(byte[] fileData, string remotePath, FtpExists existsMode = FtpExists.Overwrite, bool createRemoteDir = false,
            IProgress<double> progress = null)
        {
            throw new NotImplementedException();
        }

        public bool DownloadFile(string localPath, string remotePath, bool overwrite = true, FtpVerify verifyOptions = FtpVerify.None,
            IProgress<double> progress = null)
        {
            File.WriteAllText(localPath, "dummy");
            return true;
        }

        public virtual bool Download(Stream outStream, string remotePath, IProgress<double> progress = null)
        {
            outStream.Write(new byte[]{1,2,10,100}, 0, 4);
            return true;
        }

        public bool Download(out byte[] outBytes, string remotePath, IProgress<double> progress = null)
        {
            throw new NotImplementedException();
        }

        public Task<int> UploadFilesAsync(IEnumerable<string> localPaths, string remoteDir, FtpExists existsMode, bool createRemoteDir,
            FtpVerify verifyOptions, FtpError errorHandling, CancellationToken token)
        {
            throw new NotImplementedException();
        }

        public Task<int> UploadFilesAsync(IEnumerable<string> localPaths, string remoteDir, FtpExists existsMode = FtpExists.Overwrite,
            bool createRemoteDir = true, FtpVerify verifyOptions = FtpVerify.None, FtpError errorHandling = FtpError.None)
        {
            throw new NotImplementedException();
        }

        public Task<int> DownloadFilesAsync(string localDir, IEnumerable<string> remotePaths, bool overwrite, FtpVerify verifyOptions,
            FtpError errorHandling, CancellationToken token)
        {
            throw new NotImplementedException();
        }

        public Task<int> DownloadFilesAsync(string localDir, IEnumerable<string> remotePaths, bool overwrite = true, FtpVerify verifyOptions = FtpVerify.None,
            FtpError errorHandling = FtpError.None)
        {
            throw new NotImplementedException();
        }

        public Task<bool> UploadFileAsync(string localPath, string remotePath, FtpExists existsMode, bool createRemoteDir,
            FtpVerify verifyOptions, CancellationToken token, IProgress<double> progress)
        {
            throw new NotImplementedException();
        }

        public Task<bool> UploadFileAsync(string localPath, string remotePath, FtpExists existsMode = FtpExists.Overwrite,
            bool createRemoteDir = false, FtpVerify verifyOptions = FtpVerify.None)
        {
            throw new NotImplementedException();
        }

        public Task<bool> UploadAsync(Stream fileStream, string remotePath, FtpExists existsMode, bool createRemoteDir,
            CancellationToken token, IProgress<double> progress)
        {
            throw new NotImplementedException();
        }

        public Task<bool> UploadAsync(byte[] fileData, string remotePath, FtpExists existsMode, bool createRemoteDir,
            CancellationToken token, IProgress<double> progress)
        {
            throw new NotImplementedException();
        }

        public Task<bool> UploadAsync(Stream fileStream, string remotePath, FtpExists existsMode = FtpExists.Overwrite, bool createRemoteDir = false)
        {
            throw new NotImplementedException();
        }

        public Task<bool> UploadAsync(byte[] fileData, string remotePath, FtpExists existsMode = FtpExists.Overwrite, bool createRemoteDir = false)
        {
            throw new NotImplementedException();
        }

        public Task<bool> DownloadFileAsync(string localPath, string remotePath, bool overwrite, FtpVerify verifyOptions,
            CancellationToken token, IProgress<double> progress)
        {
            throw new NotImplementedException();
        }

        public Task<bool> DownloadFileAsync(string localPath, string remotePath, bool overwrite = true, FtpVerify verifyOptions = FtpVerify.None,
            IProgress<double> progress = null)
        {
            throw new NotImplementedException();
        }

        public Task<bool> DownloadAsync(Stream outStream, string remotePath, CancellationToken token, IProgress<double> progress = null)
        {
            throw new NotImplementedException();
        }

        public Task<bool> DownloadAsync(Stream outStream, string remotePath)
        {
            throw new NotImplementedException();
        }

        public Task<byte[]> DownloadAsync(string remotePath, CancellationToken token, IProgress<double> progress = null)
        {
            throw new NotImplementedException();
        }

        public Task<byte[]> DownloadAsync(string remotePath)
        {
            throw new NotImplementedException();
        }

        public FtpHashAlgorithm GetHashAlgorithm()
        {
            throw new NotImplementedException();
        }

        public void SetHashAlgorithm(FtpHashAlgorithm type)
        {
            throw new NotImplementedException();
        }

        public FtpHash GetHash(string path)
        {
            throw new NotImplementedException();
        }

        public FtpHash GetChecksum(string path)
        {
            throw new NotImplementedException();
        }

        public string GetMD5(string path)
        {
            throw new NotImplementedException();
        }

        public string GetXCRC(string path)
        {
            throw new NotImplementedException();
        }

        public string GetXMD5(string path)
        {
            throw new NotImplementedException();
        }

        public string GetXSHA1(string path)
        {
            throw new NotImplementedException();
        }

        public string GetXSHA256(string path)
        {
            throw new NotImplementedException();
        }

        public string GetXSHA512(string path)
        {
            throw new NotImplementedException();
        }

        public Task<FtpHashAlgorithm> GetHashAlgorithmAsync()
        {
            throw new NotImplementedException();
        }

        public Task SetHashAlgorithmAsync(FtpHashAlgorithm type)
        {
            throw new NotImplementedException();
        }

        public Task<FtpHash> GetHashAsync(string path)
        {
            throw new NotImplementedException();
        }

        public Task<FtpHash> GetChecksumAsync(string path)
        {
            throw new NotImplementedException();
        }

        public Task<string> GetMD5Async(string path)
        {
            throw new NotImplementedException();
        }

        public Task<string> GetXCRCAsync(string path)
        {
            throw new NotImplementedException();
        }

        public Task<string> GetXMD5Async(string path)
        {
            throw new NotImplementedException();
        }

        public Task<string> GetXSHA1Async(string path)
        {
            throw new NotImplementedException();
        }

        public Task<string> GetXSHA256Async(string path)
        {
            throw new NotImplementedException();
        }

        public Task<string> GetXSHA512Async(string path)
        {
            throw new NotImplementedException();
        }

        public bool IsDisposed
        {
            get { return _isDisposed; }
        }
        public FtpIpVersion InternetProtocolVersions { get; set; }
        public int SocketPollInterval { get; set; }
        public bool StaleDataCheck { get; set; }

        public bool IsConnected
        {
            get { return _connected; }
        }

        public bool IsDisconected
        {
            get { return _wasDisconected; }
        }

        public bool EnableThreadSafeDataConnections { get; set; }
        public Encoding Encoding { get; set; }
        public string Host { get; set; }
        public int Port { get; set; }
        public NetworkCredential Credentials { get; set; }
        public int MaximumDereferenceCount { get; set; }
        public X509CertificateCollection ClientCertificates { get; }
        public Func<string> AddressResolver { get; set; }
        public IEnumerable<int> ActivePorts { get; set; }
        public FtpDataConnectionType DataConnectionType { get; set; }
        public bool UngracefullDisconnection { get; set; }
        public int ConnectTimeout { get; set; }
        public int ReadTimeout { get; set; }
        public int DataConnectionConnectTimeout { get; set; }
        public int DataConnectionReadTimeout { get; set; }
        public bool SocketKeepAlive { get; set; }
        public FtpCapability Capabilities { get; }
        public FtpHashAlgorithm HashAlgorithms { get; }
        public FtpEncryptionMode EncryptionMode { get; set; }
        public bool DataConnectionEncryption { get; set; }
        public SslProtocols SslProtocols { get; set; }
        public string SystemType { get; }
        public string ConnectionType { get; }
        public FtpParser ListingParser { get; set; }
        public CultureInfo ListingCulture { get; set; }
        public double TimeOffset { get; set; }
        public bool RecursiveList { get; set; }
        public bool BulkListing { get; set; }
        public int BulkListingLength { get; set; }
        public int TransferChunkSize { get; set; }
        public int RetryAttempts { get; set; }
        public uint UploadRateLimit { get; set; }
        public uint DownloadRateLimit { get; set; }
        public FtpDataType UploadDataType { get; set; }
        public FtpDataType DownloadDataType { get; set; }
        public event FtpSslValidation ValidateCertificate;
    }

    public class StubFtpClientFailsToConnect : StubFtpClient
    {
        public override void Connect()
        {
            throw new Exception("can't connect");
        }
    }

    public class StubFtpClientThrowsAfterCreatingFile : StubFtpClient
    {
        public override bool Download(Stream outStream, string remotePath, IProgress<double> progress = null)
        {
            outStream.Write(new byte[]{1,2}, 0, 2);
            throw new FtpException("Something went wrong while downloading file.");
        }
    }
}
```

Answer 1

In ListNonArchiveDirectoresWithFiles Method, once you get the finishedTaskIndex you can choose to simply use tasks.RemoveAt(finishedTaskIndex); instead of

tasks = array
.Except(new[] {finishedTask})
.ToList();

by this you can avoid iterating through whole list with default comparison (EqualityComparer<TElement>.Default) and again transform back to List<>.

Answer 2

While continuing work on this code I have noticed that ListNonArchiveDirectoresWithFiles() already collects all the data retrieved by ListFilesInDirectories() and those methods could be one.

I have also came across [Toub Stephen (2010). Patterns for Parallel Programming, Microsoft]. This book contains a very clear and concise implementation of ParallelWhileNotEmpty() which is exactly what I needed.

Thanks to this the code could become:

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using DC.Meteologica.Commons.Extensions;
using DC.Meteologica.Commons.Helpers;
using FluentFTP;
using log4net;

namespace DC.Meteologica.Commons.Ftp
{
    public class ParallelFtpClient : IDisposable
    {
        private readonly ILog _logger = LogManager.GetLogger(typeof(MeteologicaFtpSyncer).FullName);
        private readonly int _maxConcurrentConnections;
        private readonly Func<IFtpClient> _factory;
        private readonly List<IFtpClient> _ftpClients;

        public ParallelFtpClient(int maxConcurrentConnections, Func<IFtpClient> factory)
        {
            _maxConcurrentConnections = maxConcurrentConnections;
            _factory = factory;
            _ftpClients = new List<IFtpClient>();
        }

        public void Connect()
        {
            var tasks = new List<Task>();
            for (var i = 0; i < _maxConcurrentConnections; i++)
            {
                var ftpClient = _factory();
                _ftpClients.Add(ftpClient);
                var connectTask = ftpClient.ConnectAsync();
                tasks.Add(connectTask);
            }
            Task.WaitAll(tasks.ToArray());
        }

        public void Disconnect()
        {
            var tasks = _ftpClients
                .Select(x => x.DisconnectAsync())
                .ToArray();
            Task.WaitAll(tasks);
        }

        public DownloadResult DownloadFilesParallel(IEnumerable<LocalRemotePathPair> ftpPathLocalPathPairs)
        {
            var clients = new ConcurrentQueue<IFtpClient>(_ftpClients);
            var downloadedFiles = new ConcurrentBag<LocalRemotePathPair>();
            var failedDownloadsBag = new ConcurrentBag<LocalRemotePathPair>();

            Parallel.ForEach(ftpPathLocalPathPairs,
                new ParallelOptions() { MaxDegreeOfParallelism = _maxConcurrentConnections },
                (pair) =>
                {
                    clients.TryDequeue(out var client);

                    _logger.Debug($"Downloading {pair.FtpFilePath} to {pair.LocalFilePath}...");
                    try
                    {
                        var memoryStream = new MemoryStream();
                        client.Download(memoryStream, pair.FtpFilePath);
                        memoryStream.Position = 0;
                        var destinationDirectory = Path.GetDirectoryName(pair.LocalFilePath);
                        if (!Directory.Exists(destinationDirectory))
                        {
                            _logger.Info($"Creating new directory {destinationDirectory}");
                            Directory.CreateDirectory(destinationDirectory);
                        }

                        File.WriteAllBytes(pair.LocalFilePath, memoryStream.ToArray());
                        downloadedFiles.Add(pair);
                    }
                    catch (Exception e)
                    {
                        _logger.Warn($"An unhandled exception occured while downloading file {pair.FtpFilePath}", e);
                        failedDownloadsBag.Add(pair);
                    }

                    clients.Enqueue(client);
                }
            );

            return new DownloadResult(downloadedFiles.ToList(), failedDownloadsBag.ToList());
        }

        public List<string> ListNonArchivedFilesRecursively(string root, CancellationToken token)
        {
            try
            {
                return ListNonArchivedFilesRecursivelyInternal(root, token);
            }
            catch (AggregateException e) when (e.InnerExceptions.Single() is OperationCanceledException)
            {
                throw new OperationCanceledException(token);
            }
        }

        private List<string> ListNonArchivedFilesRecursivelyInternal(string root, CancellationToken token)
        {
            var files = new ConcurrentBag<string>();
            var clients = new ConcurrentQueue<IFtpClient>(_ftpClients);

            Parallel2.WhileNotEmpty(
                new []{root},
                new ParallelOptions() { MaxDegreeOfParallelism = _maxConcurrentConnections },
                (path, adder) =>
            {
                token.ThrowIfCancellationRequested();
                _logger.Debug($"Listing files and folders in {path}...");
                clients.TryDequeue(out var client);
                try
                {
                    var childItems = client.GetListing(path);
                    var fileChildItems = childItems
                        .Where(x => x.Type == FtpFileSystemObjectType.File)
                        .Select(x => x.FullName);
                    var subDirectoriesToQuery = childItems
                        .Where(x => x.Type == FtpFileSystemObjectType.Directory)
                        .Where(x => x.Name != "Archive");
                    files.AddRange(fileChildItems);
                    foreach (var ftpListItem in subDirectoriesToQuery)
                    {
                        adder(ftpListItem.FullName);
                    }
                }
                catch (TimeoutException e)
                {
                    _logger.Info($"Exception while listing items for: {path}. Listing will be retried.", e);
                    adder(path);
                }
                finally
                {
                    clients.Enqueue(client);
                }
            });

            return files.ToList();
        }

        public void Dispose()
        {
            foreach (var ftpClient in _ftpClients)
            {
                ftpClient.Dispose();
            }
        }
    }

    // this is named Parallel2 so it ca coexist with System.Threading.Tasks.Parallel
    internal static class Parallel2
    {
        // from Patterns_of_Parallel_Programming_CSharp.pdf 
        public static void WhileNotEmpty<T>(IEnumerable<T> initialValues, ParallelOptions parallelOptions,
            Action<T, Action<T>> body)
        {
            var from = new ConcurrentQueue<T>(initialValues);
            var to = new ConcurrentQueue<T>();
            while ([email protected])
            {
                Action<T> addMethod = v => to.Enqueue(v);
                Parallel.ForEach(@from, parallelOptions, v => body(v, addMethod));
                @from = to;
                to = new ConcurrentQueue<T>();
            }
        }
    }

    public class DownloadResult
    {
        public DownloadResult(IReadOnlyList<LocalRemotePathPair> downloaded, IReadOnlyList<LocalRemotePathPair> failed)
        {
            Downloaded = downloaded;
            Failed = failed;
        }

        public IReadOnlyList<LocalRemotePathPair> Downloaded { get; }
        public IReadOnlyList<LocalRemotePathPair> Failed { get; }

        public static DownloadResult AllFine(IReadOnlyList<LocalRemotePathPair> downloadedFiles) => new DownloadResult(downloadedFiles, new List<LocalRemotePathPair>());
    }


    public readonly struct LocalRemotePathPair
    {
        public LocalRemotePathPair(string ftpFilePath, string localFilePath) : this()
        {
            FtpFilePath = ftpFilePath;
            LocalFilePath = localFilePath;
        }

        public string FtpFilePath { get; }
        public string LocalFilePath { get; }

        public override string ToString() => $"{FtpFilePath} => {LocalFilePath}";
    }
}

```