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

import com.malhartech.api.BaseOperator;
import com.malhartech.api.CheckpointListener;
import com.malhartech.api.Context.OperatorContext;
import com.malhartech.api.DAG;
import com.malhartech.api.DefaultInputPort;
import com.malhartech.engine.RecoverableInputOperator;
import java.io.IOException;

import java.util.HashSet;
import com.malhartech.netlet.DefaultEventLoop;
import org.junit.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * @author Chetan Narsude <chetan@malhar-inc.com>
 */
public class NodeRecoveryTest
{
  private static final Logger logger = LoggerFactory.getLogger(NodeRecoveryTest.class);
  static HashSet<Long> collection = new HashSet<Long>(20);

  @Before
  public void setup() throws IOException
  {
    StramChild.eventloop = new DefaultEventLoop("NodeRecoveryTestEventLoop");
  }

  @After
  public void teardown()
  {
    StramChild.eventloop.stop();

  }

  public static class CollectorOperator extends BaseOperator implements CheckpointListener
  {
    private boolean simulateFailure;
    private long checkPointWindowId;

    public final transient DefaultInputPort<Long> input = new DefaultInputPort<Long>(this)
    {
      @Override
      public void process(Long tuple)
      {
//        logger.debug("adding the tuple {}", Codec.getStringWindowId(tuple));
        collection.add(tuple);
      }

    };

    /**
     * @param simulateFailure the simulateFailure to set
     */
    public void setSimulateFailure(boolean simulateFailure)
    {
      this.simulateFailure = simulateFailure;
    }

    @Override
    public void setup(OperatorContext context)
    {
      simulateFailure &= (checkPointWindowId == 0);
      logger.debug("simulateFailure = {}", simulateFailure);
    }

    @Override
    public void checkpointed(long windowId)
    {
      if (this.checkPointWindowId == 0) {
        this.checkPointWindowId = windowId;
      }
    }

    @Override
    public void committed(long windowId)
    {
      logger.debug("committed window {} and checkPointWindowId {}", windowId, checkPointWindowId);
      if (simulateFailure && windowId > this.checkPointWindowId && this.checkPointWindowId > 0) {
        throw new RuntimeException("Failure Simulation from " + this + " checkpointWindowId=" + checkPointWindowId);
      }
    }

  }

  @Test
  public void testInputOperatorRecovery() throws Exception
  {
    collection.clear();
    int maxTuples = 30;
    DAG dag = new DAG();
    dag.getAttributes().attr(DAG.STRAM_CHECKPOINT_WINDOW_COUNT).set(2);
    dag.getAttributes().attr(DAG.STRAM_WINDOW_SIZE_MILLIS).set(300);
    dag.getAttributes().attr(DAG.STRAM_MAX_CONTAINERS).set(1);
    RecoverableInputOperator rip = dag.addOperator("LongGenerator", RecoverableInputOperator.class);
    rip.setMaximumTuples(maxTuples);

    CollectorOperator cm = dag.addOperator("LongCollector", CollectorOperator.class);
    dag.addStream("connection", rip.output, cm.input);

    StramLocalCluster lc = new StramLocalCluster(dag);
    lc.run();

    Assert.assertEquals("Generated Outputs", maxTuples, collection.size());
  }

  @Test
  public void testOperatorRecovery() throws Exception
  {
    collection.clear();
    int maxTuples = 30;
    DAG dag = new DAG();
    dag.getAttributes().attr(DAG.STRAM_CHECKPOINT_WINDOW_COUNT).set(2);
    dag.getAttributes().attr(DAG.STRAM_WINDOW_SIZE_MILLIS).set(300);
    dag.getAttributes().attr(DAG.STRAM_MAX_CONTAINERS).set(1);
    RecoverableInputOperator rip = dag.addOperator("LongGenerator", RecoverableInputOperator.class);
    rip.setMaximumTuples(maxTuples);

    CollectorOperator cm = dag.addOperator("LongCollector", CollectorOperator.class);
    cm.setSimulateFailure(true);
    dag.addStream("connection", rip.output, cm.input);

    StramLocalCluster lc = new StramLocalCluster(dag);
    lc.run();

//    for (Long l: collection) {
//      logger.debug(Codec.getStringWindowId(l));
//    }
    Assert.assertEquals("Generated Outputs", maxTuples, collection.size());
  }

  @Test
  public void testInlineOperatorsRecovery() throws Exception
  {
    collection.clear();
    int maxTuples = 30;
    DAG dag = new DAG();
    //dag.getAttributes().attr(DAG.STRAM_HEARTBEAT_INTERVAL_MILLIS).set(400);
    dag.getAttributes().attr(DAG.STRAM_CHECKPOINT_WINDOW_COUNT).set(2);
    dag.getAttributes().attr(DAG.STRAM_WINDOW_SIZE_MILLIS).set(300);
    dag.getAttributes().attr(DAG.STRAM_MAX_CONTAINERS).set(1);
    RecoverableInputOperator rip = dag.addOperator("LongGenerator", RecoverableInputOperator.class);
    rip.setMaximumTuples(maxTuples);

    CollectorOperator cm = dag.addOperator("LongCollector", CollectorOperator.class);
    cm.setSimulateFailure(true);
    dag.addStream("connection", rip.output, cm.input).setInline(true);

    StramLocalCluster lc = new StramLocalCluster(dag);
    lc.run();

//    for (Long l: collection) {
//      logger.debug(Codec.getStringWindowId(l));
//    }
    Assert.assertEquals("Generated Outputs", maxTuples, collection.size());
  }

}

@@ -355,24 +355,71 @@
 utOperator;%0A
+import com.malhartech.netlet.DefaultEventLoop;%0A
 import java.
@@ -434,17 +434,16 @@
 eption;%0A
-%0A
 import j
@@ -471,46 +471,73 @@
 ort 
-com.malhartech.netlet.DefaultEventLoop
+org.junit.After;%0Aimport org.junit.Assert;%0Aimport org.junit.Before
 ;%0Aim
@@ -551,17 +551,20 @@
 g.junit.
-*
+Test
 ;%0Aimport
@@ -2462,26 +2462,24 @@
     %7D%0A%0A  %7D%0A%0A
-  
 @Test%0A  publ
@@ -4135,34 +4135,32 @@
 n.size());%0A  %7D%0A%0A
-  
 @Test%0A  public v

Make inputOperator also fail and recover.