@Test public void testLoadExistingFlow() throws IOException { byte[] flowBytes = IOUtils.toByteArray(StandardFlowServiceTest.class.getResourceAsStream("/conf/all-flow.xml")); flowService.load(new StandardDataFlow(flowBytes, null, null)); flowBytes = IOUtils.toByteArray(StandardFlowServiceTest.class.getResourceAsStream("/conf/all-flow-inheritable.xml")); flowService.load(new StandardDataFlow(flowBytes, null, null)); FlowSerializer serializer = new StandardFlowSerializer(mockEncryptor); ByteArrayOutputStream baos = new ByteArrayOutputStream(); serializer.serialize(flowController, baos); String expectedFlow = new String(flowBytes).trim(); String actualFlow = new String(baos.toByteArray()).trim(); Assert.assertEquals(expectedFlow, actualFlow); }
@Override public void load(final DataFlow dataFlow) throws IOException, FlowSerializationException, FlowSynchronizationException, UninheritableFlowException { if (configuredForClustering) { final DataFlow proposedFlow = (dataFlow == null) ? createDataFlow() : dataFlow; final boolean localFlowEmpty = StandardFlowSynchronizer.isEmpty(proposedFlow); final ConnectionResponse response = connect(true, localFlowEmpty, proposedFlow); writeLock.lock(); try { if (response == null || response.shouldTryLater()) { logger.info("Flow controller will load local dataflow and suspend connection handshake until a cluster connection response is received."); loadFromBytes(proposedFlow, false); controller.setNodeId(nodeId); clusterCoordinator.setLocalNodeIdentifier(nodeId); controller.setClustered(true, null); clusterCoordinator.setConnected(false); controller.setConnectionStatus(new NodeConnectionStatus(nodeId, DisconnectionCode.NOT_YET_CONNECTED)); controller.startHeartbeating(); try { controller.onFlowInitialized(autoResumeState); } catch (final Exception ex) { logger.warn("Unable to start all processors due to invalid flow configuration."); if (logger.isDebugEnabled()) { logger.warn(StringUtils.EMPTY, ex); } } } else { try { loadFromConnectionResponse(response); dao.save(controller, true); } catch (final Exception e) { logger.error("Failed to load flow from cluster due to: " + e, e); handleConnectionFailure(e); throw new IOException(e); } } } finally { writeLock.unlock(); } } else { writeLock.lock(); try { loadFromBytes(dataFlow, true); } finally { writeLock.unlock(); } } }
StandardFlowService implements FlowService, ProtocolHandler { @Override public void load(final DataFlow dataFlow) throws IOException, FlowSerializationException, FlowSynchronizationException, UninheritableFlowException { if (configuredForClustering) { final DataFlow proposedFlow = (dataFlow == null) ? createDataFlow() : dataFlow; final boolean localFlowEmpty = StandardFlowSynchronizer.isEmpty(proposedFlow); final ConnectionResponse response = connect(true, localFlowEmpty, proposedFlow); writeLock.lock(); try { if (response == null || response.shouldTryLater()) { logger.info("Flow controller will load local dataflow and suspend connection handshake until a cluster connection response is received."); loadFromBytes(proposedFlow, false); controller.setNodeId(nodeId); clusterCoordinator.setLocalNodeIdentifier(nodeId); controller.setClustered(true, null); clusterCoordinator.setConnected(false); controller.setConnectionStatus(new NodeConnectionStatus(nodeId, DisconnectionCode.NOT_YET_CONNECTED)); controller.startHeartbeating(); try { controller.onFlowInitialized(autoResumeState); } catch (final Exception ex) { logger.warn("Unable to start all processors due to invalid flow configuration."); if (logger.isDebugEnabled()) { logger.warn(StringUtils.EMPTY, ex); } } } else { try { loadFromConnectionResponse(response); dao.save(controller, true); } catch (final Exception e) { logger.error("Failed to load flow from cluster due to: " + e, e); handleConnectionFailure(e); throw new IOException(e); } } } finally { writeLock.unlock(); } } else { writeLock.lock(); try { loadFromBytes(dataFlow, true); } finally { writeLock.unlock(); } } } }
StandardFlowService implements FlowService, ProtocolHandler { @Override public void load(final DataFlow dataFlow) throws IOException, FlowSerializationException, FlowSynchronizationException, UninheritableFlowException { if (configuredForClustering) { final DataFlow proposedFlow = (dataFlow == null) ? createDataFlow() : dataFlow; final boolean localFlowEmpty = StandardFlowSynchronizer.isEmpty(proposedFlow); final ConnectionResponse response = connect(true, localFlowEmpty, proposedFlow); writeLock.lock(); try { if (response == null || response.shouldTryLater()) { logger.info("Flow controller will load local dataflow and suspend connection handshake until a cluster connection response is received."); loadFromBytes(proposedFlow, false); controller.setNodeId(nodeId); clusterCoordinator.setLocalNodeIdentifier(nodeId); controller.setClustered(true, null); clusterCoordinator.setConnected(false); controller.setConnectionStatus(new NodeConnectionStatus(nodeId, DisconnectionCode.NOT_YET_CONNECTED)); controller.startHeartbeating(); try { controller.onFlowInitialized(autoResumeState); } catch (final Exception ex) { logger.warn("Unable to start all processors due to invalid flow configuration."); if (logger.isDebugEnabled()) { logger.warn(StringUtils.EMPTY, ex); } } } else { try { loadFromConnectionResponse(response); dao.save(controller, true); } catch (final Exception e) { logger.error("Failed to load flow from cluster due to: " + e, e); handleConnectionFailure(e); throw new IOException(e); } } } finally { writeLock.unlock(); } } else { writeLock.lock(); try { loadFromBytes(dataFlow, true); } finally { writeLock.unlock(); } } } private  StandardFlowService(
            final FlowController controller,
            final NiFiProperties nifiProperties,
            final NodeProtocolSenderListener senderListener,
            final StringEncryptor encryptor,
            final boolean configuredForClustering,
            final ClusterCoordinator clusterCoordinator,
            final RevisionManager revisionManager,
            final Authorizer authorizer); }
