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

import com.malhartech.annotation.NodeAnnotation;
import com.malhartech.annotation.PortAnnotation;
import com.malhartech.util.CircularBuffer;
import java.nio.BufferOverflowException;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map.Entry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * @author <NAME> <<EMAIL>>
 */
public abstract class AbstractInputNode implements Node
{
  private static final Logger logger = LoggerFactory.getLogger(AbstractInputNode.class);
  private transient String id;
  private transient HashMap<String, CircularBuffer<Object>> afterBeginWindows;
  private transient HashMap<String, CircularBuffer<Tuple>> afterEndWindows;
  private transient HashMap<String, Sink> outputs = new HashMap<String, Sink>();
  private transient volatile Sink[] sinks = new Sink[0];
  private transient NodeContext ctx;
  private transient int producedTupleCount;
  private transient int spinMillis;
  private transient int bufferCapacity;

  @Override
  public void setup(NodeConfiguration config)
  {
    id = config.get("id");
    spinMillis = config.getInt("SpinMillis", 100);
    bufferCapacity = config.getInt("BufferCapacity", 1024);

    afterBeginWindows = new HashMap<String, CircularBuffer<Object>>();
    afterEndWindows = new HashMap<String, CircularBuffer<Tuple>>();

    Class<? extends Node> clazz = this.getClass();
    NodeAnnotation na = clazz.getAnnotation(NodeAnnotation.class);
    if (na != null) {
      PortAnnotation[] ports = na.ports();
      for (PortAnnotation pa: ports) {
        if (pa.type() == PortAnnotation.PortType.OUTPUT || pa.type() == PortAnnotation.PortType.BIDI) {
          afterBeginWindows.put(pa.name(), new CircularBuffer<Object>(bufferCapacity));
          afterEndWindows.put(pa.name(), new CircularBuffer<Tuple>(bufferCapacity));
        }
      }
    }
  }

  @Override
  public void activate(NodeContext context)
  {
    ctx = context;
    sinks = new Sink[outputs.size()];

    int i = 0;
    for (Sink s: outputs.values()) {
      sinks[i++] = s;
    }
  }

  @Override
  public void deactivate()
  {
    sinks = new Sink[0];
  }

  @Override
  public void teardown()
  {
    outputs.clear();

    afterEndWindows.clear();
    afterEndWindows = null;
    afterBeginWindows.clear();
    afterBeginWindows = null;
    // Should make variable "shutdown" part of AbstrastInputNode, users should not have to override teardown()
    // as they may forget to call super.teardown()
    // Also move "outputconnected" here as that is a very common need
  }

  @Override
  public final Sink connect(String port, Sink component)
  {
    Sink retvalue;
    if (Component.INPUT.equals(port)) {
      retvalue = this;
    }
    else if (component == null) {
      outputs.remove(port);
      retvalue = null;
    }
    else {
      outputs.put(port, component);
      retvalue = null;
    }

    connected(port, component);
    return retvalue;
  }

  public void connected(String id, Sink dagpart)
  {
    /* implementation to be optionally overridden by the user */
  }

