private void compare(final VersionedRemoteProcessGroup rpgA, final VersionedRemoteProcessGroup rpgB, final Set<FlowDifference> differences) {
    if (compareComponents(rpgA, rpgB, differences, false, true, false)) { // do not compare comments for RPG because they come from remote system, not our local flow
        return;
    }

    addIfDifferent(differences, DifferenceType.RPG_COMMS_TIMEOUT_CHANGED, rpgA, rpgB, VersionedRemoteProcessGroup::getCommunicationsTimeout);
    addIfDifferent(differences, DifferenceType.RPG_NETWORK_INTERFACE_CHANGED, rpgA, rpgB, VersionedRemoteProcessGroup::getLocalNetworkInterface);
    addIfDifferent(differences, DifferenceType.RPG_PROXY_HOST_CHANGED, rpgA, rpgB, VersionedRemoteProcessGroup::getProxyHost);
    addIfDifferent(differences, DifferenceType.RPG_PROXY_PORT_CHANGED, rpgA, rpgB, VersionedRemoteProcessGroup::getProxyPort);
    addIfDifferent(differences, DifferenceType.RPG_PROXY_USER_CHANGED, rpgA, rpgB, VersionedRemoteProcessGroup::getProxyUser);
    addIfDifferent(differences, DifferenceType.RPG_TRANSPORT_PROTOCOL_CHANGED, rpgA, rpgB, VersionedRemoteProcessGroup::getTransportProtocol);
    addIfDifferent(differences, DifferenceType.YIELD_DURATION_CHANGED, rpgA, rpgB, VersionedRemoteProcessGroup::getYieldDuration);

    differences.addAll(compareComponents(rpgA.getInputPorts(), rpgB.getInputPorts(), this::compare));
    differences.addAll(compareComponents(rpgA.getOutputPorts(), rpgB.getOutputPorts(), this::compare));
}
 

public StatelessRemoteInputPort(final VersionedRemoteProcessGroup rpg, final VersionedRemoteGroupPort remotePort, final SSLContext sslContext) {
    final String timeout = rpg.getCommunicationsTimeout();
    final long timeoutMillis = FormatUtils.getTimeDuration(timeout, TimeUnit.MILLISECONDS);

    url = rpg.getTargetUris();
    name = remotePort.getName();

    client = new SiteToSiteClient.Builder()
        .portName(remotePort.getName())
        .timeout(timeoutMillis, TimeUnit.MILLISECONDS)
        .transportProtocol(SiteToSiteTransportProtocol.valueOf(rpg.getTransportProtocol()))
        .url(rpg.getTargetUris())
        .useCompression(remotePort.isUseCompression())
        .sslContext(sslContext)
        .eventReporter(EventReporter.NO_OP)
        .build();
}
 

public VersionedRemoteProcessGroup mapRemoteProcessGroup(final RemoteProcessGroup remoteGroup) {
    final VersionedRemoteProcessGroup rpg = new InstantiatedVersionedRemoteProcessGroup(remoteGroup.getIdentifier(), remoteGroup.getProcessGroupIdentifier());
    rpg.setIdentifier(getId(remoteGroup.getVersionedComponentId(), remoteGroup.getIdentifier()));
    rpg.setGroupIdentifier(getGroupId(remoteGroup.getProcessGroupIdentifier()));
    rpg.setComments(remoteGroup.getComments());
    rpg.setCommunicationsTimeout(remoteGroup.getCommunicationsTimeout());
    rpg.setLocalNetworkInterface(remoteGroup.getNetworkInterface());
    rpg.setName(remoteGroup.getName());
    rpg.setInputPorts(remoteGroup.getInputPorts().stream()
        .map(port -> mapRemotePort(port, ComponentType.REMOTE_INPUT_PORT))
        .collect(Collectors.toSet()));
    rpg.setOutputPorts(remoteGroup.getOutputPorts().stream()
        .map(port -> mapRemotePort(port, ComponentType.REMOTE_OUTPUT_PORT))
        .collect(Collectors.toSet()));
    rpg.setPosition(mapPosition(remoteGroup.getPosition()));
    rpg.setProxyHost(remoteGroup.getProxyHost());
    rpg.setProxyPort(remoteGroup.getProxyPort());
    rpg.setProxyUser(remoteGroup.getProxyUser());
    rpg.setTargetUri(remoteGroup.getTargetUri());
    rpg.setTargetUris(remoteGroup.getTargetUris());
    rpg.setTransportProtocol(remoteGroup.getTransportProtocol().name());
    rpg.setYieldDuration(remoteGroup.getYieldDuration());
    return rpg;
}
 

