import com.datatorrent.common.util.BaseOperator; //导入依赖的package包/类
@Override
public void emitTuples()
{
  if (testTuples == null || testTuples.isEmpty()) {
    if (blockEndStream) {
      return;
    }
    BaseOperator.shutdown();
  }

  if (first) {
    List<T> tuples = testTuples.remove(0);
    for (T t: tuples) {
      output.emit(t);
      LOG.debug("sent tuple ==={}===", t);
    }
    first = false;
  }
}
 

import com.datatorrent.common.util.BaseOperator; //导入依赖的package包/类
@Test
public void testArraySerialization() throws Exception
{
  OperatorDiscoverer od = new OperatorDiscoverer();
  Assert.assertNotNull(od.getOperatorClass(BaseOperator.class.getName()));
  JSONObject desc = od.describeClass(ArraysHolder.class);
  String debugInfo = "\ntype info for " + ArraysHolder.class + ":\n" + desc.toString(2) + "\n";

  JSONArray props = desc.getJSONArray("properties");
  ArraysHolder ah = new ArraysHolder();

  JSONObject beanArray = getJSONProperty(props, "beanArray");
  Assert.assertEquals(debugInfo + "type " + ah.beanArray.getClass(), ah.beanArray.getClass().getName(), beanArray.get("type"));

  JSONObject intArray = getJSONProperty(props, "intArray");
  Assert.assertEquals(debugInfo + "type " + ah.intArray.getClass(), ah.intArray.getClass().getName(), intArray.get("type"));

  ObjectMapper mapper = ObjectMapperFactory.getOperatorValueSerializer();
  String s = mapper.writeValueAsString(ah);

  ArraysHolder clone = mapper.readValue(s, ArraysHolder.class);
  Assert.assertNotNull(clone.intArray);
  Assert.assertArrayEquals(ah.intArray, clone.intArray);

}
 

import com.datatorrent.common.util.BaseOperator; //导入依赖的package包/类
@Override
public void emitTuples()
{
  int lstart = s_start;
  int lend = s_end;

  if (lstart < lend) {
    for (int i = lstart; i < lend; i++) {
      emitTuple(i);
    }
  } else {
    for (int i = lstart; i > lend; i--) {
      emitTuple(i);
    }
  }
  // done generating data
  LOG.info("Finished generating data.");
  BaseOperator.shutdown();
}
 

import com.datatorrent.common.util.BaseOperator; //导入依赖的package包/类
@Override
public void endWindow() {
  if (outputWatermark >= BoundedWindow.TIMESTAMP_MAX_VALUE.getMillis()) {
    // terminate the stream
    if (traceTuples) {
      LOG.debug("terminating input after final watermark");
    }
    try {
      // see BEAM-1140 for why the delay after mark was emitted
      Thread.sleep(100);
    } catch (InterruptedException e) {
    }
    BaseOperator.shutdown();
  }
}
 

import com.datatorrent.common.util.BaseOperator; //导入依赖的package包/类
@Override
public void beginWindow(long windowId)
{
  if (++windowCount == 3) {
    BaseOperator.shutdown();
  }
}
 

import com.datatorrent.common.util.BaseOperator; //导入依赖的package包/类
@Override
public void committed(long windowId)
{
  logger.debug("{}, windowId={}", this, Codec.getStringWindowId(windowId));
  if (simulateFailure && firstRun && checkpointedWindowId > 0 && windowId > checkpointedWindowId) {
    throw new RuntimeException("Failure Simulation from " + this);
  }

  // we have emitted enough tuples and we have tested recovery, so we can shutdown once we have emitted enough.
  if (maximumTuples == 0) {
    BaseOperator.shutdown();
  }
}
 

import com.datatorrent.common.util.BaseOperator; //导入依赖的package包/类
@Override
public void beginWindow(long windowId)
{
  if (shutdown) {
    BaseOperator.shutdown();
  }
  this.windowId = windowId;
}
 

import com.datatorrent.common.util.BaseOperator; //导入依赖的package包/类
@Override
public void emitTuples()
{
  Object tuple;
  while ((tuple = this.externallyAddedTuples.poll()) != null) {
    outport.emit(tuple);
  }

  if (remainingSleepTime > 0) {
    try {
      Thread.sleep(spinMillis);
      remainingSleepTime -= spinMillis;
    } catch (InterruptedException ie) {
      throw new RuntimeException(ie);
    }
  } else if (maxTuples != 0) {
    generatedTuples++;
    LOG.debug("sending tuple " + generatedTuples);
    outport.emit(String.valueOf(generatedTuples));
    if (maxTuples > 0 && maxTuples <= generatedTuples) {
      BaseOperator.shutdown();
      throw new RuntimeException(new InterruptedException("done emitting all."));
    }
    remainingSleepTime = emitInterval;
  } else {
    remainingSleepTime = emitInterval;
  }
}
 

import com.datatorrent.common.util.BaseOperator; //导入依赖的package包/类
@SuppressWarnings("SameParameterValue")
FailableOperator(BaseOperator baseOperator, int windows, boolean afterCommit)
{
  failureSimulated = false;
  this.baseOperator = baseOperator.getClass().getSimpleName();
  this.simulateFailureWindows = windows;
  this.simulateFailureAfterCommit = afterCommit;
}
 

import com.datatorrent.common.util.BaseOperator; //导入依赖的package包/类
@Override
public void endWindow()
{
  assert (threadId == Thread.currentThread().getId());
  BaseOperator.shutdown();
}
 

import com.datatorrent.common.util.BaseOperator; //导入依赖的package包/类
@Override
public void emitTuples()
{
  BaseOperator.shutdown();
}
 

import com.datatorrent.common.util.BaseOperator; //导入依赖的package包/类
@Test(expected = IllegalArgumentException.class)
public void testNullOperatorName()
{
  dag.addOperator(null, BaseOperator.class);
}
 

import com.datatorrent.common.util.BaseOperator; //导入依赖的package包/类
@Test(expected = IllegalArgumentException.class)
public void testEmptyOperatorName()
{
  dag.addOperator("", BaseOperator.class);
}