  @Override
  public final void process(Object payload)
  {
    Tuple t = (Tuple)payload;
    switch (t.getType()) {
      case BEGIN_WINDOW:
        for (int i = sinks.length; i-- > 0;) {
          try {
            sinks[i].process(payload);
          }
          catch (MutatedSinkException mse) {
            final Sink newSink = mse.getNewSink();
            newSink.process(payload);
            sinks[i] = newSink;
            replaceOutput(mse.getOldSink(), newSink);
          }
        }

        for (Entry<String, CircularBuffer<Object>> e: afterBeginWindows.entrySet()) {
          final Sink s = outputs.get(e.getKey());
          CircularBuffer<?> cb = e.getValue();
          for (int i = cb.size(); i > 0; i--) {
            s.process(cb.get());
          }
        }
        break;

      case END_WINDOW:
        for (Entry<String, CircularBuffer<Object>> e: afterBeginWindows.entrySet()) {
          final Sink s = outputs.get(e.getKey());
          CircularBuffer<?> cb = e.getValue();
          for (int i = cb.size(); i > 0; i--) {
            s.process(cb.get());
          }
        }
        for (final Sink s: sinks) {
          s.process(payload);
        }

        ctx.report(producedTupleCount, 0L, ((Tuple)payload).getWindowId());
        producedTupleCount = 0;

        // the default is UNSPECIFIED which we ignore anyways as we ignore everything
        // that we do not understand!
        try {
          switch (ctx.getRequestType()) {
            case BACKUP:
              ctx.backup(this, ((Tuple)payload).getWindowId());
              break;

            case RESTORE:
              logger.info("restore requests are not implemented");
              break;
          }
        }
        catch (Exception e) {
          logger.warn("Exception while catering to external request", e.getLocalizedMessage());
        }

        // i think there should be just one queue instead of one per port - lets defer till we find an example.
        for (Entry<String, CircularBuffer<Tuple>> e: afterEndWindows.entrySet()) {
          final Sink s = outputs.get(e.getKey());
          CircularBuffer<?> cb = e.getValue();
          for (int i = cb.size(); i > 0; i--) {
            s.process(cb.get());
          }
        }
        break;

      default:
        for (final Sink s: sinks) {
          s.process(payload);
        }
    }
  }

