#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>
#include <numeric>
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 map<string, int> SplitIntoWords(const string& str)
{
    map<string, int> result;
    auto str_begin = begin(str);
    const auto str_end = end(str);
    while (true)
    {
        auto it = find(str_begin , str_end , ' ');

        result[string(str_begin , it)]++;

        if (it == str_end)
        {
            break;
        }
        else
        {
            str_begin = it + 1;
        }
    }
    return result;
}


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

    Stats lineStats;
    for (const string& key : key_words)
    {
        if (words.count(key) > 0)
        {
            lineStats.word_frequences[key] = words.at(key);
        }
    }

    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 = 8)
{
    vector<stringstream> streams;
    streams.reserve(threadCount);

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

    size_t streamNum = 0;
    for (string line; getline(input, line);)
    {
        streams[streamNum++] << line << "\n";

        if(streamNum==threadCount)
        {
            streamNum=0;
        }
    }

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


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

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

    return result;
}