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("key", byte[].class).
        create();
    configurator.getDefaultTupleInputStream().setSchema(in);
  }

  @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 key
    byte [] key = tuple.get("key");
    Read read = new Read(key);
    ReadOperationExecutor executor =
      getFlowletLaunchContext().getReadExecutor();
    byte [] value = executor.execute(read).getKeyResult();
  
    if (Common.debug)
      System.out.println(this.getClass().getSimpleName() + ": Read value (" +
          new String(value) + ") for key (" + new String(key) + ")");

  }
}

@@ -750,16 +750,19 @@
     Read
+Key
  read = 
@@ -769,16 +769,19 @@
 new Read
+Key
 (key);%0A 
@@ -913,23 +913,8 @@
 ead)
-.getKeyResult()
 ;%0A  

Change read-write flow to use ReadKey