StandardFlowService implements FlowService, ProtocolHandler { @Override public void load(final DataFlow dataFlow) throws IOException, FlowSerializationException, FlowSynchronizationException, UninheritableFlowException { if (configuredForClustering) { final DataFlow proposedFlow = (dataFlow == null) ? createDataFlow() : dataFlow; final boolean localFlowEmpty = StandardFlowSynchronizer.isEmpty(proposedFlow); final ConnectionResponse response = connect(true, localFlowEmpty, proposedFlow); writeLock.lock(); try { if (response == null || response.shouldTryLater()) { logger.info("Flow controller will load local dataflow and suspend connection handshake until a cluster connection response is received."); loadFromBytes(proposedFlow, false); controller.setNodeId(nodeId); clusterCoordinator.setLocalNodeIdentifier(nodeId); controller.setClustered(true, null); clusterCoordinator.setConnected(false); controller.setConnectionStatus(new NodeConnectionStatus(nodeId, DisconnectionCode.NOT_YET_CONNECTED)); controller.startHeartbeating(); try { controller.onFlowInitialized(autoResumeState); } catch (final Exception ex) { logger.warn("Unable to start all processors due to invalid flow configuration."); if (logger.isDebugEnabled()) { logger.warn(StringUtils.EMPTY, ex); } } } else { try { loadFromConnectionResponse(response); dao.save(controller, true); } catch (final Exception e) { logger.error("Failed to load flow from cluster due to: " + e, e); handleConnectionFailure(e); throw new IOException(e); } } } finally { writeLock.unlock(); } } else { writeLock.lock(); try { loadFromBytes(dataFlow, true); } finally { writeLock.unlock(); } } } private  StandardFlowService(
            final FlowController controller,
            final NiFiProperties nifiProperties,
            final NodeProtocolSenderListener senderListener,
            final StringEncryptor encryptor,
            final boolean configuredForClustering,
            final ClusterCoordinator clusterCoordinator,
            final RevisionManager revisionManager,
            final Authorizer authorizer); static StandardFlowService createStandaloneInstance(
            final FlowController controller,
            final NiFiProperties nifiProperties,
            final StringEncryptor encryptor,
            final RevisionManager revisionManager,
            final Authorizer authorizer); static StandardFlowService createClusteredInstance(
            final FlowController controller,
            final NiFiProperties nifiProperties,
            final NodeProtocolSenderListener senderListener,
            final ClusterCoordinator coordinator,
            final StringEncryptor encryptor,
            final RevisionManager revisionManager,
            final Authorizer authorizer); @Override void saveFlowChanges(); @Override void saveFlowChanges(final OutputStream outStream); @Override void overwriteFlow(final InputStream is); @Override void saveFlowChanges(final TimeUnit delayUnit, final long delay); @Override void saveFlowChanges(final TimeUnit delayUnit, final long delay, final boolean archive); @Override boolean isRunning(); @Override void start(); @Override void stop(final boolean force); @Override boolean canHandle(final ProtocolMessage msg); @Override ProtocolMessage handle(final ProtocolMessage request); @Override void load(final DataFlow dataFlow); @Override StandardDataFlow createDataFlow(); List<Template> loadTemplates(); @Override void copyCurrentFlow(final OutputStream os); void loadSnippets(final byte[] bytes); }
