@Test public void testSendHttp() throws Exception { setupMock(SiteToSiteTransportProtocol.HTTP, TransferDirection.SEND); setupMockProcessSession(); final String peerUrl = "https: final PeerDescription peerDescription = new PeerDescription("node1.example.com", 8080, true); final HttpCommunicationsSession commsSession = new HttpCommunicationsSession(); commsSession.setUserDn("nifi.node1.example.com"); final Peer peer = new Peer(peerDescription, commsSession, peerUrl, REMOTE_CLUSTER_URL); final String flowFileEndpointUri = "https: doReturn(peer).when(transaction).getCommunicant(); commsSession.setDataTransferUrl(flowFileEndpointUri); final MockFlowFile flowFile = processSession.createFlowFile("0123456789".getBytes()); sessionState.getFlowFileQueue().offer(flowFile); port.onTrigger(processContext, processSession); final List<ProvenanceEventRecord> provenanceEvents = sessionState.getProvenanceEvents(); assertEquals(1, provenanceEvents.size()); final ProvenanceEventRecord provenanceEvent = provenanceEvents.get(0); assertEquals(ProvenanceEventType.SEND, provenanceEvent.getEventType()); assertEquals(flowFileEndpointUri, provenanceEvent.getTransitUri()); assertEquals("Remote DN=nifi.node1.example.com", provenanceEvent.getDetails()); }
@Override public void onTrigger(final ProcessContext context, final ProcessSession session) { if (!remoteGroup.isTransmitting()) { logger.debug("{} {} is not transmitting; will not send/receive", this, remoteGroup); return; } if (getConnectableType() == ConnectableType.REMOTE_INPUT_PORT && session.getQueueSize().getObjectCount() == 0) { logger.debug("{} No data to send", this); return; } final String url = getRemoteProcessGroup().getTargetUri(); final FlowFile firstFlowFile; if (getConnectableType() == ConnectableType.REMOTE_INPUT_PORT) { firstFlowFile = session.get(); if (firstFlowFile == null) { return; } } else { firstFlowFile = null; } final SiteToSiteClient client = getSiteToSiteClient(); final Transaction transaction; try { transaction = client.createTransaction(transferDirection); } catch (final PortNotRunningException e) { context.yield(); this.targetRunning.set(false); final String message = String.format("%s failed to communicate with %s because the remote instance indicates that the port is not in a valid state", this, url); logger.error(message); session.rollback(); remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); return; } catch (final UnknownPortException e) { context.yield(); this.targetExists.set(false); final String message = String.format("%s failed to communicate with %s because the remote instance indicates that the port no longer exists", this, url); logger.error(message); session.rollback(); remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); return; } catch (final IOException e) { final String message = String.format("%s failed to communicate with %s due to %s", this, url, e.toString()); logger.error(message); if (logger.isDebugEnabled()) { logger.error("", e); } session.rollback(); remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); return; } if (transaction == null) { logger.debug("{} Unable to create transaction to communicate with; all peers must be penalized, so yielding context", this); session.rollback(); context.yield(); return; } try { if (getConnectableType() == ConnectableType.REMOTE_INPUT_PORT) { transferFlowFiles(transaction, context, session, firstFlowFile); } else { final int numReceived = receiveFlowFiles(transaction, context, session); if (numReceived == 0) { context.yield(); } } session.commit(); } catch (final Throwable t) { final String message = String.format("%s failed to communicate with remote NiFi instance due to %s", this, t.toString()); logger.error("{} failed to communicate with remote NiFi instance due to {}", this, t.toString()); if (logger.isDebugEnabled()) { logger.error("", t); } remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); transaction.error(); session.rollback(); } }
StandardRemoteGroupPort extends RemoteGroupPort { @Override public void onTrigger(final ProcessContext context, final ProcessSession session) { if (!remoteGroup.isTransmitting()) { logger.debug("{} {} is not transmitting; will not send/receive", this, remoteGroup); return; } if (getConnectableType() == ConnectableType.REMOTE_INPUT_PORT && session.getQueueSize().getObjectCount() == 0) { logger.debug("{} No data to send", this); return; } final String url = getRemoteProcessGroup().getTargetUri(); final FlowFile firstFlowFile; if (getConnectableType() == ConnectableType.REMOTE_INPUT_PORT) { firstFlowFile = session.get(); if (firstFlowFile == null) { return; } } else { firstFlowFile = null; } final SiteToSiteClient client = getSiteToSiteClient(); final Transaction transaction; try { transaction = client.createTransaction(transferDirection); } catch (final PortNotRunningException e) { context.yield(); this.targetRunning.set(false); final String message = String.format("%s failed to communicate with %s because the remote instance indicates that the port is not in a valid state", this, url); logger.error(message); session.rollback(); remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); return; } catch (final UnknownPortException e) { context.yield(); this.targetExists.set(false); final String message = String.format("%s failed to communicate with %s because the remote instance indicates that the port no longer exists", this, url); logger.error(message); session.rollback(); remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); return; } catch (final IOException e) { final String message = String.format("%s failed to communicate with %s due to %s", this, url, e.toString()); logger.error(message); if (logger.isDebugEnabled()) { logger.error("", e); } session.rollback(); remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); return; } if (transaction == null) { logger.debug("{} Unable to create transaction to communicate with; all peers must be penalized, so yielding context", this); session.rollback(); context.yield(); return; } try { if (getConnectableType() == ConnectableType.REMOTE_INPUT_PORT) { transferFlowFiles(transaction, context, session, firstFlowFile); } else { final int numReceived = receiveFlowFiles(transaction, context, session); if (numReceived == 0) { context.yield(); } } session.commit(); } catch (final Throwable t) { final String message = String.format("%s failed to communicate with remote NiFi instance due to %s", this, t.toString()); logger.error("{} failed to communicate with remote NiFi instance due to {}", this, t.toString()); if (logger.isDebugEnabled()) { logger.error("", t); } remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); transaction.error(); session.rollback(); } } }
StandardRemoteGroupPort extends RemoteGroupPort { @Override public void onTrigger(final ProcessContext context, final ProcessSession session) { if (!remoteGroup.isTransmitting()) { logger.debug("{} {} is not transmitting; will not send/receive", this, remoteGroup); return; } if (getConnectableType() == ConnectableType.REMOTE_INPUT_PORT && session.getQueueSize().getObjectCount() == 0) { logger.debug("{} No data to send", this); return; } final String url = getRemoteProcessGroup().getTargetUri(); final FlowFile firstFlowFile; if (getConnectableType() == ConnectableType.REMOTE_INPUT_PORT) { firstFlowFile = session.get(); if (firstFlowFile == null) { return; } } else { firstFlowFile = null; } final SiteToSiteClient client = getSiteToSiteClient(); final Transaction transaction; try { transaction = client.createTransaction(transferDirection); } catch (final PortNotRunningException e) { context.yield(); this.targetRunning.set(false); final String message = String.format("%s failed to communicate with %s because the remote instance indicates that the port is not in a valid state", this, url); logger.error(message); session.rollback(); remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); return; } catch (final UnknownPortException e) { context.yield(); this.targetExists.set(false); final String message = String.format("%s failed to communicate with %s because the remote instance indicates that the port no longer exists", this, url); logger.error(message); session.rollback(); remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); return; } catch (final IOException e) { final String message = String.format("%s failed to communicate with %s due to %s", this, url, e.toString()); logger.error(message); if (logger.isDebugEnabled()) { logger.error("", e); } session.rollback(); remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); return; } if (transaction == null) { logger.debug("{} Unable to create transaction to communicate with; all peers must be penalized, so yielding context", this); session.rollback(); context.yield(); return; } try { if (getConnectableType() == ConnectableType.REMOTE_INPUT_PORT) { transferFlowFiles(transaction, context, session, firstFlowFile); } else { final int numReceived = receiveFlowFiles(transaction, context, session); if (numReceived == 0) { context.yield(); } } session.commit(); } catch (final Throwable t) { final String message = String.format("%s failed to communicate with remote NiFi instance due to %s", this, t.toString()); logger.error("{} failed to communicate with remote NiFi instance due to {}", this, t.toString()); if (logger.isDebugEnabled()) { logger.error("", t); } remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); transaction.error(); session.rollback(); } } StandardRemoteGroupPort(final String id, final String name, final ProcessGroup processGroup, final RemoteProcessGroup remoteGroup,
            final TransferDirection direction, final ConnectableType type, final SSLContext sslContext, final ProcessScheduler scheduler,
        final NiFiProperties nifiProperties); }
