import com.datatorrent.api.Context.OperatorContext; //导入依赖的package包/类
@Override
public void setup(OperatorContext context)
{
  OutputPort<?> unifierOutputPort = getOutputPort();
  unifierOutputPort.setSink(
      new Sink<Object>()
      {
        @Override
        public void put(Object tuple)
        {
          outputPort.emit(tuple);
        }

        @Override
        public int getCount(boolean reset)
        {
          return 0;
        }
      }
  );
  unifier.setup(context);
  spinMillis = context.getValue(OperatorContext.SPIN_MILLIS);
}
 

import com.datatorrent.api.Context.OperatorContext; //导入依赖的package包/类
public static URI uriHelper(OperatorContext context, URI uri)
{
  if (uri == null) {
    if (context.getValue(DAG.GATEWAY_CONNECT_ADDRESS) == null) {
      throw new IllegalArgumentException("The uri property is not set and the dt.attr.GATEWAY_CONNECT_ADDRESS is not defined");
    }

    try {
      uri = PubSubHelper.getURI(context);
    } catch (Exception ex) {
      throw new RuntimeException(ex);
    }
  }

  return uri;
}
 

import com.datatorrent.api.Context.OperatorContext; //导入依赖的package包/类
private void purgeCheckpoints()
{
  for (Pair<PTOperator, Long> p : purgeCheckpoints) {
    final PTOperator operator = p.getFirst();
    if (!operator.isOperatorStateLess()) {
      final long windowId = p.getSecond();
      Runnable r = new Runnable()
      {
        @Override
        public void run()
        {
          try {
            operator.getOperatorMeta().getValue(OperatorContext.STORAGE_AGENT).delete(operator.getId(), windowId);
          } catch (IOException ex) {
            LOG.error("Failed to purge checkpoint for operator {} for windowId {}", operator, windowId, ex);
          }
        }
      };
      poolExecutor.submit(r);
    }
  }
  purgeCheckpoints.clear();
}
 

import com.datatorrent.api.Context.OperatorContext; //导入依赖的package包/类
@Override
protected void starting(Description description)
{
  super.starting(description);
  TestUtils.deleteTargetTestClassFolder(description);
  try {
    FileUtils.forceMkdir(new File(getDir()));
    Attribute.AttributeMap.DefaultAttributeMap attributeMap = new Attribute.AttributeMap.DefaultAttributeMap();
    attributeMap.put(Context.DAGContext.CHECKPOINT_WINDOW_COUNT, 60);
    attributeMap.put(Context.OperatorContext.SPIN_MILLIS, 500);

    testOperatorContext = mockOperatorContext(0, attributeMap);
  } catch (IOException e) {
    throw new RuntimeException(e);
  }
}
 

import com.datatorrent.api.Context.OperatorContext; //导入依赖的package包/类
public void testLinearOperatorRecovery() throws Exception
{
  RecoverableInputOperator.initGenTuples();
  CollectorOperator.collection.clear();
  CollectorOperator.duplicates.clear();

  dag.getAttributes().put(LogicalPlan.CHECKPOINT_WINDOW_COUNT, 2);
  dag.getAttributes().put(LogicalPlan.STREAMING_WINDOW_SIZE_MILLIS, 300);
  dag.getAttributes().put(LogicalPlan.CONTAINERS_MAX_COUNT, 1);
  RecoverableInputOperator rip = dag.addOperator("LongGenerator", RecoverableInputOperator.class);
  rip.setMaximumTuples(maxTuples);
  rip.setSimulateFailure(true);

  CollectorOperator cm = dag.addOperator("LongCollector", CollectorOperator.class);
  cm.setSimulateFailure(true);
  dag.getMeta(cm).getAttributes().put(OperatorContext.PROCESSING_MODE, processingMode);

  dag.addStream("connection", rip.output, cm.input);

  StramLocalCluster lc = new StramLocalCluster(dag);
  lc.run();
}
 

import com.datatorrent.api.Context.OperatorContext; //导入依赖的package包/类
@Test
public void testJdbcInputOperator()
{
  JdbcStore store = new JdbcStore();
  store.setDatabaseDriver(DB_DRIVER);
  store.setDatabaseUrl(URL);

  com.datatorrent.api.Attribute.AttributeMap.DefaultAttributeMap attributeMap = new com.datatorrent.api.Attribute.AttributeMap.DefaultAttributeMap();
  attributeMap.put(DAG.APPLICATION_ID, APP_ID);
  OperatorContext context = mockOperatorContext(OPERATOR_ID, attributeMap);

  TestInputOperator inputOperator = new TestInputOperator();
  inputOperator.setStore(store);
  insertEventsInTable(10);

  CollectorTestSink<Object> sink = new CollectorTestSink<>();
  inputOperator.outputPort.setSink(sink);

  inputOperator.setup(context);
  inputOperator.beginWindow(0);
  inputOperator.emitTuples();
  inputOperator.endWindow();

  Assert.assertEquals("rows from db", 10, sink.collectedTuples.size());
}
 

