class StandardFDConnectionService extends BaseComponentService<VersionedConnection, FDConnection> implements com.cloudera.cem.efm.service.component.FDConnectionService {

    @Override
    public FDConnection update(final String flowId, final VersionedConnection requestComponentConfig) {
        final FDFlow currentFlow = getFlowOrNotFound(flowId);
        final VersionedProcessGroup flowContents = currentFlow.getFlowContent();

        // locate the connection component and ensure it exists
        final String componentId = requestComponentConfig.getIdentifier();
        final VersionedConnection existingConnection = ComponentUtils.getComponentOrNotFound(
                componentId, flowContents, VersionedProcessGroup::getConnections);

        // locate the connection group and ensure it exists
        final String connectionGroupId = existingConnection.getGroupIdentifier();
        final VersionedProcessGroup connectionGroup = ComponentUtils.getProcessGroupOrIllegalArgument(connectionGroupId, flowContents);

        final ConnectableComponent requestSource = requestComponentConfig.getSource();
        final ConnectableComponent existingSource = existingConnection.getSource();

        // if a source was provided then ensure it is the same source as the existing connection
        if (isNotNull(requestSource) && isNotNull(requestSource.getId()) && !requestSource.getId().equals(existingSource.getId())) {
            throw new IllegalStateException("Connection with id " + existingConnection.getIdentifier() + " has conflicting source id");
        }

        // If a processor specifies a non-null set of selected relationships, then it must not be empty
        if (existingSource.getType() == ConnectableComponentType.PROCESSOR
                && isNotNull(requestComponentConfig.getSelectedRelationships())
                && requestComponentConfig.getSelectedRelationships().isEmpty()) {
            throw new IllegalArgumentException("Selected relationships cannot be empty");
        }

        // validate the common request fields for create/update
        validateRequestConfig(requestComponentConfig);

        // determine if the destination changed
        final ConnectableComponent proposedDestination = requestComponentConfig.getDestination();
        if (proposedDestination != null) {
            if (proposedDestination.getId() == null) {
                throw new IllegalArgumentException("Proposed destination must contain an id");
            }

            if (proposedDestination.getType() == null) {
                throw new IllegalArgumentException("Proposed destination must contain a type");
            }

            if (proposedDestination.getType() == ConnectableComponentType.FUNNEL
                    && proposedDestination.getId().equals(existingSource.getId())) {
                throw new IllegalArgumentException("Connecting a funnel to itself is not allowed");
            }

            final ConnectableComponent currentDestination = existingConnection.getDestination();

            // handle remote input port differently
            if (ConnectableComponentType.REMOTE_OUTPUT_PORT == proposedDestination.getType()) {
                throw new IllegalArgumentException("Destination cannot be a remote output port");
            } else if (ConnectableComponentType.REMOTE_INPUT_PORT == proposedDestination.getType()) {

                final String proposedDestinationId = proposedDestination.getId();
                final String proposedDestinationGroupId = proposedDestination.getGroupId();

                try {
                    UUID.fromString(proposedDestinationId);
                } catch (Exception  e) {
                    throw new IllegalArgumentException("The ID of a remote input port must be a UUID");
                }

                // the group id must be specified
                if (proposedDestinationGroupId == null) {
                    throw new IllegalArgumentException("When the destination is a remote input port its group id is required.");
                }

                // if the current destination is a remote input port
                boolean isDifferentRemoteProcessGroup = false;
                if (currentDestination.getType() == ConnectableComponentType.REMOTE_INPUT_PORT) {
                    if (!proposedDestinationGroupId.equals(currentDestination.getGroupId())) {
                        isDifferentRemoteProcessGroup = true;
                    }
                }

                // if the destination is changing or the previous destination was a different remote process group
                if (!proposedDestinationId.equals(currentDestination.getId()) || isDifferentRemoteProcessGroup) {
                    // locate the remote process group or throw exception if not found
                    final VersionedRemoteProcessGroup proposedRemoteProcessGroup = connectionGroup.getRemoteProcessGroups().stream()
                            .filter(rpg -> rpg.getIdentifier().equals(proposedDestinationGroupId))
                            .findFirst()
                            .orElseThrow(() -> new IllegalArgumentException("Unable to find the specified remote process group for the destination"));

                    // locate an existing remote input port with a target id that matches the proposed destination id
                    final Optional<VersionedRemoteGroupPort> remoteInputPort = proposedRemoteProcessGroup.getInputPorts().stream()
                            .filter(p -> p.getTargetId().equals(proposedDestinationId))
                            .findFirst();

                    // if an existing port wasn't found then create one
                    final VersionedRemoteGroupPort newRemoteInputPort = new VersionedRemoteGroupPort();
                    if (!remoteInputPort.isPresent()) {
                        newRemoteInputPort.setIdentifier(UUID.randomUUID().toString());
                        newRemoteInputPort.setGroupIdentifier(connectionGroup.getIdentifier());
                        newRemoteInputPort.setRemoteGroupId(proposedDestinationGroupId);
                        newRemoteInputPort.setTargetId(proposedDestinationId);
                        newRemoteInputPort.setName(proposedDestinationId);
                        newRemoteInputPort.setComponentType(ComponentType.REMOTE_INPUT_PORT);
                    }

                    // determine if any other connections are still using the original remote input port, and if not remove it
                    if (!isDestStillUsed(existingConnection, currentDestination, connectionGroup)) {
                        removeRemoteInputPort(connectionGroup, currentDestination);
                    }

                    // add the port and update destination on the existing connection
                    proposedRemoteProcessGroup.getInputPorts().add(newRemoteInputPort);
                    proposedDestination.setName(proposedDestinationId);
                    existingConnection.setDestination(proposedDestination);
                }

            } else {

                // if there is a different destination id
                if (!proposedDestination.getId().equals(currentDestination.getId())) {
                    // if the destination connectable's group id has not been set, its inferred to be the current group
                    if (proposedDestination.getGroupId() == null) {
                        proposedDestination.setGroupId(existingConnection.getGroupIdentifier());
                    }

                    // locate the destination group and ensure it exists
                    final VersionedProcessGroup destinationGroup = ComponentUtils.getProcessGroupOrIllegalArgument(
                            proposedDestination.getGroupId(), currentFlow.getFlowContent());

                    // locate destination component in the destination group and ensure it exists
                    final Optional<VersionedComponent> newDestination = getVersionedComponentConnectable(
                            destinationGroup, proposedDestination);

                    if (!newDestination.isPresent()) {
                        throw new IllegalArgumentException("Unable to find the specified destination.");
                    }

                    // if the group of the new dest component is not the group the connection is in, then ensure
                    // the group is a child of the connection's group, and ensure the component is an input port

                    final String existingConnectionGroupId = existingConnection.getGroupIdentifier();
                    final String newDestinationGroupId = newDestination.get().getGroupIdentifier();

                    if (!existingConnectionGroupId.equals(newDestinationGroupId)) {
                        final ComponentType newDestinationComponentType = newDestination.get().getComponentType();
                        if (newDestinationComponentType != ComponentType.INPUT_PORT) {
                            throw new IllegalStateException("Cannot update connection because the destination is in " +
                                    "a different process group and is not an input port");
                        }

                        final boolean connectionGroupContainsDestGroup = containsChildGroup(connectionGroup, newDestinationGroupId);
                        if (!connectionGroupContainsDestGroup) {
                            throw new IllegalStateException("Cannot update connection because the new destination " +
                                    "group is not a child group of the connection group");
                        }
                    }

                    // update the existing connection's destination
                    existingConnection.setDestination(proposedDestination);
                }
            }
        }

        final FDConnection connection = updateComponent(currentFlow, requestComponentConfig, existingConnection, null);

        getValidationService().clearValidationErrors(connection.getComponentConfiguration().getSource().getId());
        getValidationService().clearValidationErrors(connection.getComponentConfiguration().getDestination().getId());

        return connection;
    }

    public  StandardFDConnectionService(final FDFlowManager flowManager, final ValidationService validationService);

}