StandardRemoteGroupPort extends RemoteGroupPort { @Override public void onTrigger(final ProcessContext context, final ProcessSession session) { if (!remoteGroup.isTransmitting()) { logger.debug("{} {} is not transmitting; will not send/receive", this, remoteGroup); return; } if (getConnectableType() == ConnectableType.REMOTE_INPUT_PORT && session.getQueueSize().getObjectCount() == 0) { logger.debug("{} No data to send", this); return; } final String url = getRemoteProcessGroup().getTargetUri(); final FlowFile firstFlowFile; if (getConnectableType() == ConnectableType.REMOTE_INPUT_PORT) { firstFlowFile = session.get(); if (firstFlowFile == null) { return; } } else { firstFlowFile = null; } final SiteToSiteClient client = getSiteToSiteClient(); final Transaction transaction; try { transaction = client.createTransaction(transferDirection); } catch (final PortNotRunningException e) { context.yield(); this.targetRunning.set(false); final String message = String.format("%s failed to communicate with %s because the remote instance indicates that the port is not in a valid state", this, url); logger.error(message); session.rollback(); remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); return; } catch (final UnknownPortException e) { context.yield(); this.targetExists.set(false); final String message = String.format("%s failed to communicate with %s because the remote instance indicates that the port no longer exists", this, url); logger.error(message); session.rollback(); remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); return; } catch (final IOException e) { final String message = String.format("%s failed to communicate with %s due to %s", this, url, e.toString()); logger.error(message); if (logger.isDebugEnabled()) { logger.error("", e); } session.rollback(); remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); return; } if (transaction == null) { logger.debug("{} Unable to create transaction to communicate with; all peers must be penalized, so yielding context", this); session.rollback(); context.yield(); return; } try { if (getConnectableType() == ConnectableType.REMOTE_INPUT_PORT) { transferFlowFiles(transaction, context, session, firstFlowFile); } else { final int numReceived = receiveFlowFiles(transaction, context, session); if (numReceived == 0) { context.yield(); } } session.commit(); } catch (final Throwable t) { final String message = String.format("%s failed to communicate with remote NiFi instance due to %s", this, t.toString()); logger.error("{} failed to communicate with remote NiFi instance due to {}", this, t.toString()); if (logger.isDebugEnabled()) { logger.error("", t); } remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); transaction.error(); session.rollback(); } } StandardRemoteGroupPort(final String id, final String name, final ProcessGroup processGroup, final RemoteProcessGroup remoteGroup,
            final TransferDirection direction, final ConnectableType type, final SSLContext sslContext, final ProcessScheduler scheduler,
        final NiFiProperties nifiProperties); @Override boolean isTargetRunning(); void setTargetRunning(final boolean targetRunning); @Override boolean isTriggerWhenEmpty(); @Override Resource getResource(); @Override Authorizable getParentAuthorizable(); @Override void shutdown(); @Override void onSchedulingStart(); @Override void onTrigger(final ProcessContext context, final ProcessSession session); @Override String getYieldPeriod(); @Override boolean getTargetExists(); @Override boolean isValid(); @Override Collection<ValidationResult> getValidationErrors(); @Override void verifyCanStart(); @Override void setUseCompression(final boolean useCompression); @Override boolean isUseCompression(); @Override String toString(); @Override RemoteProcessGroup getRemoteProcessGroup(); @Override TransferDirection getTransferDirection(); void setTargetExists(final boolean exists); @Override void removeConnection(final Connection connection); @Override SchedulingStrategy getSchedulingStrategy(); @Override boolean isSideEffectFree(); @Override String getComponentType(); }
