import org.apache.nifi.util.StopWatch; //导入依赖的package包/类
@Override
public FlowFile enrich(ProcessSession session, FlowFile ff, Map<String, String> content) {
    StopWatch stopWatch = new StopWatch(true);
    ff = session.write(ff, new StreamingJSONAppender(this.field, makeJsonFromMapOfJsons(content)));
    session.getProvenanceReporter().modifyContent(ff, stopWatch.getElapsed(TimeUnit.MILLISECONDS));
    return ff;
}
 

import org.apache.nifi.util.StopWatch; //导入依赖的package包/类
CheckedStopWatch(long timeframeMilliseconds) {
    this.stopWatch = new StopWatch(false);
    this.timeFrameMilliseconds = timeframeMilliseconds;
}
 

import org.apache.nifi.util.StopWatch; //导入依赖的package包/类
@Override
public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
    FlowFile flowFile = session.get();
    if ( flowFile == null ) {
        return;
    }

    final String outboundAddress = context.getProperty(OUTBOUND_ADDRESS).evaluateAttributeExpressions(flowFile).getValue();

    try {
        final StopWatch stopWatch = new StopWatch(true);

        final byte[] buffer = new byte[(int) flowFile.getSize()];

        session.read(flowFile, new InputStreamCallback() {
            @Override
            public void process(final InputStream in) throws IOException {
                StreamUtils.fillBuffer(in, buffer, false);
            }
        });

        // TODO detect flowFile MIME_TYPE and convert buffer into JSON, String or buffer
        JsonObject message = new JsonObject(new String(buffer, StandardCharsets.UTF_8));

        if(attributePattern != null) {
            DeliveryOptions options = new DeliveryOptions();
            for(String attributeName : flowFile.getAttributes().keySet()) {
                if (attributePattern != null && attributePattern.matcher(attributeName).matches()) {
                    options.addHeader(attributeName, flowFile.getAttribute(attributeName));
                }
            }
            put(outboundAddress, message, options);
        } else {
            put(outboundAddress, message);
        }

        final Map<String, String> attributes = new HashMap<>();
        attributes.put("eventbus.address", outboundAddress);
        flowFile = session.putAllAttributes(flowFile, attributes);
        session.getProvenanceReporter().receive(flowFile, outboundAddress, "sent message to eventBus", stopWatch.getElapsed(TimeUnit.MILLISECONDS));
        getLogger().info("Successfully sent {} ({}) to EventBuss in {} millis", new Object[]{flowFile, flowFile.getSize(), stopWatch.getElapsed(TimeUnit.MILLISECONDS)});
        session.transfer(flowFile, REL_SUCCESS);
    } catch (ProcessException pe) {
        getLogger().error("Failed to send {} to EventBuss due to {}; routing to failure", new Object[] {flowFile, pe.toString()}, pe);
        session.transfer(flowFile, REL_FAILURE);
    }
}
 

import org.apache.nifi.util.StopWatch; //导入依赖的package包/类
@Override
    public void onTrigger(final ProcessContext context, final ProcessSession session) throws ProcessException {
        final ProcessorLog logger = getLogger();
        final Message<JsonObject> message = messageQueue.poll();
        if (message == null) {
            context.yield();
            return;
        }

        FlowFile flowFile = session.create();
        try {
            final StopWatch stopWatch = new StopWatch(true);

            flowFile = session.write(flowFile, new OutputStreamCallback() {
                @Override
                public void process(final OutputStream out) throws IOException {
                    out.write(message.body().encode().getBytes(StandardCharsets.UTF_8));
                }
            });

            if (flowFile.getSize() == 0L) {
                session.remove(flowFile);
            } else {
                final Map<String, String> attributes = new HashMap<>();

                if (headerPattern != null) {
                    MultiMap headers = message.headers();
                    headers.names()
                            .stream()
                            .filter(headerName -> headerPattern.matcher(headerName).matches())
                            .forEach(headerName -> attributes.put(headerName, headers.get(headerName)));

//                    attributes.putAll(
//                        headers.entries()
//                                .stream()
//                                .filter(entry -> headerPattern.matcher(entry.getKey()).matches())
//                                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue))
//                    );
                }
                attributes.put(CoreAttributes.MIME_TYPE.key(), "application/json");
                attributes.put(CoreAttributes.FILENAME.key(), flowFile.getAttribute(CoreAttributes.FILENAME.key()) + ".json");
                attributes.put("eventbus.address", message.address());

                flowFile = session.putAllAttributes(flowFile, attributes);
                session.getProvenanceReporter().receive(flowFile, message.address(), "received eventBus message", stopWatch.getElapsed(TimeUnit.MILLISECONDS));
                getLogger().info("Successfully received {} ({}) from EventBus in {} millis", new Object[]{flowFile, flowFile.getSize(), stopWatch.getElapsed(TimeUnit.MILLISECONDS)});
                session.transfer(flowFile, REL_SUCCESS);
            }
        } catch (Exception e) {
            session.remove(flowFile);
            throw e;
        }
    }