import org.kitesdk.morphline.base.Notifications; //导入依赖的package包/类
@Test
public void testLoadSolrBasic() throws Exception {
  //System.setProperty("ENV_SOLR_HOME", testSolrHome + "/collection1");
  morphline = createMorphline("test-morphlines/loadSolrBasic");    
  //System.clearProperty("ENV_SOLR_HOME");
  Record record = new Record();
  record.put(Fields.ID, "id0");
  record.put("first_name", "Nadja"); // will be sanitized
  startSession();
  Notifications.notifyBeginTransaction(morphline);
  assertTrue(morphline.process(record));
  assertEquals(1, collector.getNumStartEvents());
  Notifications.notifyCommitTransaction(morphline);
  Record expected = new Record();
  expected.put(Fields.ID, "id0");
  assertEquals(Arrays.asList(expected), collector.getRecords());
  assertEquals(1, queryResultSetSize("*:*"));
  Notifications.notifyRollbackTransaction(morphline);
  Notifications.notifyShutdown(morphline);
}
 

import org.kitesdk.morphline.base.Notifications; //导入依赖的package包/类
@Test
public void testTokenizeText() throws Exception {
  morphline = createMorphline("test-morphlines" + File.separator + "tokenizeText");
  for (int i = 0; i < 3; i++) {
    Record record = new Record();
    record.put(Fields.MESSAGE, "Hello World!");
    record.put(Fields.MESSAGE, "\[email protected] #%()123");
    Record expected = record.copy();
    expected.getFields().putAll("tokens", Arrays.asList("hello", "world", "foo", "bar.com", "123"));
    collector.reset();
    startSession();
    Notifications.notifyBeginTransaction(morphline);
    assertTrue(morphline.process(record));
    assertEquals(1, collector.getNumStartEvents());
    Notifications.notifyCommitTransaction(morphline);
    assertEquals(expected, collector.getFirstRecord());
  }
}
 

import org.kitesdk.morphline.base.Notifications; //导入依赖的package包/类
@Override
public void process(Event event) {
  numRecords.mark();
  Timer.Context timerContext = mappingTimer.time();
  try {
    Record record = new Record();
    for (Entry<String, String> entry : event.getHeaders().entrySet()) {
      record.put(entry.getKey(), entry.getValue());
    }
    byte[] bytes = event.getBody();
    if (bytes != null && bytes.length > 0) {
      record.put(Fields.ATTACHMENT_BODY, bytes);
    }    
    try {
      Notifications.notifyStartSession(morphline);
      if (!morphline.process(record)) {
        numFailedRecords.mark();
        LOG.warn("Morphline {} failed to process record: {}", morphlineFileAndId, record);
      }
    } catch (RuntimeException t) {
      numExceptionRecords.mark();
      morphlineContext.getExceptionHandler().handleException(t, record);
    }
  } finally {
    timerContext.stop();
  }
}
 

import org.kitesdk.morphline.base.Notifications; //导入依赖的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.Notifications; //导入依赖的package包/类
public static List<Record> executePipeline(Pipeline pipeline, Record inputRecord) {
  Command morphline = pipeline.getMorphline();

  try {
    LOG.trace("Input Record: {}", inputRecord);

    // Process the Record
    Notifications.notifyStartSession(morphline);
    boolean success = morphline.process(inputRecord);
    Notifications.notifyCommitTransaction(morphline);

    if (!success) {
      throw new MorphlineRuntimeException("Morphline failed to process incoming Record: " + inputRecord);
    }

    // Collect the output
    List<Record> outputRecords = pipeline.getCollector().getRecords();
    if (!outputRecords.iterator().hasNext()) {
      throw new MorphlineRuntimeException("Morphline did not produce output Record(s)");
    }
    LOG.trace("Output Record(s): {}", outputRecords);

    return outputRecords;

  } catch (RuntimeException e) {
    Notifications.notifyRollbackTransaction(morphline);
    // TODO : Review exception handling
    LOG.warn("Morphline failed to execute properly on incoming Record: " + inputRecord, e);
    throw e;
  }
}
 

import org.kitesdk.morphline.base.Notifications; //导入依赖的package包/类
@Override
public void notify(Record notification) {
  for (Object event : Notifications.getLifecycleEvents(notification)) {
    if (event == Notifications.LifecycleEvent.START_SESSION) {
      reset();
    }
  }
}
 

