import org.apache.flume.Source; //导入依赖的package包/类
@Override
public void stop() {

  runner.shouldStop.set(true);

  try {
    runnerThread.interrupt();
    runnerThread.join();
  } catch (InterruptedException e) {
    logger.warn("Interrupted while waiting for polling runner to stop. Please report this.", e);
    Thread.currentThread().interrupt();
  }

  Source source = getSource();
  source.stop();
  ChannelProcessor cp = source.getChannelProcessor();
  cp.close();

  lifecycleState = LifecycleState.STOP;
}
 

import org.apache.flume.Source; //导入依赖的package包/类
@Test
public void testDuplicateCreate()  {

  Source avroSource1 = sourceFactory.create("avroSource1", "avro");
  Source avroSource2 = sourceFactory.create("avroSource2", "avro");

  Assert.assertNotNull(avroSource1);
  Assert.assertNotNull(avroSource2);
  Assert.assertNotSame(avroSource1, avroSource2);
  Assert.assertTrue(avroSource1 instanceof AvroSource);
  Assert.assertTrue(avroSource2 instanceof AvroSource);

  Source s1 = sourceFactory.create("avroSource1", "avro");
  Source s2 = sourceFactory.create("avroSource2", "avro");

  Assert.assertNotSame(avroSource1, s1);
  Assert.assertNotSame(avroSource2, s2);

}
 

import org.apache.flume.Source; //导入依赖的package包/类
/**
 * Started the agent. Can only be called after a successful call to
 * configure().
 *
 * @throws FlumeException if a component cannot be started
 * @throws IllegalStateException if the agent has not been configured or is
 * already started
 */
public void start()
    throws FlumeException {
  if (state == State.STARTED) {
    throw new IllegalStateException("Cannot be started while started");
  } else if (state == State.NEW) {
    throw new IllegalStateException("Cannot be started before being " +
        "configured");
  }
  // This check needs to be done before doStart(),
  // as doStart() accesses sourceRunner.getSource()
  Source source = Preconditions.checkNotNull(sourceRunner.getSource(),
      "Source runner returned null source");
  if (source instanceof EmbeddedSource) {
    embeddedSource = (EmbeddedSource)source;
  } else {
    throw new IllegalStateException("Unknown source type: " + source.getClass().getName());
  }
  doStart();
  state = State.STARTED;
}
 

import org.apache.flume.Source; //导入依赖的package包/类
@Override
public Source create(String name, String type) throws FlumeException {
  Preconditions.checkNotNull(name, "name");
  Preconditions.checkNotNull(type, "type");
  logger.info("Creating instance of source {}, type {}", name, type);
  Class<? extends Source> sourceClass = getClass(type);
  try {
    Source source = sourceClass.newInstance();
    source.setName(name);
    return source;
  } catch (Exception ex) {
    throw new FlumeException("Unable to create source: " + name
        + ", type: " + type + ", class: " + sourceClass.getName(), ex);
  }
}
 

import org.apache.flume.Source; //导入依赖的package包/类
@Override
public void start() {
  Source source = getSource();
  ChannelProcessor cp = source.getChannelProcessor();
  cp.initialize();
  source.start();
  lifecycleState = LifecycleState.START;
}
 

import org.apache.flume.Source; //导入依赖的package包/类
@Override
public void stop() {
  Source source = getSource();
  source.stop();
  ChannelProcessor cp = source.getChannelProcessor();
  cp.close();
  lifecycleState = LifecycleState.STOP;
}
 

import org.apache.flume.Source; //导入依赖的package包/类
private void verifySourceCreation(String name, String type,
    Class<?> typeClass) throws Exception {
  Source src = sourceFactory.create(name, type);
  Assert.assertNotNull(src);
  Assert.assertTrue(typeClass.isInstance(src));
}
 

import org.apache.flume.Source; //导入依赖的package包/类
private void verifySourceCreation(String name, String type,
    Class<?> typeClass) throws FlumeException {
  Source src = sourceFactory.create(name, type);
  Assert.assertNotNull(src);
  Assert.assertTrue(typeClass.isInstance(src));
}
 

import org.apache.flume.Source; //导入依赖的package包/类
@After
public void cleanUp() {
  for (Source source : sources) {
    source.stop();
  }
}
 

import org.apache.flume.Source; //导入依赖的package包/类
@SuppressWarnings("unchecked")
public PeriodicHandler(Source source, int emitFreqMS, Class<?> interceptorClass) {
  this.source = source;
  this.emitFreqMS = emitFreqMS;
  this.interceptorClass = (Class<? extends AnalyticInterceptor>) interceptorClass;
}
 

import org.apache.flume.Source; //导入依赖的package包/类
private void verifySourceCreation(String name, String type, Class<?> typeClass) throws FlumeException {
    SourceFactory sourceFactory = new DefaultSourceFactory();
    Source source = sourceFactory.create(name, type);
    Assert.assertNotNull(source);
    Assert.assertTrue(typeClass.isInstance(source));
}