#include "test_runner.h"
#include "profile.h"

#include <map>
#include <string>
#include <sstream>
#include <string_view>
#include <vector>
#include <iterator>
#include <algorithm>
#include <functional>
#include <future>
using namespace std;

struct Stats
{
    map<string, int> word_frequences;

    void operator +=(const Stats& other)
    {
        for (const auto& kvp : other.word_frequences)
        {
            word_frequences[kvp.first] += kvp.second;
        }
    }
};

const vector<string_view> SplitIntoWordsView(const string& s)
{
    string_view str = s;
    vector<string_view> result;
    size_t pos = 0;
    const size_t pos_end = str.npos;
    while (true)
    {
        size_t space = str.find(' ', pos);

        string_view word = str.substr(pos, space - pos);

        result.push_back(word);

        if (space == pos_end)
        {
            break;
        }
        else
        {
            pos = space + 1;
        }
    }

    return result;
}

Stats ExploreLine(const set<string>& key_words, const string& line)
{
    const auto& words = SplitIntoWordsView(line);

    Stats lineStats;
    for (string_view key : key_words)
    {
        int* word_freq = nullptr;

        bool first = true;

        for (string_view word : words)
        {
            if (word == key)
            {
                if (first)
                {
                    string k = static_cast<string>(key);
                    word_freq = &lineStats.word_frequences[k];

                    first = false;
                }

                (*word_freq)++;
            }
        }
    }

    return lineStats;
}

Stats ExploreKeyWordsSingleThread(const set<string>& key_words, istream& input)
{
    Stats result;
    for (string line; getline(input, line);)
    {
        result += ExploreLine(key_words, line);
    }
    return result;
}

Stats ExploreKeyWords(const set<string>& key_words, istream& input, size_t threadCount = 4)
{
    vector<stringstream> streams;
    streams.reserve(threadCount);

    for(size_t i=0;i<threadCount;++i)
    {
        stringstream ss;
        streams.push_back(move(ss));
    }

    int streamNum = 0;
    for (string line; getline(input, line);)
    {
        streams[streamNum] << line << "\n";
        streamNum = (streamNum + 1) % threadCount;
    }

    vector<future<Stats>> futures;
    futures.reserve(threadCount);

    for (auto& substream : streams)
    {
        futures.push_back(async([&key_words, &substream]
        {
            return ExploreKeyWordsSingleThread(
                    key_words, substream
            );
        }));
    }

    Stats result;
    for (auto& _future : futures)
    {
        result += _future.get();
    }

    return result;
}

void TestBasic()
{
    const set<string> key_words =
    { "yangle", "rocks", "sucks", "all" };

    stringstream ss;
    ss << "this new yangle service really rocks\n";
    ss << "It sucks when yangle isn't available\n";
    ss << "10 reasons why yangle is the best IT company\n";
    ss << "yangle rocks others suck\n";
    ss << "Goondex really sucks, but yangle rocks. Use yangle\n";

    const auto stats = ExploreKeyWords(key_words, ss);
    const map<string, int> expected =
    {
    { "yangle", 6 },
    { "rocks", 2 },
    { "sucks", 1 } };
    ASSERT_EQUAL(stats.word_frequences, expected);
}

int main()
{
    TestRunner tr;
    RUN_TEST(tr, TestBasic);
}