import org.apache.nifi.components.state.Scope; //导入依赖的package包/类
private Map<String, String> incrementAndSet(StateManager stateManager, double value, boolean resetFirst) throws IOException {
    StateMap stateMap = stateManager.getState(Scope.LOCAL);

    String oldSumString = stateMap.get(TUMBLING_WINDOW_SUM_KEY);
    String oldCountString = stateMap.get(TUMBLING_WINDOW_COUNT_KEY);

    if (oldCountString == null || oldSumString == null) {
        oldSumString = "0.0";
        oldCountString = "0";
    }

    String newSumString = String.valueOf(value);
    String newCountString = "1";

    if (!resetFirst) {
        newSumString = String.valueOf(Double.valueOf(oldSumString) + value);
        newCountString = String.valueOf(Long.valueOf(oldCountString) + 1);
    }

    Map<String, String> newState = ImmutableMap.of(TUMBLING_WINDOW_SUM_KEY, newSumString, TUMBLING_WINDOW_COUNT_KEY, newCountString);
    stateManager.setState(newState, Scope.LOCAL);

    return resetFirst ?
        ImmutableMap.of(TUMBLING_WINDOW_SUM_KEY, oldSumString, TUMBLING_WINDOW_COUNT_KEY, oldCountString) :
        newState;
}
 

import org.apache.nifi.components.state.Scope; //导入依赖的package包/类
/**
 * Demonstrates reading and writing processor state values
 * @throws Exception
 */
@Test
public void testStateJavascript() throws Exception {
    final TestRunner runner = TestRunners.newTestRunner(new ExecuteScript());
    runner.setValidateExpressionUsage(false);
    runner.setProperty(SCRIPT_ENGINE, "ECMAScript");
    runner.setProperty(ScriptingComponentUtils.SCRIPT_FILE, "src/test/resources/executescript/state/state.js");
    runner.setProperty(ScriptingComponentUtils.MODULES, "src/test/resources/executescript");
    runner.assertValid();

    StateManager stateManager = runner.getStateManager();
    stateManager.clear(Scope.CLUSTER);
    Map<String, String> initialStateValues = new HashMap<>();
    initialStateValues.put("some-state", "foo");
    stateManager.setState(initialStateValues, Scope.CLUSTER);

    runner.enqueue("sample text".getBytes(StandardCharsets.UTF_8));
    runner.run();

    runner.assertAllFlowFilesTransferred("success", 1);
    StateMap resultStateValues = stateManager.getState(Scope.CLUSTER);
    Assert.assertEquals("foobar", resultStateValues.get("some-state"));
}
 

import org.apache.nifi.components.state.Scope; //导入依赖的package包/类
/**
 * Demonstrates reading and writing processor state values
 * @throws Exception
 */
@Test
public void testStatePython() throws Exception {
    final TestRunner runner = TestRunners.newTestRunner(new ExecuteScript());
    runner.setValidateExpressionUsage(false);
    runner.setProperty(SCRIPT_ENGINE, "python");
    runner.setProperty(ScriptingComponentUtils.SCRIPT_FILE, "src/test/resources/executescript/state/state.py");
    runner.setProperty(ScriptingComponentUtils.MODULES, "src/test/resources/executescript");
    runner.assertValid();

    StateManager stateManager = runner.getStateManager();
    stateManager.clear(Scope.CLUSTER);
    Map<String, String> initialStateValues = new HashMap<>();
    initialStateValues.put("some-state", "foo");
    stateManager.setState(initialStateValues, Scope.CLUSTER);

    runner.enqueue("sample text".getBytes(StandardCharsets.UTF_8));
    runner.run();

    runner.assertAllFlowFilesTransferred("success", 1);
    StateMap resultStateValues = stateManager.getState(Scope.CLUSTER);
    Assert.assertEquals("foobar", resultStateValues.get("some-state"));
}
 

import org.apache.nifi.components.state.Scope; //导入依赖的package包/类
@Override
protected Scope getStateScope(final ProcessContext context) {
    final String location = context.getProperty(DIRECTORY_LOCATION).getValue();
    if (LOCATION_REMOTE.getValue().equalsIgnoreCase(location)) {
        return Scope.CLUSTER;
    }

    return Scope.LOCAL;
}
 

