#include "wordcount.hh"


void
WordcountMapper::map(HadoopPipes::MapContext &context)
{
  typedef boost::tokenizer<> tokenizer_t;
  tokenizer_t tokenizer(context.getInputValue());

  for(tokenizer_t::const_iterator i = tokenizer.begin();
      tokenizer.end() != i; ++i) {
    context.emit(boost::to_lower_copy(*i), "1");
  }
}

void
WordcountReducer::reduce(HadoopPipes::ReduceContext &context)
{
  uint32_t count(0);

  do {
	  ++count;
  } while(context.nextValue());

  //std::cout << context.getInputKey() << endl;
  context.emit(context.getInputKey(),
               boost::lexical_cast<std::string>(count));
}


int
main(int argc, char *argv[])
{
  HadoopPipes::TemplateFactory2<WordcountMapper,
                                WordcountReducer> factory;
  return HadoopPipes::runTask(factory);
}