/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.nifi.controller;

import org.apache.nifi.bundle.BundleCoordinate;
import org.apache.nifi.connectable.Connectable;
import org.apache.nifi.connectable.ConnectableType;
import org.apache.nifi.connectable.Connection;
import org.apache.nifi.connectable.Funnel;
import org.apache.nifi.connectable.Port;
import org.apache.nifi.connectable.Position;
import org.apache.nifi.connectable.Size;
import org.apache.nifi.controller.exception.ProcessorInstantiationException;
import org.apache.nifi.controller.flow.FlowManager;
import org.apache.nifi.controller.label.Label;
import org.apache.nifi.controller.queue.FlowFileQueue;
import org.apache.nifi.controller.queue.LoadBalanceStrategy;
import org.apache.nifi.controller.service.ControllerServiceNode;
import org.apache.nifi.flowfile.FlowFilePrioritizer;
import org.apache.nifi.groups.FlowFileConcurrency;
import org.apache.nifi.groups.FlowFileOutboundPolicy;
import org.apache.nifi.groups.ProcessGroup;
import org.apache.nifi.groups.RemoteProcessGroup;
import org.apache.nifi.groups.RemoteProcessGroupPortDescriptor;
import org.apache.nifi.logging.LogLevel;
import org.apache.nifi.nar.ExtensionDefinition;
import org.apache.nifi.nar.ExtensionManager;
import org.apache.nifi.parameter.ParameterContext;
import org.apache.nifi.processor.Processor;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.registry.flow.StandardVersionControlInformation;
import org.apache.nifi.registry.flow.VersionControlInformation;
import org.apache.nifi.remote.PublicPort;
import org.apache.nifi.remote.StandardRemoteProcessGroupPortDescriptor;
import org.apache.nifi.remote.protocol.SiteToSiteTransportProtocol;
import org.apache.nifi.scheduling.ExecutionNode;
import org.apache.nifi.scheduling.SchedulingStrategy;
import org.apache.nifi.util.BundleUtils;
import org.apache.nifi.util.SnippetUtils;
import org.apache.nifi.web.api.dto.BatchSettingsDTO;
import org.apache.nifi.web.api.dto.BundleDTO;
import org.apache.nifi.web.api.dto.ConnectableDTO;
import org.apache.nifi.web.api.dto.ConnectionDTO;
import org.apache.nifi.web.api.dto.ControllerServiceDTO;
import org.apache.nifi.web.api.dto.FlowSnippetDTO;
import org.apache.nifi.web.api.dto.FunnelDTO;
import org.apache.nifi.web.api.dto.LabelDTO;
import org.apache.nifi.web.api.dto.PortDTO;
import org.apache.nifi.web.api.dto.PositionDTO;
import org.apache.nifi.web.api.dto.ProcessGroupDTO;
import org.apache.nifi.web.api.dto.ProcessorConfigDTO;
import org.apache.nifi.web.api.dto.ProcessorDTO;
import org.apache.nifi.web.api.dto.RelationshipDTO;
import org.apache.nifi.web.api.dto.RemoteProcessGroupContentsDTO;
import org.apache.nifi.web.api.dto.RemoteProcessGroupDTO;
import org.apache.nifi.web.api.dto.RemoteProcessGroupPortDTO;
import org.apache.nifi.web.api.entity.ParameterContextReferenceEntity;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

public class StandardFlowSnippet implements FlowSnippet {

    private final FlowSnippetDTO dto;
    private final ExtensionManager extensionManager;

    public StandardFlowSnippet(final FlowSnippetDTO dto, final ExtensionManager extensionManager) {
        this.dto = dto;
        this.extensionManager = extensionManager;
    }

    public void validate(final ProcessGroup group) {
        // validate the names of Input Ports
        for (final PortDTO port : dto.getInputPorts()) {
            if (group.getInputPortByName(port.getName()) != null) {
                throw new IllegalStateException("One or more of the proposed Port names is not available in the process group");
            }
        }

        // validate the names of Output Ports
        for (final PortDTO port : dto.getOutputPorts()) {
            if (group.getOutputPortByName(port.getName()) != null) {
                throw new IllegalStateException("One or more of the proposed Port names is not available in the process group");
            }
        }

        verifyComponentTypesInSnippet();

        SnippetUtils.verifyNoVersionControlConflicts(dto, group);
    }

