package com.continuuity.examples.countandfilterwords;

import com.continuuity.api.annotation.Output;
import com.continuuity.api.flow.flowlet.AbstractFlowlet;
import com.continuuity.api.flow.flowlet.OutputEmitter;

public class Tokenizer extends AbstractFlowlet {

  @Output("tokens")
  private OutputEmitter<String> tokenOutput;

  @Output("counts")
  private OutputEmitter<String> countOutput;

  public void process(String line) {
    // Tokenize and emit each token to the filters
    for (String token : line.split("[ .-]")) {
      tokenOutput.emit(token);
    }
    // Also emit once to the 'all' counter
    countOutput.emit("all");
  }
}

@@ -557,20 +557,16 @@
 ken);%0A  
-  %7D%0A
     // A
@@ -578,13 +578,8 @@
 mit 
-once 
 to t
@@ -594,17 +594,34 @@
  counter
-%0A
+ for each token%0A  
     coun
@@ -641,14 +641,20 @@
 %22all%22);%0A
+    %7D%0A
   %7D%0A%7D%0A

Update to make countAll count tokens not lines