import org.apache.nifi.components.state.StateManager; //导入依赖的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.StateManager; //导入依赖的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.StateManager; //导入依赖的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.StateManager; //导入依赖的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.StateManager; //导入依赖的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.StateManager; //导入依赖的package包/类
@Override
public StateManager getStateManager() {
    return null;
}
 

import org.apache.nifi.components.state.StateManager; //导入依赖的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);
}