/*
 *  Copyright (c) 2012 Malhar, Inc.
 *  All Rights Reserved.
 */
package com.malhartech.engine;

import com.malhartech.api.CheckpointListener;
import com.malhartech.api.Context.OperatorContext;
import com.malhartech.api.DefaultOutputPort;
import com.malhartech.api.InputOperator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * @author Chetan Narsude <chetan@malhar-inc.com>
 */
public class RecoverableInputOperator implements InputOperator, CheckpointListener
{
  public final transient DefaultOutputPort<Long> output = new DefaultOutputPort<Long>(this);
  transient boolean first;
  transient long windowId;
  int checkpointCount;
  int maximumTuples = 20;

  public void setMaximumTuples(int count)
  {
    maximumTuples = count;
  }

  @Override
  public void emitTuples()
  {
    if (first) {
//      logger.debug("generating tuple {}", Codec.getStringWindowId(windowId));
      output.emit(windowId);
      first = false;
      if (--maximumTuples == 0) {
        throw new RuntimeException(new InterruptedException("Just want to stop!"));
      }
    }
  }

  @Override
  public void beginWindow(long windowId)
  {
    this.windowId = windowId;
    first = true;
  }

  @Override
  public void endWindow()
  {
  }

  @Override
  public void setup(OperatorContext context)
  {
    if (checkpointCount > 0) {
      checkpointCount = 3;
    }

    logger.debug("checkpointCount = {}", checkpointCount);
  }

  @Override
  public void teardown()
  {
  }

  @Override
  public void checkpointed(long windowId)
  {
    checkpointCount++;
  }

  @Override
  public void committed(long windowId)
  {
    logger.debug("committed window {} and checkpoint {}", windowId, checkpointCount);
    if (checkpointCount == 2) {
      throw new RuntimeException("Failure Simulation from " + this);
    }
  }

  private static final Logger logger = LoggerFactory.getLogger(RecoverableInputOperator.class);
}

@@ -573,16 +573,72 @@
 (this);%0A
+  long checkpointedWindowId;%0A  boolean firstRun = true;%0A
   transi
@@ -687,31 +687,8 @@
 Id;%0A
-  int checkpointCount;%0A
   in
@@ -852,88 +852,8 @@
 ) %7B%0A
-//      logger.debug(%22generating tuple %7B%7D%22, Codec.getStringWindowId(windowId));%0A
     
@@ -997,25 +997,12 @@
 on(%22
-Just want to stop
+Done
 !%22))
@@ -1248,126 +1248,45 @@
     
-if (checkpointCount %3E 0) %7B%0A      checkpointCount = 3;%0A    %7D%0A%0A    logger.debug(%22checkpointCount = %7B%7D%22, checkpointCount)
+firstRun &= checkpointedWindowId == 0
 ;%0A  
@@ -1396,16 +1396,20 @@
   %7B%0A    
+if (
 checkpoi
@@ -1414,16 +1414,71 @@
 oint
-Count++;
+edWindowId == 0) %7B%0A      checkpointedWindowId = windowId;%0A    %7D
 %0A  %7D
@@ -1542,61 +1542,51 @@
     
-logger.debug(%22committed window %7B%7D and checkpoint %7B%7D%22,
+if (firstRun && checkpointedWindowId %3E 0 &&
  win
@@ -1590,17 +1590,18 @@
 windowId
-,
+ %3E
  checkpo
@@ -1607,44 +1607,18 @@
 oint
-Count);%0A    if (checkpointCount == 2
+edWindowId
 ) %7B%0A

Make inputOperator also fail and recover.