StandardRemoteGroupPort extends RemoteGroupPort { @Override public void onTrigger(final ProcessContext context, final ProcessSession session) { if (!remoteGroup.isTransmitting()) { logger.debug("{} {} is not transmitting; will not send/receive", this, remoteGroup); return; } if (getConnectableType() == ConnectableType.REMOTE_INPUT_PORT && session.getQueueSize().getObjectCount() == 0) { logger.debug("{} No data to send", this); return; } final String url = getRemoteProcessGroup().getTargetUri(); final FlowFile firstFlowFile; if (getConnectableType() == ConnectableType.REMOTE_INPUT_PORT) { firstFlowFile = session.get(); if (firstFlowFile == null) { return; } } else { firstFlowFile = null; } final SiteToSiteClient client = getSiteToSiteClient(); final Transaction transaction; try { transaction = client.createTransaction(transferDirection); } catch (final PortNotRunningException e) { context.yield(); this.targetRunning.set(false); final String message = String.format("%s failed to communicate with %s because the remote instance indicates that the port is not in a valid state", this, url); logger.error(message); session.rollback(); remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); return; } catch (final UnknownPortException e) { context.yield(); this.targetExists.set(false); final String message = String.format("%s failed to communicate with %s because the remote instance indicates that the port no longer exists", this, url); logger.error(message); session.rollback(); remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); return; } catch (final IOException e) { final String message = String.format("%s failed to communicate with %s due to %s", this, url, e.toString()); logger.error(message); if (logger.isDebugEnabled()) { logger.error("", e); } session.rollback(); remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); return; } if (transaction == null) { logger.debug("{} Unable to create transaction to communicate with; all peers must be penalized, so yielding context", this); session.rollback(); context.yield(); return; } try { if (getConnectableType() == ConnectableType.REMOTE_INPUT_PORT) { transferFlowFiles(transaction, context, session, firstFlowFile); } else { final int numReceived = receiveFlowFiles(transaction, context, session); if (numReceived == 0) { context.yield(); } } session.commit(); } catch (final Throwable t) { final String message = String.format("%s failed to communicate with remote NiFi instance due to %s", this, t.toString()); logger.error("{} failed to communicate with remote NiFi instance due to {}", this, t.toString()); if (logger.isDebugEnabled()) { logger.error("", t); } remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); transaction.error(); session.rollback(); } } StandardRemoteGroupPort(final String id, final String name, final ProcessGroup processGroup, final RemoteProcessGroup remoteGroup,
            final TransferDirection direction, final ConnectableType type, final SSLContext sslContext, final ProcessScheduler scheduler,
        final NiFiProperties nifiProperties); @Override boolean isTargetRunning(); void setTargetRunning(final boolean targetRunning); @Override boolean isTriggerWhenEmpty(); @Override Resource getResource(); @Override Authorizable getParentAuthorizable(); @Override void shutdown(); @Override void onSchedulingStart(); @Override void onTrigger(final ProcessContext context, final ProcessSession session); @Override String getYieldPeriod(); @Override boolean getTargetExists(); @Override boolean isValid(); @Override Collection<ValidationResult> getValidationErrors(); @Override void verifyCanStart(); @Override void setUseCompression(final boolean useCompression); @Override boolean isUseCompression(); @Override String toString(); @Override RemoteProcessGroup getRemoteProcessGroup(); @Override TransferDirection getTransferDirection(); void setTargetExists(final boolean exists); @Override void removeConnection(final Connection connection); @Override SchedulingStrategy getSchedulingStrategy(); @Override boolean isSideEffectFree(); @Override String getComponentType(); static final String USER_AGENT; static final String CONTENT_TYPE; static final int GZIP_COMPRESSION_LEVEL; }