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

import com.malhartech.api.ActivationListener;
import com.malhartech.api.Operator;
import com.malhartech.api.Operator.OutputPort;
import com.malhartech.api.Sink;
import com.malhartech.engine.Operators.PortMappingDescriptor;
import java.lang.reflect.Array;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map.Entry;
import java.util.concurrent.BlockingQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * @param <OPERATOR>
 * @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<Object>> outputs = new HashMap<String, CounterSink<Object>>();
  @SuppressWarnings(value = "VolatileArrayField")
  protected volatile CounterSink<Object>[] sinks = CounterSink.NO_SINKS;
  protected final int spinMillis = 10;
  protected final int bufferCapacity = 1024 * 1024;
  protected boolean alive;
  protected final OPERATOR operator;
  protected final PortMappingDescriptor descriptor;
  protected long currentWindowId;
  protected int applicationWindowCount;

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

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

  public Operator getOperator()
  {
    return operator;
  }

  public void connectOutputPort(String port, final Sink<Object> sink)
  {
    @SuppressWarnings("unchecked")
    OutputPort<Object> outputPort = (OutputPort<Object>)descriptor.outputPorts.get(port);
    if (outputPort != null) {
      if (sink instanceof CounterSink) {
        outputPort.setSink(sink);
        outputs.put(port, (CounterSink<Object>)sink);
      }
      else if (sink == null) {
        outputPort.setSink(null);
        outputs.remove(port);
      }
      else {
        /*
         * if streams implemented CounterSink, this would not get called.
         */
        CounterSink<Object> cs = new CounterSink<Object>()
        {
          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<Object> connectInputPort(String port, final Sink<? extends Object> sink);

  OperatorContext context;

  @SuppressWarnings("unchecked")
  public void activate(OperatorContext context)
  {
    boolean activationListener = operator instanceof ActivationListener;

    activateSinks();
    this.alive = true;
    this.context = context;
    this.applicationWindowCount = context.getAttributes().attr(OperatorContext.APPLICATION_WINDOW_COUNT).setIfAbsent(1);<commit_msg>Use alternate api to get the default value for window if ApplicationWindowCount attribute is not set.<commit_after>/*
 *  Copyright (c) 2012 Malhar, Inc.
 *  All Rights Reserved.
 */
package com.malhartech.engine;

import com.malhartech.api.ActivationListener;
import com.malhartech.api.Operator;
import com.malhartech.api.Operator.OutputPort;
import com.malhartech.api.Sink;
import com.malhartech.engine.Operators.PortMappingDescriptor;
import java.lang.reflect.Array;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Map.Entry;
import java.util.concurrent.BlockingQueue;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 *
 * @param <OPERATOR>
 * @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<Object>> outputs = new HashMap<String, CounterSink<Object>>();
  @SuppressWarnings(value = "VolatileArrayField")
  protected volatile CounterSink<Object>[] sinks = CounterSink.NO_SINKS;
  protected final int spinMillis = 10;
  protected final int bufferCapacity = 1024 * 1024;
  protected boolean alive;
  protected final OPERATOR operator;
  protected final PortMappingDescriptor descriptor;
  protected long currentWindowId;
  protected int applicationWindowCount;

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

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

  public Operator getOperator()
  {
    return operator;
  }

  public void connectOutputPort(String port, final Sink<Object> sink)
  {
    @SuppressWarnings("unchecked")
    OutputPort<Object> outputPort = (OutputPort<Object>)descriptor.outputPorts.get(port);
    if (outputPort != null) {
      if (sink instanceof CounterSink) {
        outputPort.setSink(sink);
        outputs.put(port, (CounterSink<Object>)sink);
      }
      else if (sink == null) {
        outputPort.setSink(null);
        outputs.remove(port);
      }
      else {
        /*
         * if streams implemented CounterSink, this would not get called.
         */
        CounterSink<Object> cs = new CounterSink<Object>()
        {
          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<Object> connectInputPort(String port, final Sink<? extends Object> sink);

  OperatorContext context;

  @SuppressWarnings("unchecked")
  public void activate(OperatorContext context)
  {
    boolean activationListener = operator instanceof ActivationListener;

    activateSinks();
    this.alive = true;
    this.context = context;
    this.applicationWindowCount = context.getAttributes().attrValue(OperatorContext.APPLICATION_WINDOW_COUNT, 1);
{'fraction_non_alphanumeric': 0.09419980048453755, 'fraction_numerical': 0.005700441784238278, 'mean_word_length': 2.9943084803642575, 'pattern_counts': {'":': 0, '<': 39, '<?xml version=': 0, '>': 39, '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': '5211996', 'n_tokens_mistral': 1991, 'n_tokens_neox': 1943, 'n_words': 529}