#include "mapreduce.hpp"
#include <iostream>
#include <istream>
#include <ostream>
#include <sstream>


using namespace std;
using namespace mapreduce;


int
StringToInt(std::string str)
{
  std::istringstream sin(str);
  int ret;
  sin>>ret;
  return ret;
}

std::string
IntToString(int i)
{
  std::ostringstream sout;
  sout<<i;
  return sout.str();
}


class WordCounter : public Mapper
{
public:
  virtual void Map(const MapInput& input) 
  {
    const string& text = input.value();
    const int n = text.size();

    cerr<<text<<endl;

    for (int i = 0; i < n; )
      {
	// Skip past leading whitespace
	while ((i < n) && isspace(text[i]))
	  i++;

	// Find word end
	int start = i;
	while ((i < n) && !isspace(text[i]))
	  i++;

	if (start < i)
	  {
	    cerr<<"Emitting: "<<text.substr(start, i-start)<<"\n";
	    Emit(text.substr(start,i-start),"1");
	  }
      }
  }
};

// REGISTER_MAPPER(WordCounter);


class Adder : public Reducer
{
  virtual void Reduce(ReduceInput* input)
  {
    // Iterate over all entries with the
    // same key and add the values
    int value = 0;
    while (!input->done())
      {
	value += StringToInt(input->value());
	input->NextValue();
      }

    cerr<<"Key: "<<input->key()<<"\n";
    // Emit sum for input->key()
    Emit(IntToString(value));
  }
};

// REGISTER_REDUCER(Adder);


int
main()
{
  WordCounter wc;
  Adder a;
  MapReduceSpecification spec;

  spec.mapper = &wc;
  spec.reducer = &a;

  MapReduceInput* input = spec.add_input();
  input->set_filebase("./data/input/");

  MapReduceOutput* out = spec.output();
  out->set_filebase("./data/output/");

  MapReduceResult result;
  if (!MapReduce(spec, result)) abort();

  return 0;
}