import com.datatorrent.api.Context.OperatorContext; //导入依赖的package包/类
@Test
public void TestAerospikeInputOperator()
{
  AerospikeStore store = getStore();
  OperatorContext context = getOperatorContext(APP_ID);
  TestInputOperator inputOperator = new TestInputOperator();

  inputOperator.setStore(store);
  inputOperator.insertEventsInTable(NUM_TUPLES);

  CollectorTestSink<Object> sink = new CollectorTestSink<Object>();
  inputOperator.outputPort.setSink(sink);

  inputOperator.setup(context);
  inputOperator.beginWindow(0);
  inputOperator.emitTuples();
  inputOperator.endWindow();

  Assert.assertEquals("rows from db", NUM_TUPLES, sink.collectedTuples.size());
}
 

import com.datatorrent.api.Context.OperatorContext; //导入依赖的package包/类
@Override
public void activate(OperatorContext ctx)
{
  dataGeneratorThread = new Thread("String Generator")
  {
    @Override
    public void run()
    {
      try {
        int i = 0;
        while (dataGeneratorThread != null && i < maxTuple) {
          stringBuffer.put((++i) + "###testString " + i);
        }
        stringBuffer.put((maxTuple + 1) + "###" + KafkaOperatorTestBase.END_TUPLE);
      } catch (InterruptedException ie) {
        //
      }
    }
  };
  dataGeneratorThread.start();
}
 

import com.datatorrent.api.Context.OperatorContext; //导入依赖的package包/类
@Test
public void testValidInputOperatorDeployInfoType()
{
  TestGeneratorInputOperator.ValidInputOperator o1 = dag.addOperator("o1", TestGeneratorInputOperator.ValidInputOperator.class);
  GenericTestOperator o2 = dag.addOperator("o2", GenericTestOperator.class);

  dag.addStream("stream1", o1.outport, o2.inport1);

  dag.setAttribute(OperatorContext.STORAGE_AGENT, new MemoryStorageAgent());
  StreamingContainerManager scm = new StreamingContainerManager(dag);

  PhysicalPlan physicalPlan = scm.getPhysicalPlan();
  List<PTContainer> containers = physicalPlan.getContainers();
  for (int i = 0; i < containers.size(); ++i) {
    assignContainer(scm, "container" + (i + 1));
  }
  OperatorMeta o1Meta = dag.getMeta(o1);
  PTOperator o1Physical = physicalPlan.getOperators(o1Meta).get(0);

  String containerId = o1Physical.getContainer().getExternalId();

  OperatorDeployInfo o1DeployInfo = getDeployInfo(scm.getContainerAgent(containerId)).get(0);
  Assert.assertEquals("type " + o1DeployInfo, OperatorDeployInfo.OperatorType.INPUT, o1DeployInfo.type);
}
 

import com.datatorrent.api.Context.OperatorContext; //导入依赖的package包/类
@Override
public void setup(OperatorContext context)
{
  try {
    fs = getHDFSInstance();
  } catch (IOException ex) {
    throw new RuntimeException(ex);
  }

  this.context = context;
  lastTimeStamp = System.currentTimeMillis();

  fileCounters.setCounter(Counters.TOTAL_BYTES_WRITTEN, new MutableLong());
  fileCounters.setCounter(Counters.TOTAL_TIME_ELAPSED, new MutableLong());
  super.setup(context);
}
 

import com.datatorrent.api.Context.OperatorContext; //导入依赖的package包/类
/**
 * {@inheritDoc}
 * Initialize various map and list fields
 */
@Override
public void setup(OperatorContext context)
{
  if (null == wordMapFile) {
    wordMapFile = new HashMap<>();
  }
  if (null == wordMapGlobal) {
    wordMapGlobal = new HashMap<>();
  }
  resultPerFile = new ArrayList(1);
  resultGlobal = new ArrayList(1);
  // singleton map {<fileName> => fileFinalList}; cannot populate it yet since we need fileName
  resultFileFinal = new HashMap<>(1);
  fileFinalList = new ArrayList<>();
}
 

import com.datatorrent.api.Context.OperatorContext; //导入依赖的package包/类
@Test
public void testNumberOfUnifiersWithEvenPartitions()
{
  LogicalPlan dag = new LogicalPlan();
  dag.setAttribute(OperatorContext.STORAGE_AGENT, new StramTestSupport.MemoryStorageAgent());
  GenericTestOperator node1 = dag.addOperator("node1", GenericTestOperator.class);
  GenericTestOperator node2 = dag.addOperator("node2", GenericTestOperator.class);
  dag.addStream("node1.outport1", node1.outport1, node2.inport1);
  dag.setOperatorAttribute(node1, OperatorContext.PARTITIONER, new StatelessPartitioner<GenericTestOperator>(8));
  dag.setOutputPortAttribute(node1.outport1, PortContext.UNIFIER_LIMIT, 4);
  PhysicalPlan plan = new PhysicalPlan(dag, new TestPlanContext());
  List<PTContainer> containers = plan.getContainers();
  int unifierCount = 0;
  int totalOperators = 0;
  for (PTContainer container : containers) {
    List<PTOperator> operators = container.getOperators();
    for (PTOperator operator : operators) {
      totalOperators++;
      if (operator.isUnifier()) {
        unifierCount++;
      }
    }
  }
  Assert.assertEquals("Number of operators", 12, totalOperators);
  Assert.assertEquals("Number of unifiers", 3, unifierCount);
}
 

