import org.apache.nifi.processor.io.StreamCallback; //导入依赖的package包/类
@Override
public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
	FlowFile flowFile = session.get();
	if (flowFile == null) {
		flowFile = session.create();
	}
	try {
		flowFile.getAttributes();
		flowFile = session.putAttribute(flowFile, "mime.type", "application/json");
		flowFile = session.write(flowFile, new StreamCallback() {
			@Override
			public void process(InputStream inputStream, OutputStream outputStream) throws IOException {
				Tika tika = new Tika();
				String text = "";
				try {
					text = tika.parseToString(inputStream);
				} catch (TikaException e) {
					getLogger().error("Apache Tika failed to parse input " + e.getLocalizedMessage());
					e.printStackTrace();
				}
				// TODO: wrap in JSON???
				outputStream.write(text.getBytes());
			}
		});
		session.transfer(flowFile, REL_SUCCESS);
		session.commit();
	} catch (final Throwable t) {
		getLogger().error("Unable to process ExtractTextProcessor file " + t.getLocalizedMessage());
		getLogger().error("{} failed to process due to {}; rolling back session", new Object[] { this, t });
		throw t;
	}
}