package SimpleWriteAndRead;

import com.continuuity.api.data.*;
import com.continuuity.api.flow.flowlet.*;
import com.continuuity.api.flow.flowlet.builders.*;

public class ReaderFlowlet extends AbstractComputeFlowlet {

  @Override
  public void configure(StreamsConfigurator configurator) {
    TupleSchema in = new TupleSchemaBuilder().
        add("row", byte[].class).
        add("column", byte[].class).
        create();
    configurator.getDefaultTupleInputStream().setSchema(in);

    TupleSchema out = new TupleSchemaBuilder().
        add("row", byte[].class).
        add("column", byte[].class).
        add("value", byte[].class).
        create();
    configurator.getDefaultTupleOutputStream().setSchema(out);
  }

  @Override
  public void process(Tuple tuple, TupleContext tupleContext, OutputCollector outputCollector) {
    if (Common.debug)
      System.out.println(this.getClass().getSimpleName() + ": Received tuple " + tuple);

    // perform inline read of row+column
    byte [] row = tuple.get("row");
    byte [] column = tuple.get("column");
    Read read = new Read(row, column);
    ReadOperationExecutor executor =
      getFlowletLaunchContext().getReadExecutor();
    byte [] value = executor.execute(read).get(column);
    
    // output a tuple, adding the value
    Tuple outputTuple = new TupleBuilder().
        set("row", row).
        set("column", column).
        set("value", value).
        create();
    outputCollector.emit(outputTuple);
  }
}

@@ -486,245 +486,8 @@
 in);
-%0A%0A    TupleSchema out = new TupleSchemaBuilder().%0A        add(%22row%22, byte%5B%5D.class).%0A        add(%22column%22, byte%5B%5D.class).%0A        add(%22value%22, byte%5B%5D.class).%0A        create();%0A    configurator.getDefaultTupleOutputStream().setSchema(out);
 %0A  %7D
@@ -1017,237 +1017,226 @@
 ;%0A  
+%0A
   
-%0A    // output a tuple, adding the value%0A    Tuple outputTuple = new TupleBuilder().%0A        set(%22row%22, row).%0A        set(%22column%22, column).%0A        set(%22value%22, value).%0A        create();%0A    outputCollector.emit(outputTuple
+  if (Common.debug)%0A      System.out.println(this.getClass().getSimpleName() + %22: Read value (%22 +%0A          new String(value) + %22) for row (%22 + new String(row) + %22) column (%22 +%0A          new String(column) + %22)%22
 );%0A
+%0A
   %7D%0A

Remove emit of tuple from final flowlet for now