    public void verifyComponentTypesInSnippet() {
        final Map<String, Set<BundleCoordinate>> processorClasses = new HashMap<>();
        for (final ExtensionDefinition extensionDefinition : extensionManager.getExtensions(Processor.class)) {
            final String name = extensionDefinition.getImplementationClassName();
            processorClasses.put(name, extensionManager.getBundles(name).stream().map(bundle -> bundle.getBundleDetails().getCoordinate()).collect(Collectors.toSet()));
        }
        verifyProcessorsInSnippet(dto, processorClasses);

        final Map<String, Set<BundleCoordinate>> controllerServiceClasses = new HashMap<>();
        for (final ExtensionDefinition extensionDefinition : extensionManager.getExtensions(ControllerService.class)) {
            final String name = extensionDefinition.getImplementationClassName();
            controllerServiceClasses.put(name, extensionManager.getBundles(name).stream().map(bundle -> bundle.getBundleDetails().getCoordinate()).collect(Collectors.toSet()));
        }
        verifyControllerServicesInSnippet(dto, controllerServiceClasses);

        final Set<String> prioritizerClasses = new HashSet<>();
        for (final ExtensionDefinition extensionDefinition : extensionManager.getExtensions(FlowFilePrioritizer.class)) {
            final String name = extensionDefinition.getImplementationClassName();
            prioritizerClasses.add(name);
        }

        final Set<ConnectionDTO> allConns = new HashSet<>();
        allConns.addAll(dto.getConnections());
        for (final ProcessGroupDTO childGroup : dto.getProcessGroups()) {
            allConns.addAll(findAllConnections(childGroup));
        }

        for (final ConnectionDTO conn : allConns) {
            final List<String> prioritizers = conn.getPrioritizers();
            if (prioritizers != null) {
                for (final String prioritizer : prioritizers) {
                    if (!prioritizerClasses.contains(prioritizer)) {
                        throw new IllegalStateException("Invalid FlowFile Prioritizer Type: " + prioritizer);
                    }
                }
            }
        }
    }

    public void instantiate(final FlowManager flowManager, final ProcessGroup group) throws ProcessorInstantiationException {
        instantiate(flowManager, group, true);
    }


    /**
     * Recursively finds all ConnectionDTO's
     *
     * @param group group
     * @return connection dtos
     */
    private Set<ConnectionDTO> findAllConnections(final ProcessGroupDTO group) {
        final Set<ConnectionDTO> conns = new HashSet<>();
        conns.addAll(group.getContents().getConnections());

        for (final ProcessGroupDTO childGroup : group.getContents().getProcessGroups()) {
            conns.addAll(findAllConnections(childGroup));
        }
        return conns;
    }

    private void verifyControllerServicesInSnippet(final FlowSnippetDTO templateContents, final Map<String, Set<BundleCoordinate>> supportedTypes) {
        if (templateContents.getControllerServices() != null) {
            templateContents.getControllerServices().forEach(controllerService -> {
                if (supportedTypes.containsKey(controllerService.getType())) {
                    if (controllerService.getBundle() == null) {
                        throw new IllegalArgumentException("Controller Service bundle must be specified.");
                    }

                    verifyBundleInSnippet(controllerService.getBundle(), supportedTypes.get(controllerService.getType()));
                } else {
                    throw new IllegalStateException("Invalid Controller Service Type: " + controllerService.getType());
                }
            });
        }

        if (templateContents.getProcessGroups() != null) {
            templateContents.getProcessGroups().forEach(processGroup -> verifyControllerServicesInSnippet(processGroup.getContents(), supportedTypes));
        }
    }

    private void verifyBundleInSnippet(final BundleDTO requiredBundle, final Set<BundleCoordinate> supportedBundles) {
        final BundleCoordinate requiredCoordinate = new BundleCoordinate(requiredBundle.getGroup(), requiredBundle.getArtifact(), requiredBundle.getVersion());
        if (!supportedBundles.contains(requiredCoordinate)) {
            throw new IllegalStateException("Unsupported bundle: " + requiredCoordinate);
        }
    }

