class 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();

        // If we are sending data, we need to ensure that we have at least 1 FlowFile to send. Otherwise,
        // we don't want to create a transaction at all.
        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) {
            // we do not yield here because the 'peer' will be penalized, and we won't communicate with that particular nifi instance
            // for a while due to penalization, but we can continue to talk to other nifi instances
            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();
        }
    }

    public  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);

}