import org.kitesdk.morphline.base.Compiler; //导入依赖的package包/类
@Test (expected = MorphlineCompilationException.class)
public void morphlineCompilationError(
    final @Mocked Compiler compiler
    ) throws Exception {

  new Expectations() {{
    config.getString(MorphlineTranslator.MORPHLINE); result = getResourcePath(MORPHLINE_FILE);
    config.getStringList(MorphlineTranslator.FIELD_NAMES); result = Lists.newArrayList("bar");
    config.getStringList(MorphlineTranslator.FIELD_TYPES); result = Lists.newArrayList("int");

    compiler.compile((File) any, anyString, (MorphlineContext) any, (Command) any); result = new Exception("Compilation exception");
  }};

  stringMorphline.configure(config);
  stringMorphline.translate("The Key", "The Message");
}
 

import org.kitesdk.morphline.base.Compiler; //导入依赖的package包/类
@Override
public void configure(Context context) {
  String morphlineFile = context.getString(MORPHLINE_FILE_PARAM);
  String morphlineId = context.getString(MORPHLINE_ID_PARAM);
  if (morphlineFile == null || morphlineFile.trim().length() == 0) {
    throw new MorphlineCompilationException("Missing parameter: " + MORPHLINE_FILE_PARAM, null);
  }
  morphlineFileAndId = morphlineFile + "@" + morphlineId;
  
  if (morphlineContext == null) {
    FaultTolerance faultTolerance = new FaultTolerance(
        context.getBoolean(FaultTolerance.IS_PRODUCTION_MODE, false), 
        context.getBoolean(FaultTolerance.IS_IGNORING_RECOVERABLE_EXCEPTIONS, false),
        context.getString(FaultTolerance.RECOVERABLE_EXCEPTION_CLASSES));
    
    morphlineContext = new MorphlineContext.Builder()
      .setExceptionHandler(faultTolerance)
      .setMetricRegistry(SharedMetricRegistries.getOrCreate(morphlineFileAndId))
      .build();
  }
  
  Config override = ConfigFactory.parseMap(
      context.getSubProperties(MORPHLINE_VARIABLE_PARAM + "."));
  morphline = new Compiler().compile(
      new File(morphlineFile), morphlineId, morphlineContext, finalChild, override);
  
  this.mappingTimer = morphlineContext.getMetricRegistry().timer(
      MetricRegistry.name("morphline.app", Metrics.ELAPSED_TIME));
  this.numRecords = morphlineContext.getMetricRegistry().meter(
      MetricRegistry.name("morphline.app", Metrics.NUM_RECORDS));
  this.numFailedRecords = morphlineContext.getMetricRegistry().meter(
      MetricRegistry.name("morphline.app", "numFailedRecords"));
  this.numExceptionRecords = morphlineContext.getMetricRegistry().meter(
      MetricRegistry.name("morphline.app", "numExceptionRecords"));
}
 

import org.kitesdk.morphline.base.Compiler; //导入依赖的package包/类
/**
 *
 * @param morphlineFile
 * @param morphlineId
 * @param collector
 * @param isProduction
 * @return
 */
public static Pipeline setPipeline(String morphlineFile, String morphlineId, Collector collector, boolean isProduction) {
  LOG.debug("Constructing Pipeline[{}#{}]", morphlineFile, morphlineId);

  // Set up the Morphline context and handler
  MorphlineContext context = new MorphlineContext.Builder()
      .setExceptionHandler(new FaultTolerance(isProduction, false))
      .build();

  // Compile the Morphline process
  Command morphline;
  try {
    morphline = new Compiler().compile(
        new File(morphlineFile),
        morphlineId,
        context,
        collector);
  } catch (Exception e) {
    throw new MorphlineCompilationException("Morphline compilation error", null, e);
  }

  // Create the pipeline wrapper
  Pipeline pipeline = new Pipeline(morphline, collector);

  // Ensure shutdown notification to Morphline commands esp in streaming environments
  JVMUtils.closeAtShutdown(pipeline);

  // Prep the pipeline
  Notifications.notifyBeginTransaction(pipeline.getMorphline());

  // Register the pipeline into the cache
  if (null == pipelineCache.get()) {
    pipelineCache.set(new HashMap<String, Pipeline>());
  }
  pipelineCache.get().put(morphlineFile + SEPARATOR + morphlineId, pipeline);

  LOG.trace("Pipeline[{}#{}] prepared", morphlineFile, morphlineId);
  return pipeline;
}
 

import org.kitesdk.morphline.base.Compiler; //导入依赖的package包/类
@SuppressWarnings("rawtypes")
@Override
public void prepare(Map stormConf, TopologyContext topologyContext, OutputCollector collector) {
    LOG.info("START prepare");

    this.collector = collector;

    File confFile = loadFile(topologyContext.getThisWorkerPort().toString());

    if (morphlineContext == null) {
        //TODO Make FaultTolerance configurable
        FaultTolerance faultTolerance = new FaultTolerance(true, false, null);

        morphlineContext = new MorphlineContext.Builder()
                .setExceptionHandler(faultTolerance)
                //.setMetricRegistry(SharedMetricRegistries.getOrCreate(customMorphlineId))
                .build();
    }

    Config override = ConfigFactory.parseMap(new HashMap<String, Object>());
    finalChild = new SimpleCommandCollector();
    morphline = new Compiler().compile(confFile, morphlineId, morphlineContext, finalChild, override);

    if (!EmptyUtils.nullOrEmpty(outputFieldNames)) {
        terminalBolt = false;
    }
}
 

import org.kitesdk.morphline.base.Compiler; //导入依赖的package包/类
protected Command parse(String file, String collection) throws IOException {
  SolrLocator locator = new SolrLocator(createMorphlineContext());
  locator.setCollectionName(collection);
  locator.setZkHost(zkServer.getZkAddress());
  //locator.setServerUrl(cloudJettys.get(0).url); // TODO: download IndexSchema from solrUrl not yet implemented
  //locator.setSolrHomeDir(SOLR_HOME_DIR.getPath());
  Config config = new Compiler().parse(new File(RESOURCES_DIR + "/" + file + ".conf"), locator.toConfig("SOLR_LOCATOR"));
  config = config.getConfigList("morphlines").get(0);
  return createMorphline(config);
}
 

import org.kitesdk.morphline.base.Compiler; //导入依赖的package包/类
private Config parse(String file) throws IOException {
  SolrLocator locator = new SolrLocator(createMorphlineContext());
  locator.setSolrHomeDir(testSolrHome + "/collection1");
  File morphlineFile;
  if (new File(file).isAbsolute()) {
    morphlineFile = new File(file + ".conf");
  } else {
    morphlineFile = new File(RESOURCES_DIR + "/" + file + ".conf");
  }
  Config config = new Compiler().parse(morphlineFile, locator.toConfig("SOLR_LOCATOR"));
  config = config.getConfigList("morphlines").get(0);
  return config;
}