public static <P extends Platform<P>>
    JTailProducer<P, Tuple2<String, Tuple2<Option<Long>, Long>>> generate(
          JProducer<P, String> source, Store<P, ?, String, Long> store) {
    return source.flatMap(new Function<String, Iterable<String>>() {
      public Iterable<String> apply(String sentence) {
        return toWords(sentence);
      }
    }).mapToKeyed(new Function<String, Tuple2<String, Long>>() {
      public Tuple2<String, Long> apply(String word) {
        return new Tuple2<String, Long>(word, 1L);
      }
    }).sumByKey(store, longMonoid());
  }

import static com.twitter.algebird.javaapi.Monoids.longMonoid;
import scala.Option;
import scala.Tuple2;
import com.twitter.summingbird.Platform;
import com.twitter.summingbird.javaapi.Function;
import com.twitter.summingbird.javaapi.JProducer;
import com.twitter.summingbird.javaapi.JTailProducer;
import com.twitter.summingbird.javaapi.Store;