import org.kitesdk.morphline.base.Notifications; //导入依赖的package包/类
@Override
protected void doNotify(Record notification) {
  if (Notifications.containsLifecycleEvent(notification, Notifications.LifecycleEvent.START_SESSION)) {
    recordCounter = 0; // reset
  }
  super.doNotify(notification);
}
 

import org.kitesdk.morphline.base.Notifications; //导入依赖的package包/类
@Test
public void testSimpleCSV() throws Exception {
  morphline = createMorphline("test-morphlines/simpleCSV");

  Notifications.notifyBeginTransaction(morphline);

  InputStream in = new FileInputStream(new File(RESOURCES_DIR + "/test-documents/simpleCSV.txt"));
  Record record = new Record();
  record.put(Fields.ATTACHMENT_BODY, in);
  record.put(Fields.ATTACHMENT_MIME_TYPE, "text/plain");

  // Actually process the input file.
  assertTrue(morphline.process(record));

  assertEquals(collector.getRecords().size(), 2);
  Record rec = collector.getRecords().get(0);

  // Since id and timestamp vary with run, just see if they have anything in them
  assertTrue(rec.get("id").toString().length() > 5);
  assertTrue(rec.get("timestamp").toString().length() > 5);
  assertEquals(rec.get("text").toString(), "[text for body]");

  // Now look at second record
  rec = collector.getRecords().get(1);

  assertTrue(rec.get("id").toString().length() > 5);
  assertTrue(rec.get("timestamp").toString().length() > 5);
  assertEquals(rec.get("text").toString(), "[second record]");

  in.close();
  Notifications.notifyCommitTransaction(morphline);
}
 

import org.kitesdk.morphline.base.Notifications; //导入依赖的package包/类
@Override
public void map(Result result, SolrUpdateWriter solrUpdateWriter) {
    numRecords.mark();
    Timer.Context timerContext = mappingTimer.time();
    try {
        Record record = new Record();
        record.put(Fields.ATTACHMENT_BODY, result);
        record.put(Fields.ATTACHMENT_MIME_TYPE, MorphlineResultToSolrMapper.OUTPUT_MIME_TYPE);
        for (Map.Entry<String, String> entry : forcedRecordFields.entrySet()) {
            record.replaceValues(entry.getKey(), entry.getValue());
        }
        collector.reset(solrUpdateWriter);
        try {
            Notifications.notifyStartSession(morphline);
            if (!morphline.process(record)) {
                numFailedRecords.mark();
                LOG.warn("Morphline {} failed to process record: {}", morphlineFileAndId, record);
            }
        } catch (RuntimeException t) {
            numExceptionRecords.mark();
            morphlineContext.getExceptionHandler().handleException(t, record);
        }
    } finally {
        collector.reset(null);
        timerContext.stop();
    }
}
 

import org.kitesdk.morphline.base.Notifications; //导入依赖的package包/类
@Override
public void beginTransaction() {
  Notifications.notifyBeginTransaction(morphline);      
}
 

import org.kitesdk.morphline.base.Notifications; //导入依赖的package包/类
@Override
public void commitTransaction() {
  Notifications.notifyCommitTransaction(morphline);      
}
 

import org.kitesdk.morphline.base.Notifications; //导入依赖的package包/类
@Override
public void rollbackTransaction() {
  Notifications.notifyRollbackTransaction(morphline);            
}
 

import org.kitesdk.morphline.base.Notifications; //导入依赖的package包/类
@Override
public void stop() {
  Notifications.notifyShutdown(morphline);
}
 

import org.kitesdk.morphline.base.Notifications; //导入依赖的package包/类
@Override
public void close() throws IOException {
  Notifications.notifyShutdown(this.morphline);
}
 