    private void verifyProcessorsInSnippet(final FlowSnippetDTO templateContents, final Map<String, Set<BundleCoordinate>> supportedTypes) {
        if (templateContents.getProcessors() != null) {
            templateContents.getProcessors().forEach(processor -> {
                if (processor.getBundle() == null) {
                    throw new IllegalArgumentException("Processor bundle must be specified.");
                }

                if (supportedTypes.containsKey(processor.getType())) {
                    verifyBundleInSnippet(processor.getBundle(), supportedTypes.get(processor.getType()));
                } else {
                    throw new IllegalStateException("Invalid Processor Type: " + processor.getType());
                }
            });
        }

        if (templateContents.getProcessGroups() != null) {
            templateContents.getProcessGroups().forEach(processGroup -> verifyProcessorsInSnippet(processGroup.getContents(), supportedTypes));
        }
    }


    public void instantiate(final FlowManager flowManager, final ProcessGroup group, final boolean topLevel) {
        //
        // Instantiate Controller Services
        //
        final List<ControllerServiceNode> serviceNodes = new ArrayList<>();
        try {
            for (final ControllerServiceDTO controllerServiceDTO : dto.getControllerServices()) {
                final BundleCoordinate bundleCoordinate = BundleUtils.getBundle(extensionManager, controllerServiceDTO.getType(), controllerServiceDTO.getBundle());
                final ControllerServiceNode serviceNode = flowManager.createControllerService(controllerServiceDTO.getType(), controllerServiceDTO.getId(),
                    bundleCoordinate, Collections.emptySet(), true,true);

                serviceNode.pauseValidationTrigger();
                serviceNodes.add(serviceNode);

                serviceNode.setAnnotationData(controllerServiceDTO.getAnnotationData());
                serviceNode.setComments(controllerServiceDTO.getComments());
                serviceNode.setName(controllerServiceDTO.getName());
                if (!topLevel) {
                    serviceNode.setVersionedComponentId(controllerServiceDTO.getVersionedComponentId());
                }

                group.addControllerService(serviceNode);
            }

            // configure controller services. We do this after creating all of them in case 1 service
            // references another service.
            for (final ControllerServiceDTO controllerServiceDTO : dto.getControllerServices()) {
                final String serviceId = controllerServiceDTO.getId();
                final ControllerServiceNode serviceNode = flowManager.getControllerServiceNode(serviceId);
                serviceNode.setProperties(controllerServiceDTO.getProperties());
            }
        } finally {
            serviceNodes.forEach(ControllerServiceNode::resumeValidationTrigger);
        }

        //
        // Instantiate the labels
        //
        for (final LabelDTO labelDTO : dto.getLabels()) {
            final Label label = flowManager.createLabel(labelDTO.getId(), labelDTO.getLabel());
            label.setPosition(toPosition(labelDTO.getPosition()));
            if (labelDTO.getWidth() != null && labelDTO.getHeight() != null) {
                label.setSize(new Size(labelDTO.getWidth(), labelDTO.getHeight()));
            }

            label.setStyle(labelDTO.getStyle());
            if (!topLevel) {
                label.setVersionedComponentId(labelDTO.getVersionedComponentId());
            }

            group.addLabel(label);
        }

        // Instantiate the funnels
        for (final FunnelDTO funnelDTO : dto.getFunnels()) {
            final Funnel funnel = flowManager.createFunnel(funnelDTO.getId());
            funnel.setPosition(toPosition(funnelDTO.getPosition()));
            if (!topLevel) {
                funnel.setVersionedComponentId(funnelDTO.getVersionedComponentId());
            }

            group.addFunnel(funnel);
        }

        //
        // Instantiate Input Ports & Output Ports
        //
        for (final PortDTO portDTO : dto.getInputPorts()) {
            final Port inputPort;
            if (group.isRootGroup() || Boolean.TRUE.equals(portDTO.getAllowRemoteAccess())) {
                final String portName = generatePublicInputPortName(flowManager, portDTO.getName());
                inputPort = flowManager.createPublicInputPort(portDTO.getId(), portName);
                if (portDTO.getGroupAccessControl() != null) {
                    ((PublicPort) inputPort).setGroupAccessControl(portDTO.getGroupAccessControl());
                }
                if (portDTO.getUserAccessControl() != null) {
                    ((PublicPort) inputPort).setUserAccessControl(portDTO.getUserAccessControl());
                }
            } else {
                inputPort = flowManager.createLocalInputPort(portDTO.getId(), portDTO.getName());
            }

            if (!topLevel) {
                inputPort.setVersionedComponentId(portDTO.getVersionedComponentId());
            }
            inputPort.setPosition(toPosition(portDTO.getPosition()));
            inputPort.setProcessGroup(group);
            inputPort.setMaxConcurrentTasks(portDTO.getConcurrentlySchedulableTaskCount());
            inputPort.setComments(portDTO.getComments());
            if (portDTO.getState().equals(ScheduledState.DISABLED.toString())) {
                inputPort.disable();
            }
            group.addInputPort(inputPort);
        }

        for (final PortDTO portDTO : dto.getOutputPorts()) {
            final Port outputPort;
            if (group.isRootGroup() || Boolean.TRUE.equals(portDTO.getAllowRemoteAccess())) {
                final String portName = generatePublicOutputPortName(flowManager, portDTO.getName());
                outputPort = flowManager.createPublicOutputPort(portDTO.getId(), portName);
                if (portDTO.getGroupAccessControl() != null) {
                    ((PublicPort) outputPort).setGroupAccessControl(portDTO.getGroupAccessControl());
                }
                if (portDTO.getUserAccessControl() != null) {
                    ((PublicPort) outputPort).setUserAccessControl(portDTO.getUserAccessControl());
                }
            } else {
                outputPort = flowManager.createLocalOutputPort(portDTO.getId(), portDTO.getName());
            }

            if (!topLevel) {
                outputPort.setVersionedComponentId(portDTO.getVersionedComponentId());
            }
            outputPort.setPosition(toPosition(portDTO.getPosition()));
            outputPort.setProcessGroup(group);
            outputPort.setMaxConcurrentTasks(portDTO.getConcurrentlySchedulableTaskCount());
            outputPort.setComments(portDTO.getComments());
            if (portDTO.getState().equals(ScheduledState.DISABLED.toString())) {
                outputPort.disable();
            }
            group.addOutputPort(outputPort);
        }

        //
        // Instantiate the processors
        //
        for (final ProcessorDTO processorDTO : dto.getProcessors()) {
            final BundleCoordinate bundleCoordinate = BundleUtils.getBundle(extensionManager, processorDTO.getType(), processorDTO.getBundle());
            final ProcessorNode procNode = flowManager.createProcessor(processorDTO.getType(), processorDTO.getId(), bundleCoordinate);
            procNode.pauseValidationTrigger();

            try {
                procNode.setPosition(toPosition(processorDTO.getPosition()));
                procNode.setProcessGroup(group);
                if (!topLevel) {
                    procNode.setVersionedComponentId(processorDTO.getVersionedComponentId());
                }

                final ProcessorConfigDTO config = processorDTO.getConfig();
                procNode.setComments(config.getComments());
                if (config.isLossTolerant() != null) {
                    procNode.setLossTolerant(config.isLossTolerant());
                }
                procNode.setName(processorDTO.getName());

                procNode.setYieldPeriod(config.getYieldDuration());
                procNode.setPenalizationPeriod(config.getPenaltyDuration());
                procNode.setBulletinLevel(LogLevel.valueOf(config.getBulletinLevel()));
                procNode.setAnnotationData(config.getAnnotationData());
                procNode.setStyle(processorDTO.getStyle());

                if (config.getRunDurationMillis() != null) {
                    procNode.setRunDuration(config.getRunDurationMillis(), TimeUnit.MILLISECONDS);
                }

                if (config.getSchedulingStrategy() != null) {
                    procNode.setSchedulingStrategy(SchedulingStrategy.valueOf(config.getSchedulingStrategy()));
                }

                if (config.getExecutionNode() != null) {
                    procNode.setExecutionNode(ExecutionNode.valueOf(config.getExecutionNode()));
                }

                if (processorDTO.getState().equals(ScheduledState.DISABLED.toString())) {
                    procNode.disable();
                }

                // ensure that the scheduling strategy is set prior to these values
                procNode.setMaxConcurrentTasks(config.getConcurrentlySchedulableTaskCount());
                procNode.setScheduldingPeriod(config.getSchedulingPeriod());

                final Set<Relationship> relationships = new HashSet<>();
                if (processorDTO.getRelationships() != null) {
                    for (final RelationshipDTO rel : processorDTO.getRelationships()) {
                        if (rel.isAutoTerminate()) {
                            relationships.add(procNode.getRelationship(rel.getName()));
                        }
                    }
                    procNode.setAutoTerminatedRelationships(relationships);
                }

                if (config.getProperties() != null) {
                    procNode.setProperties(config.getProperties());
                }

                group.addProcessor(procNode);
            } finally {
                procNode.resumeValidationTrigger();
            }
        }

        //
        // Instantiate Remote Process Groups
        //
        for (final RemoteProcessGroupDTO remoteGroupDTO : dto.getRemoteProcessGroups()) {
            final RemoteProcessGroup remoteGroup = flowManager.createRemoteProcessGroup(remoteGroupDTO.getId(), remoteGroupDTO.getTargetUris());
            remoteGroup.setComments(remoteGroupDTO.getComments());
            remoteGroup.setPosition(toPosition(remoteGroupDTO.getPosition()));
            remoteGroup.setCommunicationsTimeout(remoteGroupDTO.getCommunicationsTimeout());
            remoteGroup.setYieldDuration(remoteGroupDTO.getYieldDuration());
            if (!topLevel) {
                remoteGroup.setVersionedComponentId(remoteGroupDTO.getVersionedComponentId());
            }

            if (remoteGroupDTO.getTransportProtocol() == null) {
                remoteGroup.setTransportProtocol(SiteToSiteTransportProtocol.RAW);
            } else {
                remoteGroup.setTransportProtocol(SiteToSiteTransportProtocol.valueOf(remoteGroupDTO.getTransportProtocol()));
            }

            remoteGroup.setProxyHost(remoteGroupDTO.getProxyHost());
            remoteGroup.setProxyPort(remoteGroupDTO.getProxyPort());
            remoteGroup.setProxyUser(remoteGroupDTO.getProxyUser());
            remoteGroup.setProxyPassword(remoteGroupDTO.getProxyPassword());
            remoteGroup.setProcessGroup(group);

            // set the input/output ports
            if (remoteGroupDTO.getContents() != null) {
                final RemoteProcessGroupContentsDTO contents = remoteGroupDTO.getContents();

                // ensure there are input ports
                if (contents.getInputPorts() != null) {
                    remoteGroup.setInputPorts(convertRemotePort(contents.getInputPorts()), false);
                }

                // ensure there are output ports
                if (contents.getOutputPorts() != null) {
                    remoteGroup.setOutputPorts(convertRemotePort(contents.getOutputPorts()), false);
                }
            }

            group.addRemoteProcessGroup(remoteGroup);
        }

        //
        // Instantiate ProcessGroups
        //
        for (final ProcessGroupDTO groupDTO : dto.getProcessGroups()) {
            final ProcessGroup childGroup = flowManager.createProcessGroup(groupDTO.getId());
            childGroup.setParent(group);
            childGroup.setPosition(toPosition(groupDTO.getPosition()));
            childGroup.setComments(groupDTO.getComments());
            childGroup.setName(groupDTO.getName());

            final String flowfileConcurrentName = groupDTO.getFlowfileConcurrency();
            if (flowfileConcurrentName != null) {
                childGroup.setFlowFileConcurrency(FlowFileConcurrency.valueOf(flowfileConcurrentName));
            }

            final String outboundPolicyName = groupDTO.getFlowfileOutboundPolicy();
            if (outboundPolicyName != null) {
                childGroup.setFlowFileOutboundPolicy(FlowFileOutboundPolicy.valueOf(outboundPolicyName));
            }

            if (groupDTO.getVariables() != null) {
                childGroup.setVariables(groupDTO.getVariables());
            }

            final ParameterContextReferenceEntity parameterContextReference = groupDTO.getParameterContext();
            if (parameterContextReference != null) {
                final ParameterContext parameterContext = flowManager.getParameterContextManager().getParameterContext(parameterContextReference.getId());
                if (parameterContext != null) {
                    childGroup.setParameterContext(parameterContext);
                }
            }

            final String defaultFlowFileExpiration = groupDTO.getDefaultFlowFileExpiration();
            if (defaultFlowFileExpiration != null) {
                childGroup.setDefaultFlowFileExpiration(defaultFlowFileExpiration);
            }

            final Long defaultBackPressureObjectThreshold = groupDTO.getDefaultBackPressureObjectThreshold();
            if (defaultBackPressureObjectThreshold != null) {
                childGroup.setDefaultBackPressureObjectThreshold(defaultBackPressureObjectThreshold);
            }

            final String defaultBackPressureDataSizeThreshold = groupDTO.getDefaultBackPressureDataSizeThreshold();
            if (defaultBackPressureDataSizeThreshold != null) {
                childGroup.setDefaultBackPressureDataSizeThreshold(defaultBackPressureDataSizeThreshold);
            }

            // If this Process Group is 'top level' then we do not set versioned component ID's.
            // We do this only if this component is the child of a Versioned Component.
            if (!topLevel) {
                childGroup.setVersionedComponentId(groupDTO.getVersionedComponentId());
            }

            group.addProcessGroup(childGroup);

            final FlowSnippetDTO contents = groupDTO.getContents();

            // we want this to be recursive, so we will create a new template that contains only
            // the contents of this child group and recursively call ourselves.
            final FlowSnippetDTO childTemplateDTO = new FlowSnippetDTO();
            childTemplateDTO.setConnections(contents.getConnections());
            childTemplateDTO.setInputPorts(contents.getInputPorts());
            childTemplateDTO.setLabels(contents.getLabels());
            childTemplateDTO.setOutputPorts(contents.getOutputPorts());
            childTemplateDTO.setProcessGroups(contents.getProcessGroups());
            childTemplateDTO.setProcessors(contents.getProcessors());
            childTemplateDTO.setFunnels(contents.getFunnels());
            childTemplateDTO.setRemoteProcessGroups(contents.getRemoteProcessGroups());
            childTemplateDTO.setControllerServices(contents.getControllerServices());

            final StandardFlowSnippet childSnippet = new StandardFlowSnippet(childTemplateDTO, extensionManager);
            childSnippet.instantiate(flowManager, childGroup, false);

            if (groupDTO.getVersionControlInformation() != null) {
                final VersionControlInformation vci = StandardVersionControlInformation.Builder
                    .fromDto(groupDTO.getVersionControlInformation())
                    .build();
                childGroup.setVersionControlInformation(vci, Collections.emptyMap());
            }
        }

        //
        // Instantiate Connections
        //
        for (final ConnectionDTO connectionDTO : dto.getConnections()) {
            final ConnectableDTO sourceDTO = connectionDTO.getSource();
            final ConnectableDTO destinationDTO = connectionDTO.getDestination();
            final Connectable source;
            final Connectable destination;

            // Locate the source and destination connectable. If this is a remote port we need to locate the remote process groups. Otherwise, we need to
            // find the connectable given its parent group.
            //
            // NOTE: (getConnectable returns ANY connectable, when the parent is not this group only input ports or output ports should be returned. If something
            // other than a port is returned, an exception will be thrown when adding the connection below.)

            // See if the source connectable is a remote port
            if (ConnectableType.REMOTE_OUTPUT_PORT.name().equals(sourceDTO.getType())) {
                final RemoteProcessGroup remoteGroup = group.getRemoteProcessGroup(sourceDTO.getGroupId());
                source = remoteGroup.getOutputPort(sourceDTO.getId());
            } else {
                final ProcessGroup sourceGroup = getConnectableParent(group, sourceDTO.getGroupId(), flowManager);
                source = sourceGroup.getConnectable(sourceDTO.getId());
            }

            // see if the destination connectable is a remote port
            if (ConnectableType.REMOTE_INPUT_PORT.name().equals(destinationDTO.getType())) {
                final RemoteProcessGroup remoteGroup = group.getRemoteProcessGroup(destinationDTO.getGroupId());
                destination = remoteGroup.getInputPort(destinationDTO.getId());
            } else {
                final ProcessGroup destinationGroup = getConnectableParent(group, destinationDTO.getGroupId(), flowManager);
                destination = destinationGroup.getConnectable(destinationDTO.getId());
            }

            // determine the selection relationships for this connection
            final Set<String> relationships = new HashSet<>();
            if (connectionDTO.getSelectedRelationships() != null) {
                relationships.addAll(connectionDTO.getSelectedRelationships());
            }

            final Connection connection = flowManager.createConnection(connectionDTO.getId(), connectionDTO.getName(), source, destination, relationships);
            if (!topLevel) {
                connection.setVersionedComponentId(connectionDTO.getVersionedComponentId());
            }

            if (connectionDTO.getBends() != null) {
                final List<Position> bendPoints = new ArrayList<>();
                for (final PositionDTO bend : connectionDTO.getBends()) {
                    bendPoints.add(new Position(bend.getX(), bend.getY()));
                }
                connection.setBendPoints(bendPoints);
            }

            final FlowFileQueue queue = connection.getFlowFileQueue();
            queue.setBackPressureDataSizeThreshold(connectionDTO.getBackPressureDataSizeThreshold());
            queue.setBackPressureObjectThreshold(connectionDTO.getBackPressureObjectThreshold());
            queue.setFlowFileExpiration(connectionDTO.getFlowFileExpiration());

            final List<String> prioritizers = connectionDTO.getPrioritizers();
            if (prioritizers != null) {
                final List<String> newPrioritizersClasses = new ArrayList<>(prioritizers);
                final List<FlowFilePrioritizer> newPrioritizers = new ArrayList<>();
                for (final String className : newPrioritizersClasses) {
                    try {
                        newPrioritizers.add(flowManager.createPrioritizer(className));
                    } catch (final ClassNotFoundException | InstantiationException | IllegalAccessException e) {
                        throw new IllegalArgumentException("Unable to set prioritizer " + className + ": " + e);
                    }
                }
                queue.setPriorities(newPrioritizers);
            }

            final String loadBalanceStrategyName = connectionDTO.getLoadBalanceStrategy();
            if (loadBalanceStrategyName != null) {
                final LoadBalanceStrategy loadBalanceStrategy = LoadBalanceStrategy.valueOf(loadBalanceStrategyName);
                final String partitioningAttribute = connectionDTO.getLoadBalancePartitionAttribute();
                queue.setLoadBalanceStrategy(loadBalanceStrategy, partitioningAttribute);
            }

            connection.setProcessGroup(group);
            group.addConnection(connection);
        }
    }