  @SuppressWarnings("SleepWhileInLoop")
  public void emit(String id, Object payload)
  {
    // once we have annotation done property, we would like to send the tuples to<commit_msg>Looks like there is some cost to try catch block and I would like to minimize it. WIth the current MutatingSink implementation, we pay the small cost on every begin window. An attempt to minimize it. May be we need a better method to mutate the sink.<commit_after>/*
 *  Copyright (c) 2012 Malhar, Inc.
 *  All Rights Reserved.
 */
package com.malhartech.dag;

import com.malhartech.annotation.NodeAnnotation;
import com.malhartech.annotation.PortAnnotation;
import com.malhartech.util.CircularBuffer;
import java.nio.BufferOverflowException;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map.Entry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * @author <NAME> <<EMAIL>>
 */
public abstract class AbstractInputNode implements Node
{
  private static final Logger logger = LoggerFactory.getLogger(AbstractInputNode.class);
  private transient String id;
  private transient HashMap<String, CircularBuffer<Object>> afterBeginWindows;
  private transient HashMap<String, CircularBuffer<Tuple>> afterEndWindows;
  private transient HashMap<String, Sink> outputs = new HashMap<String, Sink>();
  private transient volatile Sink[] sinks = new Sink[0];
  private transient NodeContext ctx;
  private transient int producedTupleCount;
  private transient int spinMillis;
  private transient int bufferCapacity;

  @Override
  public void setup(NodeConfiguration config)
  {
    id = config.get("id");
    spinMillis = config.getInt("SpinMillis", 100);
    bufferCapacity = config.getInt("BufferCapacity", 1024);

    afterBeginWindows = new HashMap<String, CircularBuffer<Object>>();
    afterEndWindows = new HashMap<String, CircularBuffer<Tuple>>();

    Class<? extends Node> clazz = this.getClass();
    NodeAnnotation na = clazz.getAnnotation(NodeAnnotation.class);
    if (na != null) {
      PortAnnotation[] ports = na.ports();
      for (PortAnnotation pa: ports) {
        if (pa.type() == PortAnnotation.PortType.OUTPUT || pa.type() == PortAnnotation.PortType.BIDI) {
          afterBeginWindows.put(pa.name(), new CircularBuffer<Object>(bufferCapacity));
          afterEndWindows.put(pa.name(), new CircularBuffer<Tuple>(bufferCapacity));
        }
      }
    }
  }

  @Override
  public void activate(NodeContext context)
  {
    ctx = context;
    sinks = new Sink[outputs.size()];

    int i = 0;
    for (Sink s: outputs.values()) {
      sinks[i++] = s;
    }
  }

  @Override
  public void deactivate()
  {
    sinks = new Sink[0];
  }

  @Override
  public void teardown()
  {
    outputs.clear();

    afterEndWindows.clear();
    afterEndWindows = null;
    afterBeginWindows.clear();
    afterBeginWindows = null;
    // Should make variable "shutdown" part of AbstrastInputNode, users should not have to override teardown()
    // as they may forget to call super.teardown()
    // Also move "outputconnected" here as that is a very common need
  }

  @Override
  public final Sink connect(String port, Sink component)
  {
    Sink retvalue;
    if (Component.INPUT.equals(port)) {
      retvalue = this;
    }
    else if (component == null) {
      outputs.remove(port);
      retvalue = null;
    }
    else {
      outputs.put(port, component);
      retvalue = null;
    }

    connected(port, component);
    return retvalue;
  }

  public void connected(String id, Sink dagpart)
  {
    /* implementation to be optionally overridden by the user */
  }

  @Override
  public final void process(Object payload)
  {
    int i;
    Tuple t = (Tuple)payload;
    switch (t.getType()) {
      case BEGIN_WINDOW:
        i = sinks.length;
        do {
          try {
            while (i-- > 0) {
              sinks[i].process(payload);
            }
          }
          catch (MutatedSinkException mse) {
            final Sink newSink = mse.getNewSink();
            newSink.process(payload);
            sinks[i] = newSink;
            replaceOutput(mse.getOldSink(), newSink);
          }
        }
        while (i > 0);

        for (Entry<String, CircularBuffer<Object>> e: afterBeginWindows.entrySet()) {
          final Sink s = outputs.get(e.getKey());
          CircularBuffer<?> cb = e.getValue();
          for (i = cb.size(); i > 0; i--) {
            s.process(cb.get());
          }
        }
        break;

      case END_WINDOW:
        for (Entry<String, CircularBuffer<Object>> e: afterBeginWindows.entrySet()) {
          final Sink s = outputs.get(e.getKey());
          CircularBuffer<?> cb = e.getValue();
          for (i = cb.size(); i > 0; i--) {
            s.process(cb.get());
          }
        }
        for (final Sink s: sinks) {
          s.process(payload);
        }

        ctx.report(producedTupleCount, 0L, ((Tuple)payload).getWindowId());
        producedTupleCount = 0;

        // the default is UNSPECIFIED which we ignore anyways as we ignore everything
        // that we do not understand!
        try {
          switch (ctx.getRequestType()) {
            case BACKUP:
              ctx.backup(this, ((Tuple)payload).getWindowId());
              break;

            case RESTORE:
              logger.info("restore requests are not implemented");
              break;
          }
        }
        catch (Exception e) {
          logger.warn("Exception while catering to external request", e.getLocalizedMessage());
        }

        // i think there should be just one queue instead of one per port - lets defer till we find an example.
        for (Entry<String, CircularBuffer<Tuple>> e: afterEndWindows.entrySet()) {
          final Sink s = outputs.get(e.getKey());
          CircularBuffer<?> cb = e.getValue();
          for (i = cb.size(); i > 0; i--) {
            s.process(cb.get());
          }
        }
        break;

      default:
        for (final Sink s: sinks) {
          s.process(payload);
        }
    }
  }

  @SuppressWarnings("SleepWhileInLoop")
  public void emit(String id, Object payload)
  {
    // once we have annotation done property, we would like to send the tuples to
{'fraction_non_alphanumeric': 0.11343157894736842, 'fraction_numerical': 0.003789473684210526, 'mean_word_length': 2.5177725118483414, 'pattern_counts': {'":': 0, '<': 53, '<?xml version=': 0, '>': 62, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4152660', 'n_tokens_mistral': 3475, 'n_tokens_neox': 3402, 'n_words': 957}