import org.kitesdk.morphline.base.Notifications; //导入依赖的package包/类
@Override
public void execute(Tuple tuple) {
    try {
        finalChild.reset();

        Record record = new Record();
        record.put(org.kitesdk.morphline.base.Fields.ATTACHMENT_BODY, tupleMapper.map(tuple));

        Notifications.notifyStartSession(morphline);
        boolean exceptionRaised = false;
        try {
            boolean processed = morphline.process(record);
            if (!processed) {
                //TODO handle Morphline returned false
                LOG.error("Morphline processing returned false. inputTuple = {}", tuple);
                collector.fail(tuple);
                return;
            }
        } catch (RuntimeException rt) {
            exceptionRaised = true;
            morphlineContext.getExceptionHandler().handleException(rt, record);
        }

        if (terminalBolt) {
            collector.ack(tuple);
            return;
        }

        if (exceptionRaised) {
            //Decide if you need extra handling apart from FaultTolerance handler provided by Morphline
        }

        List<Record> morphlineResults = finalChild.getRecords();
        if (morphlineResults.size() == 0) {
            //TODO handle zero result
            LOG.warn("Zero result by morphline processing. inputTuple: {}", tuple);
            collector.ack(tuple);
            return;
        }
        if (morphlineResults.size() > 1) {
            //TODO Emit to error stream, ignore or fail tuple
            LOG.error("Morphline must not generate more than one output record per input record. returnedSize="
                    + morphlineResults.size());
            collector.fail(tuple);
        }

        Object finalResults = recordMapper.map(morphlineResults.get(0));

        // Useful when expected more than one output from Morphline execution
        /*Object[] finalResults = new Object[recordMappers.size()];
        for (int i = 0; i < morphlineResults.size(); i++) {
            finalResults[i] = recordMappers.get(i).map(morphlineResults.get(i));
        }*/

        if (anchorTuple) {
            collector.emit(tuple, new Values(finalResults));
        } else {
            collector.emit(new Values(finalResults));
        }

        collector.ack(tuple);

    } catch (Exception e) {
        this.collector.reportError(e);
        collector.fail(tuple);
    }
}
 

import org.kitesdk.morphline.base.Notifications; //导入依赖的package包/类
@Override
public void cleanup() {
    if (morphline != null) {
        Notifications.notifyShutdown(morphline);
    }
}
 

import org.kitesdk.morphline.base.Notifications; //导入依赖的package包/类
@Override
protected void commit() throws Exception {
  Notifications.notifyCommitTransaction(morphline);    
  super.commit();
}
 

import org.kitesdk.morphline.base.Notifications; //导入依赖的package包/类
protected void startSession() {
  Notifications.notifyStartSession(morphline);
}
 

import org.kitesdk.morphline.base.Notifications; //导入依赖的package包/类
private boolean load(Record record) {
  Notifications.notifyStartSession(morphline);
  return morphline.process(record);
}
 

import org.kitesdk.morphline.base.Notifications; //导入依赖的package包/类
/**
 * Extract content from the path specified in the value. Key is useless.
 */
public void map(String value, Configuration configuration, Context context) throws IOException {
  LOG.info("Processing file {}", value);
  InputStream in = null;
  Record record = null;
  Timer.Context timerContext = elapsedTime.time();
  try {
    PathParts parts = new PathParts(value.toString(), configuration);
    record = getRecord(parts);
    if (record == null) {
      return; // ignore
    }
    for (Map.Entry<String, String> entry : commandLineMorphlineHeaders.entrySet()) {
      record.replaceValues(entry.getKey(), entry.getValue());
    }
    long fileLength = parts.getFileStatus().getLen();
    if (disableFileOpen) {
      in = new ByteArrayInputStream(new byte[0]);
    } else {
      in = new BufferedInputStream(parts.getFileSystem().open(parts.getUploadPath()));
    }
    record.put(Fields.ATTACHMENT_BODY, in);
    Notifications.notifyStartSession(morphline);
    if (!morphline.process(record)) {
      LOG.warn("Morphline {} failed to process record: {}", morphlineFileAndId, record);
    }
    if (context != null) {
      context.getCounter(MorphlineCounters.class.getName(), MorphlineCounters.FILES_READ.toString()).increment(1);
      context.getCounter(MorphlineCounters.class.getName(), MorphlineCounters.FILE_BYTES_READ.toString()).increment(fileLength);
    }
  } catch (Exception e) {
    LOG.error("Unable to process file " + value, e);
    if (context != null) {
      context.getCounter(getClass().getName() + ".errors", e.getClass().getName()).increment(1);
    }
    morphlineContext.getExceptionHandler().handleException(e, record);
  } finally {
    timerContext.stop();
    if (in != null) {
      in.close();
    }
  }
}