reversewords.cs

// https://msdn.microsoft.com/en-us/library/hh228604(v=vs.110).aspx

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Net.Http;
using System.Text;
using System.Threading.Tasks;
using System.Threading.Tasks.Dataflow;

namespace ReverseWords
{
    public class WordReverser
    {
        public WordReverser()
        {
        }

        public void FindReversals()
        {
            var logger = new Logger();
            var readerParallelism = 2; // DataflowBlockOptions.Unbounded;
            var analyserParallelism = 1; // DataflowBlockOptions.Unbounded;
            var printerParallelism = 1;

            var loadFileAsync = new TransformBlock<string, Optional<string>>(
                new Func<string, Task<Optional<string>>>(async path =>
                {
                    logger.Debug(this, "Loading {0}", (object)path);

                    Optional<string> optionalString = null;

                    try
                    {
                        var uniencoding = new UTF8Encoding();
                        byte[] fileText;

                        using (FileStream SourceStream = File.Open(path, FileMode.Open))
                        {
                            fileText = new byte[SourceStream.Length];
                            await SourceStream.ReadAsync(fileText, 0, (int)SourceStream.Length);
                        }

                        optionalString = new Optional<string>(uniencoding.GetString(fileText));
                    }
                    catch (Exception ex)
                    {
                        optionalString = new Optional<string>(ex);
                    }

                    return optionalString;
                }), new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = readerParallelism });

            var downloadStringAsync = new TransformBlock<string, Optional<string>>(
                new Func<string, Task<Optional<string>>>(async uri =>
                {
                    logger.Debug(this, "Downloading {0} ...", (object)uri);

                    Optional<string> optionalString = null;
                    try
                    {
                        var client = new HttpClient();
                        var result = await client.GetStringAsync(uri);
                        optionalString = new Optional<string>(result);
                    }
                    catch (Exception ex)
                    {
                        optionalString = new Optional<string>(ex);
                    }

                    return optionalString;
                }));

            //var downloadString = new TransformBlock<string, string>((uri) =>
            //{
            //    Console.WriteLine("Downloading {0} ...", uri);

            //    return new WebClient().DownloadString(uri);
            //});

            var createWordList = new TransformBlock<Optional<string>, Optional<string[]>>((optionalText) =>
            {
                if (optionalText.IsFaulted)
                    return new Optional<string[]>(optionalText.Fault);

                logger.Debug(this, "Creating word list...");

                var text = optionalText.Value;

                char[] tokens = text.ToArray();
                for (int i = 0; i < tokens.Length; i++)
                {
                    if (!char.IsLetter(tokens[i]))
                        tokens[i] = ' ';
                }
                text = new string(tokens);

                var words = text.Split(new char[] { ' ' }, StringSplitOptions.RemoveEmptyEntries);

                return new Optional<string[]>(words);
            }, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = analyserParallelism });

            var filterWordList = new TransformBlock<Optional<string[]>, Optional<string[]>>((optionalWords) =>
            {
                if (optionalWords.IsFaulted)
                    return new Optional<string[]>(optionalWords.Fault);

                logger.Debug(this, "Filtering word list...");

                var wordList = optionalWords.Value
                    .Where(word => word.Length > 2)
                    .OrderBy(word => word.ToLower())
                    .Distinct()
                    .ToArray();

                logger.Debug(this, "Found {0} words", wordList.Length);

                return new Optional<string[]>(wordList);
            }, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = analyserParallelism });

            var findReversedWords = new TransformManyBlock<Optional<string[]>, Optional<string>>((optionalWords) =>
            {
                var reversedWords = new ConcurrentQueue<Optional<string>>();

                if (optionalWords.IsFaulted)
                    reversedWords.Enqueue(new Optional<string>(optionalWords.Fault));
                else
                {
                    logger.Debug(this, "Checking for reversible words...");

                    var words = optionalWords.Value;

                    Parallel.ForEach(words, word =>
                    {
                        var reverse = new string(word.Reverse().ToArray());

                        if (Array.BinarySearch<string>(words, reverse) >= 0 && word != reverse)
                            reversedWords.Enqueue(new Optional<string>(word));
                    });
                }

                return reversedWords;
            }, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = analyserParallelism });

            var printReversedWords = new ActionBlock<Optional<string>>((optionalWord) =>
            {
                if (optionalWord.IsFaulted)
                {
                    logger.Debug(this, "Failed: {0}", optionalWord.Fault);
                    return;
                }

                var reversedWord = optionalWord.Value;
                logger.Debug(this, "Found reversed word {0} / {1}", reversedWord, (object)new string(reversedWord.Reverse().ToArray()));
            }, new ExecutionDataflowBlockOptions { MaxDegreeOfParallelism = printerParallelism });

            loadFileAsync.LinkTo(createWordList, new DataflowLinkOptions { PropagateCompletion = true });
            //downloadStringAsync.LinkTo(createWordList, new DataflowLinkOptions { PropagateCompletion = true });
            createWordList.LinkTo(filterWordList, new DataflowLinkOptions { PropagateCompletion = true });
            filterWordList.LinkTo(findReversedWords, new DataflowLinkOptions { PropagateCompletion = true });
            findReversedWords.LinkTo(printReversedWords, new DataflowLinkOptions { PropagateCompletion = true });

            Task.Run(async () =>
            {
                await loadFileAsync.SendAsync("C:\\Projects\\ReverseWords\\Iliad.txt");
                await loadFileAsync.SendAsync("C:\\Projects\\ReverseWords\\Odyssey.txt");
                await loadFileAsync.SendAsync("C:\\Projects\\ReverseWords\\Ion.txt");
                //await downloadStringAsync.SendAsync("http://www.gutenberg.org/files/6130/6130-0.txt");
                //await downloadStringAsync.SendAsync("http://www.gutenberg.org/cache/epub/1727/pg1727.txt");
                //downloadStringAsync.Post("http://www.gutenberg.org/cache/epub/1635/pg1635.txt");
                //downloadStringAsync.Complete();
                loadFileAsync.Complete();
            });


            printReversedWords.Completion.Wait();
            
            Console.Read();
        }
    }
}