import com.datatorrent.api.InputOperator; //导入依赖的package包/类
public Set<String> getAllDTInstantiableGenericOperators()
{
  TypeGraphVertex tgv = typeGraph.get(GenericOperator.class.getName());
  if (tgv == null) {
    return null;
  }
  Set<String> result = new TreeSet<>();
  for (TypeGraphVertex node : tgv.allInstantiableDescendants) {
    if ((isAncestor(InputOperator.class.getName(), node.typeName) || isAncestor(Module.class.getName(), node.typeName)
        || !getAllInputPorts(node).isEmpty())) {
      result.add(node.typeName);
    }
  }

  return result;
}
 

import com.datatorrent.api.InputOperator; //导入依赖的package包/类
@SuppressWarnings("unchecked")
public static Node<?> retrieveNode(Object operator, OperatorContext context, OperatorDeployInfo.OperatorType type)
{
  logger.debug("type={}, operator class={}", type, operator.getClass());

  Node<?> node;
  if (operator instanceof InputOperator && type == OperatorDeployInfo.OperatorType.INPUT) {
    node = new InputNode((InputOperator)operator, context);
  } else if (operator instanceof Unifier && type == OperatorDeployInfo.OperatorType.UNIFIER) {
    node = new UnifierNode((Unifier<Object>)operator, context);
  } else if (type == OperatorDeployInfo.OperatorType.OIO) {
    node = new OiONode((Operator)operator, context);
  } else {
    node = new GenericNode((Operator)operator, context);
  }

  return node;
}
 

import com.datatorrent.api.InputOperator; //导入依赖的package包/类
@Override
public void emitTuples()
{
  if (operator instanceof InputOperator) {
    ((InputOperator)operator).emitTuples();
  }
}
 

import com.datatorrent.api.InputOperator; //导入依赖的package包/类
public InputNode(InputOperator operator, OperatorContext context)
{
  super(operator, context);
}
 

import com.datatorrent.api.InputOperator; //导入依赖的package包/类
/**
 * Create a stream with any input operator
 * @param operator
 * @param outputPort
 * @param opts
 * @param <T>
 * @return
 */
public static <T> ApexStream<T> fromInput(InputOperator operator, Operator.OutputPort<T> outputPort, Option... opts)
{
  ApexStreamImpl<T> newStream = new ApexStreamImpl<>();
  return newStream.addOperator(operator, null, outputPort, opts);
}