StandardFlowService implements FlowService, ProtocolHandler { @Override public void load(final DataFlow dataFlow) throws IOException, FlowSerializationException, FlowSynchronizationException, UninheritableFlowException { if (configuredForClustering) { final DataFlow proposedFlow = (dataFlow == null) ? createDataFlow() : dataFlow; final boolean localFlowEmpty = StandardFlowSynchronizer.isEmpty(proposedFlow); final ConnectionResponse response = connect(true, localFlowEmpty, proposedFlow); writeLock.lock(); try { if (response == null || response.shouldTryLater()) { logger.info("Flow controller will load local dataflow and suspend connection handshake until a cluster connection response is received."); loadFromBytes(proposedFlow, false); controller.setNodeId(nodeId); clusterCoordinator.setLocalNodeIdentifier(nodeId); controller.setClustered(true, null); clusterCoordinator.setConnected(false); controller.setConnectionStatus(new NodeConnectionStatus(nodeId, DisconnectionCode.NOT_YET_CONNECTED)); controller.startHeartbeating(); try { controller.onFlowInitialized(autoResumeState); } catch (final Exception ex) { logger.warn("Unable to start all processors due to invalid flow configuration."); if (logger.isDebugEnabled()) { logger.warn(StringUtils.EMPTY, ex); } } } else { try { loadFromConnectionResponse(response); dao.save(controller, true); } catch (final Exception e) { logger.error("Failed to load flow from cluster due to: " + e, e); handleConnectionFailure(e); throw new IOException(e); } } } finally { writeLock.unlock(); } } else { writeLock.lock(); try { loadFromBytes(dataFlow, true); } finally { writeLock.unlock(); } } } private  StandardFlowService(
            final FlowController controller,
            final NiFiProperties nifiProperties,
            final NodeProtocolSenderListener senderListener,
            final StringEncryptor encryptor,
            final boolean configuredForClustering,
            final ClusterCoordinator clusterCoordinator,
            final RevisionManager revisionManager,
            final Authorizer authorizer); static StandardFlowService createStandaloneInstance(
            final FlowController controller,
            final NiFiProperties nifiProperties,
            final StringEncryptor encryptor,
            final RevisionManager revisionManager,
            final Authorizer authorizer); static StandardFlowService createClusteredInstance(
            final FlowController controller,
            final NiFiProperties nifiProperties,
            final NodeProtocolSenderListener senderListener,
            final ClusterCoordinator coordinator,
            final StringEncryptor encryptor,
            final RevisionManager revisionManager,
            final Authorizer authorizer); @Override void saveFlowChanges(); @Override void saveFlowChanges(final OutputStream outStream); @Override void overwriteFlow(final InputStream is); @Override void saveFlowChanges(final TimeUnit delayUnit, final long delay); @Override void saveFlowChanges(final TimeUnit delayUnit, final long delay, final boolean archive); @Override boolean isRunning(); @Override void start(); @Override void stop(final boolean force); @Override boolean canHandle(final ProtocolMessage msg); @Override ProtocolMessage handle(final ProtocolMessage request); @Override void load(final DataFlow dataFlow); @Override StandardDataFlow createDataFlow(); List<Template> loadTemplates(); @Override void copyCurrentFlow(final OutputStream os); void loadSnippets(final byte[] bytes);  }