import org.apache.nifi.components.state.Scope; //导入依赖的package包/类
@OnScheduled
public void onScheduled(final ProcessContext context) throws IOException {
    context.getStateManager().clear(Scope.LOCAL);

    this.stopWatch = new CheckedStopWatch(context.getProperty(TIME_WINDOW).asTimePeriod(TimeUnit.MILLISECONDS));
    this.stopWatch.start();
}
 

import org.apache.nifi.components.state.Scope; //导入依赖的package包/类
@Test
public void oneElapsed() throws InterruptedException, IOException {
    Map<String, String> attributes = ImmutableMap.of("foobar", "2");

    runner.enqueue(new byte[]{}, attributes);
    runner.enqueue(new byte[]{}, attributes);
    runner.enqueue(new byte[]{}, attributes);
    runner.enqueue(new byte[]{}, attributes);
    runner.enqueue(new byte[]{}, attributes);
    runner.enqueue(new byte[]{}, attributes);

    runner.run(5, false);

    StateMap beforeElapse = runner.getStateManager().getState(Scope.LOCAL);

    Thread.sleep(2000);
    runner.run(1, false, false);

    StateMap afterElapse = runner.getStateManager().getState(Scope.LOCAL);

    List<MockFlowFile> elapsed = runner.getFlowFilesForRelationship(ELAPSED);
    assertEquals(1, elapsed.size());
    runner.assertTransferCount(SUCCESS, 6);
    runner.assertTransferCount(FAILURE, 0);

    assertEquals("10.0", elapsed.get(0).getAttribute(TUMBLING_WINDOW_SUM_KEY));
    assertEquals("5", elapsed.get(0).getAttribute(TUMBLING_WINDOW_COUNT_KEY));

    assertEquals("10.0", beforeElapse.get(TUMBLING_WINDOW_SUM_KEY));
    assertEquals("5", beforeElapse.get(TUMBLING_WINDOW_COUNT_KEY));
    assertEquals("2.0", afterElapse.get(TUMBLING_WINDOW_SUM_KEY));
    assertEquals("1", afterElapse.get(TUMBLING_WINDOW_COUNT_KEY));
}
 

import org.apache.nifi.components.state.Scope; //导入依赖的package包/类
private long getLastEventId(StateManager stateManager) {
    try {
        final StateMap stateMap = stateManager.getState(Scope.CLUSTER);
        final String lastEventIdStr = stateMap.get("lastEventId");
        final long lastEventId = lastEventIdStr != null ? Long.parseLong(lastEventIdStr) : 0;
        return lastEventId;
    } catch (final IOException ioe) {
        getLogger().warn("Failed to retrieve the last event id from the "
                + "state manager.", ioe);
        return 0;
    }
}
 

import org.apache.nifi.components.state.Scope; //导入依赖的package包/类
@Override
protected Scope getStateScope(final ProcessContext context) {
    // Use cluster scope so that component can be run on Primary Node Only and can still
    // pick up where it left off, even if the Primary Node changes.
    return Scope.CLUSTER;
}
 

import org.apache.nifi.components.state.Scope; //导入依赖的package包/类
private void persist(final long listingTimestamp, final long processedTimestamp, final StateManager stateManager, final Scope scope) throws IOException {
    final Map<String, String> updatedState = new HashMap<>(1);
    updatedState.put(LISTING_TIMESTAMP_KEY, String.valueOf(listingTimestamp));
    updatedState.put(PROCESSED_TIMESTAMP_KEY, String.valueOf(processedTimestamp));
    stateManager.setState(updatedState, scope);
}
 

import org.apache.nifi.components.state.Scope; //导入依赖的package包/类
private void setLastEventId(StateManager stateManager, long eventId) throws IOException {
    final StateMap stateMap = stateManager.getState(Scope.CLUSTER);
    final Map<String, String> statePropertyMap = new HashMap<>(stateMap.toMap());
    statePropertyMap.put("lastEventId", Long.toString(eventId));
    stateManager.setState(statePropertyMap, Scope.CLUSTER);
}
 

import org.apache.nifi.components.state.Scope; //导入依赖的package包/类
/**
 * Returns a Scope that specifies where the state should be managed for this Processor
 *
 * @param context the ProcessContext to use in order to make a determination
 * @return a Scope that specifies where the state should be managed for this Processor
 */
protected abstract Scope getStateScope(final ProcessContext context);