    private String generatePublicInputPortName(final FlowManager flowManager, final String proposedName) {
        final Optional<Port> existingPort = flowManager.getPublicInputPort(proposedName);
        if (existingPort.isPresent()) {
            return generatePublicInputPortName(flowManager, "Copy of " + proposedName);
        } else {
            return proposedName;
        }
    }

    private String generatePublicOutputPortName(final FlowManager flowManager, final String proposedName) {
        final Optional<Port> existingPort = flowManager.getPublicOutputPort(proposedName);
        if (existingPort.isPresent()) {
            return generatePublicOutputPortName(flowManager, "Copy of " + proposedName);
        } else {
            return proposedName;
        }
    }

    private ProcessGroup getConnectableParent(final ProcessGroup group, final String parentGroupId, final FlowManager flowManager) {
        if (flowManager.areGroupsSame(group.getIdentifier(), parentGroupId)) {
            return group;
        } else {
            return group.getProcessGroup(parentGroupId);
        }
    }


    private Position toPosition(final PositionDTO dto) {
        return new Position(dto.getX(), dto.getY());
    }

    /**
     * Converts a set of ports into a set of remote process group ports.
     *
     * @param ports ports
     * @return group descriptors
     */
    private Set<RemoteProcessGroupPortDescriptor> convertRemotePort(final Set<RemoteProcessGroupPortDTO> ports) {
        Set<RemoteProcessGroupPortDescriptor> remotePorts = null;
        if (ports != null) {
            remotePorts = new LinkedHashSet<>(ports.size());
            for (final RemoteProcessGroupPortDTO port : ports) {
                final StandardRemoteProcessGroupPortDescriptor descriptor = new StandardRemoteProcessGroupPortDescriptor();
                descriptor.setId(port.getId());
                descriptor.setVersionedComponentId(port.getVersionedComponentId());
                descriptor.setTargetId(port.getTargetId());
                descriptor.setName(port.getName());
                descriptor.setComments(port.getComments());
                descriptor.setTargetRunning(port.isTargetRunning());
                descriptor.setConnected(port.isConnected());
                descriptor.setConcurrentlySchedulableTaskCount(port.getConcurrentlySchedulableTaskCount());
                descriptor.setTransmitting(port.isTransmitting());
                descriptor.setUseCompression(port.getUseCompression());
                final BatchSettingsDTO batchSettings = port.getBatchSettings();
                if (batchSettings != null) {
                    descriptor.setBatchCount(batchSettings.getCount());
                    descriptor.setBatchSize(batchSettings.getSize());
                    descriptor.setBatchDuration(batchSettings.getDuration());
                }
                remotePorts.add(descriptor);
            }
        }
        return remotePorts;
    }
}