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

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Defines the destination for tuples processed.
 *
 * @author Chetan Narsude <chetan@malhar-inc.com>
 */
public class StreamContext implements Context
{
  public static enum State
  {
    UNDEFINED,
    OUTSIDE_WINDOW,
    INSIDE_WINDOW,
    TERMINATED
  }
  private final Logger LOG = LoggerFactory.getLogger(StreamContext.class);
  private Sink sink;
  private SerDe serde;
  private int tupleCount;
  private State sinkState;

  public StreamContext()
  {
    sinkState = State.UNDEFINED;
  }

  /**
   * @param sink - target node, not required for output adapter
   */
  public void setSink(Sink sink)
  {
    LOG.debug("setSink: {}", sink);
    this.sink = sink;
  }

  public SerDe getSerDe()
  {
    return serde; // required for socket connection
  }

  public void setSerde(SerDe serde)
  {
    this.serde = serde;
  }

  public void sink(Tuple t)
  {
    //LOG.info(this + " " + t);
    switch (t.getType()) {
      case SIMPLE_DATA:
      case PARTITIONED_DATA:
        tupleCount++;
        break;

      case BEGIN_WINDOW:
        tupleCount = 0;
        break;

      case END_WINDOW:
        if (tupleCount != ((EndWindowTuple) t).getTupleCount()) {
          EndWindowTuple ewt = new EndWindowTuple();
          ewt.setTupleCount(tupleCount);
          ewt.setWindowId(t.getWindowId());
          t = ewt;
          break;
        }
    }

    t.setContext(this);
    sink.doSomething(t);
  }

  public int getTupleCount()
  {
    return tupleCount;
  }

  State getSinkState()
  {
    return sinkState;
  }

  void setSinkState(State state)
  {
    sinkState = state;
  }

  @Override
  public String toString()
  {
    return " tuples = " + tupleCount + " state = " + sinkState;
  }
}

@@ -822,24 +822,504 @@
  sink;%0A  %7D%0A%0A
+  public final void setSink(final Sink ultimateSink, final long fromWindowId)%0A  %7B%0A    LOG.debug(%22setSink: %7B%7D after window %7B%7D%22, ultimateSink, fromWindowId);%0A      %0A    sink = new Sink()%0A    %7B%0A      @Override%0A      public void doSomething(Tuple t)%0A      %7B%0A        if (fromWindowId %3C= t.getWindowId()) %7B%0A          LOG.debug(%22Sink %7B%7D kicking in after window %7B%7D%22, ultimateSink, fromWindowId);%0A          sink = ultimateSink;%0A          sink.doSomething(t);%0A        %7D%0A      %7D%0A    %7D;%0A  %7D%0A%0A
   public Ser

Provision to ignore all the tuples before a given tuple while addressing node failure.