@Override
public RemoteProcessGroupSchema apply(VersionedRemoteProcessGroup versionedRemoteProcessGroup) {
    Map<String, Object> map = new HashMap<>();
    map.put(CommonPropertyKeys.ID_KEY, versionedRemoteProcessGroup.getIdentifier());
    map.put(CommonPropertyKeys.NAME_KEY, versionedRemoteProcessGroup.getName());
    map.put(RemoteProcessGroupSchema.URL_KEY, versionedRemoteProcessGroup.getTargetUri());

    Set<VersionedRemoteGroupPort> inputPorts = versionedRemoteProcessGroup.getInputPorts();
    if (inputPorts != null) {
        map.put(CommonPropertyKeys.INPUT_PORTS_KEY, inputPorts.stream()
                .map(remotePortSchemaFunction)
                .map(RemotePortSchema::toMap)
                .collect(Collectors.toList()));
    }

    Set<VersionedRemoteGroupPort> outputPorts = versionedRemoteProcessGroup.getOutputPorts();
    if (outputPorts != null) {
        map.put(CommonPropertyKeys.OUTPUT_PORTS_KEY, outputPorts.stream()
                .map(remotePortSchemaFunction)
                .map(RemotePortSchema::toMap)
                .collect(Collectors.toList()));
    }


    map.put(CommonPropertyKeys.COMMENT_KEY, versionedRemoteProcessGroup.getComments());
    map.put(RemoteProcessGroupSchema.TIMEOUT_KEY, versionedRemoteProcessGroup.getCommunicationsTimeout());
    map.put(CommonPropertyKeys.YIELD_PERIOD_KEY, versionedRemoteProcessGroup.getYieldDuration());
    map.put(RemoteProcessGroupSchema.TRANSPORT_PROTOCOL_KEY, versionedRemoteProcessGroup.getTransportProtocol());
    map.put(RemoteProcessGroupSchema.PROXY_HOST_KEY, versionedRemoteProcessGroup.getProxyHost());
    map.put(RemoteProcessGroupSchema.PROXY_PORT_KEY, versionedRemoteProcessGroup.getProxyPort());
    map.put(RemoteProcessGroupSchema.PROXY_USER_KEY, versionedRemoteProcessGroup.getProxyUser());

    // TODO - we don't have this in registry data model, most likely templates blank it out too?
    //map.put(RemoteProcessGroupSchema.PROXY_PASSWORD_KEY, versionedRemoteProcessGroup.getProxyPassword());

    map.put(RemoteProcessGroupSchema.LOCAL_NETWORK_INTERFACE_KEY, versionedRemoteProcessGroup.getLocalNetworkInterface());
    return new RemoteProcessGroupSchema(map);
}
 

public StatelessRemoteOutputPort(final VersionedRemoteProcessGroup rpg, final VersionedRemoteGroupPort remotePort, final SSLContext sslContext) {
    final String timeout = rpg.getCommunicationsTimeout();
    final long timeoutMillis = FormatUtils.getTimeDuration(timeout, TimeUnit.MILLISECONDS);

    url = rpg.getTargetUris();
    name = remotePort.getName();

    final BatchSize batchSize = remotePort.getBatchSize();
    final int batchCount;
    final long batchBytes;
    final long batchMillis;
    if (batchSize == null) {
        batchCount = 1;
        batchBytes = 1L;
        batchMillis = 1L;
    } else {
        batchCount = batchSize.getCount() == null ? 1 : batchSize.getCount();
        batchBytes = batchSize.getSize() == null ? 1L : DataUnit.parseDataSize(batchSize.getSize(), DataUnit.B).longValue();
        batchMillis = batchSize.getDuration() == null ? 1L : FormatUtils.getTimeDuration(batchSize.getDuration(), TimeUnit.MILLISECONDS);
    }

    client = new SiteToSiteClient.Builder()
        .portName(remotePort.getName())
        .timeout(timeoutMillis, TimeUnit.MILLISECONDS)
        .requestBatchCount(batchCount)
        .requestBatchDuration(batchMillis, TimeUnit.MILLISECONDS)
        .requestBatchSize(batchBytes)
        .transportProtocol(SiteToSiteTransportProtocol.valueOf(rpg.getTransportProtocol()))
        .url(rpg.getTargetUris())
        .sslContext(sslContext)
        .useCompression(remotePort.isUseCompression())
        .eventReporter(EventReporter.NO_OP)
        .build();
}
 

private void findRemoteGroupRecursive(final VersionedProcessGroup group, final Map<String, VersionedRemoteProcessGroup> rpgs, final Map<String, VersionedRemoteGroupPort> ports) {
    for (final VersionedRemoteProcessGroup rpg : group.getRemoteProcessGroups()) {
        rpgs.put(rpg.getIdentifier(), rpg);

        rpg.getInputPorts().forEach(port -> ports.put(port.getIdentifier(), port));
        rpg.getOutputPorts().forEach(port -> ports.put(port.getIdentifier(), port));
    }
}