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

import com.malhartech.api.ActivationListener;
import com.malhartech.api.Operator;
import com.malhartech.api.Operator.InputPort;
import com.malhartech.api.Operator.OutputPort;
import com.malhartech.api.Operator.Port;
import com.malhartech.api.Sink;
import com.malhartech.api.Stats;
import com.malhartech.api.Stats.StatsReporter;
import com.malhartech.dag.Operators.PortMappingDescriptor;
import com.malhartech.util.CircularBuffer;
import java.util.*;
import java.util.Map.Entry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * @author <NAME> <<EMAIL>>
 */
public abstract class Node<OPERATOR extends Operator> implements Runnable
{
  private static final Logger logger = LoggerFactory.getLogger(Node.class);
  /*
   * if the Component is capable of taking only 1 input, call it INPUT.
   * if the Component is capable of providing only 1 output, call it OUTPUT.
   */
  public static final String INPUT = "input";
  public static final String OUTPUT = "output";
  public final String id;
  protected final HashMap<String, CounterSink<?>> outputs = new HashMap<String, CounterSink<?>>();
  protected final int spinMillis = 10;
  protected final int bufferCapacity = 1024 * 1024;
  @SuppressWarnings(value = "VolatileArrayField")
  protected volatile CounterSink[] sinks = CounterSink.NO_SINKS;
  protected boolean alive;
  protected final OPERATOR operator;
  protected final PortMappingDescriptor descriptor;
  protected long currentWindowId;

  public Node(String id, OPERATOR operator)
  {
    this.id = id;
    this.operator = operator;

    descriptor = new PortMappingDescriptor();
    Operators.describe(operator, descriptor);
  }

  public Operator getOperator()
  {
    return operator;
  }

  protected void connectOutputPort(String port, final Sink sink)
  {
    OutputPort outputPort = descriptor.outputPorts.get(port);
    if (outputPort != null) {
      outputPort.setSink(sink);

      if (sink instanceof CounterSink) {
        outputs.put(port, (CounterSink<?>)sink);
      }
      else if (sink == null) {
        outputs.remove(port);
      }
      else {
        /*
         * if streams implemented CounterSink, this would not get called.
         */
        outputs.put(port, new CounterSink()
        {
          int count;

          @Override
          public void process(Object tuple)
          {
            count++;
            sink.process(tuple);
          }

          @Override
          public int getCount()
          {
            return count;
          }

          @Override
          public int resetCount()
          {
            int ret = count;
            count = 0;
            return ret;
          }
        });
      }
    }
  }

  public abstract Sink connect(String id, Sink sink);

  protected void activateSinks()
  {<commit_msg>Make the performance test work by tapping into the Sinks registered with the output ports.<commit_after>/*
 *  Copyright (c) 2012 Malhar, Inc.
 *  All Rights Reserved.
 */
package com.malhartech.dag;

import com.malhartech.api.ActivationListener;
import com.malhartech.api.Operator;
import com.malhartech.api.Operator.InputPort;
import com.malhartech.api.Operator.OutputPort;
import com.malhartech.api.Operator.Port;
import com.malhartech.api.Sink;
import com.malhartech.api.Stats;
import com.malhartech.api.Stats.StatsReporter;
import com.malhartech.dag.Operators.PortMappingDescriptor;
import com.malhartech.util.CircularBuffer;
import java.util.*;
import java.util.Map.Entry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * @author <NAME> <<EMAIL>>
 */
public abstract class Node<OPERATOR extends Operator> implements Runnable
{
  private static final Logger logger = LoggerFactory.getLogger(Node.class);
  /*
   * if the Component is capable of taking only 1 input, call it INPUT.
   * if the Component is capable of providing only 1 output, call it OUTPUT.
   */
  public static final String INPUT = "input";
  public static final String OUTPUT = "output";
  public final String id;
  protected final HashMap<String, CounterSink<?>> outputs = new HashMap<String, CounterSink<?>>();
  protected final int spinMillis = 10;
  protected final int bufferCapacity = 1024 * 1024;
  @SuppressWarnings(value = "VolatileArrayField")
  protected volatile CounterSink[] sinks = CounterSink.NO_SINKS;
  protected boolean alive;
  protected final OPERATOR operator;
  protected final PortMappingDescriptor descriptor;
  protected long currentWindowId;

  public Node(String id, OPERATOR operator)
  {
    this.id = id;
    this.operator = operator;

    descriptor = new PortMappingDescriptor();
    Operators.describe(operator, descriptor);
  }

  public Operator getOperator()
  {
    return operator;
  }

  protected void connectOutputPort(String port, final Sink sink)
  {
    OutputPort outputPort = descriptor.outputPorts.get(port);
    if (outputPort != null) {
      if (sink instanceof CounterSink) {
        outputPort.setSink(sink);
        outputs.put(port, (CounterSink<?>)sink);
      }
      else if (sink == null) {
        outputPort.setSink(null);
        outputs.remove(port);
      }
      else {
        /*
         * if streams implemented CounterSink, this would not get called.
         */
        CounterSink cs =  new CounterSink()
        {
          int count;

          @Override
          public void process(Object tuple)
          {
            count++;
            sink.process(tuple);
          }

          @Override
          public int getCount()
          {
            return count;
          }

          @Override
          public int resetCount()
          {
            int ret = count;
            count = 0;
            return ret;
          }
        };
        outputPort.setSink(cs);
        outputs.put(port, cs);
      }
    }
  }

  public abstract Sink connect(String id, Sink sink);

  protected void activateSinks()
  {
{'fraction_non_alphanumeric': 0.09453151050350117, 'fraction_numerical': 0.006335445148382794, 'mean_word_length': 2.726086956521739, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 21, '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': '5706824', 'n_tokens_mistral': 1748, 'n_tokens_neox': 1699, 'n_words': 481}