import com.datatorrent.api.Context.OperatorContext; //导入依赖的package包/类
@Override
public void populateDAG(DAG dag, Configuration conf)
{
  String filePath = "HDFSOutputOperatorBenchmarkingApp/"
      + System.currentTimeMillis();

  dag.setAttribute(DAG.STREAMING_WINDOW_SIZE_MILLIS, 1000);

  RandomWordGenerator wordGenerator = dag.addOperator("wordGenerator", RandomWordGenerator.class);

  dag.getOperatorMeta("wordGenerator").getMeta(wordGenerator.output)
      .getAttributes().put(PortContext.QUEUE_CAPACITY, 10000);
  dag.getOperatorMeta("wordGenerator").getAttributes()
      .put(OperatorContext.APPLICATION_WINDOW_COUNT, 1);

  FSByteOutputOperator hdfsOutputOperator = dag.addOperator("hdfsOutputOperator", new FSByteOutputOperator());
  hdfsOutputOperator.setFilePath(filePath);
  dag.getOperatorMeta("hdfsOutputOperator").getAttributes()
      .put(OperatorContext.COUNTERS_AGGREGATOR, new BasicCounters.LongAggregator<MutableLong>());

  dag.addStream("Generator2HDFSOutput", wordGenerator.output, hdfsOutputOperator.input);
}
 

import com.datatorrent.api.Context.OperatorContext; //导入依赖的package包/类
@Override
public void setup(OperatorContext context) {
  this.traceTuples =
      ApexStreamTuple.Logging.isDebugEnabled(
          pipelineOptions.get().as(ApexPipelineOptions.class), this);
  try {
    reader = source.createReader(this.pipelineOptions.get(), null);
    available = reader.start();
  } catch (IOException e) {
    throw new RuntimeException(e);
  }
}
 

import com.datatorrent.api.Context.OperatorContext; //导入依赖的package包/类
@Override
public void setup(OperatorContext ctx)
{
  context = ZMQ.context(1);
  subscriber = context.socket(ZMQ.SUB);
  subscriber.connect(url);
  subscriber.subscribe(filter.getBytes());
  syncclient = context.socket(ZMQ.REQ);
  syncclient.connect(syncUrl);
  syncclient.send("".getBytes(), 0);
}
 

import com.datatorrent.api.Context.OperatorContext; //导入依赖的package包/类
@Override
public void setup(OperatorContext context)
{
  sc = new ScriptingContainer(LocalVariableBehavior.PERSISTENT);
  for (String s : setupScripts) {
    EvalUnit unit = sc.parse(s);
    unit.run();
  }
  if (type == Type.EVAL) {
    unit = sc.parse(script);
  }
}
 

import com.datatorrent.api.Context.OperatorContext; //导入依赖的package包/类
@Override
public void setup(OperatorContext context)
{
  super.setup(context);
  try {
    queryStatement = store.getConnection().createStatement();
  } catch (SQLException e) {
    throw new RuntimeException("creating query", e);
  }
}
 

import com.datatorrent.api.Context.OperatorContext; //导入依赖的package包/类
/**
 * Read available checkpoints from storage agent for all operators.
 * @param startTime
 * @param currentTime
 * @throws IOException
 */
public void syncCheckpoints(long startTime, long currentTime) throws IOException
{
  for (PTOperator oper : getAllOperators().values()) {
    StorageAgent sa = oper.operatorMeta.getValue(OperatorContext.STORAGE_AGENT);
    long[] windowIds = sa.getWindowIds(oper.getId());
    Arrays.sort(windowIds);
    oper.checkpoints.clear();
    for (long wid : windowIds) {
      if (wid != Stateless.WINDOW_ID) {
        oper.addCheckpoint(wid, startTime);
      }
    }
  }
}
 

import com.datatorrent.api.Context.OperatorContext; //导入依赖的package包/类
public boolean isOperatorStateLess()
{
  if (operatorMeta.getDAG().getValue(OperatorContext.STATELESS) || operatorMeta.getValue(OperatorContext.STATELESS)) {
    return true;
  }

  return operatorMeta.getOperator().getClass().isAnnotationPresent(Stateless.class);
}
 

import com.datatorrent.api.Context.OperatorContext; //导入依赖的package包/类
private void cleanUp(Context.OperatorContext operatorContext)
{
  FSWindowDataManager windowDataManager = new FSWindowDataManager();
  windowDataManager.setup(operatorContext);
  try {
    windowDataManager.committed(windowDataManager.getLargestCompletedWindow());
  } catch (IOException e) {
    e.printStackTrace();
  }
}