/**
 * Copyright (C) 2006-2018 Talend Inc. - www.talend.com
 *
 * Licensed 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.talend.sdk.component.studio.model.connector;

import static java.util.stream.Collectors.toList;
import static org.talend.core.model.process.EConnectionType.FLOW_MAIN;
import static org.talend.core.model.process.EConnectionType.FLOW_MERGE;
import static org.talend.core.model.process.EConnectionType.FLOW_REF;
import static org.talend.core.model.process.EConnectionType.ITERATE;
import static org.talend.core.model.process.EConnectionType.REJECT;

import java.util.ArrayList;
import java.util.List;
import java.util.Optional;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Stream;

import org.eclipse.swt.graphics.RGB;
import org.talend.core.model.process.EConnectionType;
import org.talend.core.model.process.INode;
import org.talend.core.model.process.INodeConnector;
import org.talend.sdk.component.server.front.model.ComponentDetail;

/**
 * Creates connectors for Processor component
 */
class ProcessorConnectorCreator extends AbstractConnectorCreator {

    protected ProcessorConnectorCreator(final ComponentDetail detail, final INode node) {
        super(detail, node);
    }

    /**
     * Processor component may have both incoming and outgoing connections
     */
    @Override
    protected List<INodeConnector> createMainConnectors() {
        AtomicInteger rowCount = new AtomicInteger(0);
        detail.getInputFlows().stream().forEach(input -> {
            if (FLOW_MAIN.equals(getType(input))) {
                rowCount.incrementAndGet();
            }
        });
        final String mainConnectorName = MAIN_CONNECTOR_NAME;
        final INodeConnector main = createConnector(EConnectionType.FLOW_MAIN, mainConnectorName, node);
        main.setMaxLinkInput(rowCount.get());
        if (main instanceof TaCoKitNodeConnector) {
            ((TaCoKitNodeConnector) main).setInput(false);
        }
        main.addConnectionProperty(FLOW_MAIN, FLOW_MAIN.getRGB(), FLOW_MAIN.getDefaultLineStyle());
        main.addConnectionProperty(FLOW_REF, FLOW_MAIN.getRGB(), FLOW_MAIN.getDefaultLineStyle());
        main.addConnectionProperty(FLOW_MERGE, FLOW_MAIN.getRGB(), FLOW_MAIN.getDefaultLineStyle());
        existingTypes.add(EConnectionType.FLOW_MAIN);

        List<INodeConnector> connectors = new ArrayList<>();
        connectors.add(main);

        List<INodeConnector> generatedConnectors = Stream
                .concat(detail
                        .getInputFlows()
                        .stream() //
                        .filter(input -> {
                            if (!FLOW_MAIN.equals(getType(input))) {
                                return false;
                            }
                            if (mainConnectorName.equalsIgnoreCase(getName(input))) {
                                return false;
                            }
                            return true;
                        }) //
                        .map(input -> { //
                            String name = getName(input);
                            final INodeConnector inputConnector = createConnector(getType(input), name, node);
                            inputConnector.setMaxLinkInput(1);
                            if (inputConnector instanceof TaCoKitNodeConnector) {
                                ((TaCoKitNodeConnector) inputConnector).setInput(true);
                            }
                            inputConnector.addConnectionProperty(FLOW_MAIN, FLOW_MAIN.getRGB(),
                                    FLOW_MAIN.getDefaultLineStyle());
                            inputConnector.addConnectionProperty(FLOW_REF, FLOW_MAIN.getRGB(),
                                    FLOW_MAIN.getDefaultLineStyle());
                            inputConnector.addConnectionProperty(FLOW_MERGE, FLOW_MAIN.getRGB(),
                                    FLOW_MAIN.getDefaultLineStyle());
                            existingTypes.add(getType(input));
                            return inputConnector;
                        }), detail
                                .getOutputFlows()
                                .stream() //
                                .filter(output -> FLOW_MAIN.equals(getType(output))) //
                                .map(output -> { //
                                    final INodeConnector outputConnector =
                                            createConnector(getType(output), getName(output), node);
                                    outputConnector.setMaxLinkOutput(1);
                                    outputConnector.addConnectionProperty(FLOW_REF, FLOW_REF.getRGB(),
                                            FLOW_REF.getDefaultLineStyle());
                                    outputConnector.addConnectionProperty(FLOW_MERGE, FLOW_MERGE.getRGB(),
                                            FLOW_MERGE.getDefaultLineStyle());
                                    existingTypes.add(getType(output));
                                    return outputConnector;
                                }))
                .collect(toList());

        connectors.addAll(generatedConnectors);

        return connectors;
    }

    /**
     * Processor component may have outgoing Reject connections
     */
    @Override
    protected Optional<INodeConnector> createRejectConnector() {
        return detail
                .getOutputFlows()
                .stream() //
                .filter(output -> REJECT.equals(getType(output))) //
                .findFirst() //
                .map(output -> { //
                    INodeConnector reject = createConnector(getType(output), getName(output), node, FLOW_MAIN);
                    reject.setMaxLinkOutput(1);
                    reject.addConnectionProperty(EConnectionType.FLOW_MAIN, new RGB(255, 0, 0), 2);
                    reject.getConnectionProperty(EConnectionType.FLOW_MAIN).setRGB(new RGB(255, 0, 0));
                    existingTypes.add(getType(output));
                    return reject;
                }); //
    }

    /**
     * Processor component has 0 incoming and infinite outgoing iterate connections
     */
    @Override
    protected INodeConnector createIterateConnector() {
        INodeConnector iterate = createConnector(ITERATE, ITERATE.getName(), node);
        iterate.setMinLinkInput(0);
        iterate.setMaxLinkInput(0);
        iterate.setMinLinkOutput(0);
        iterate.setMaxLinkOutput(-1);
        existingTypes.add(ITERATE);
        return iterate;
    }

}

@@ -1148,16 +1148,66 @@
 tional;%0A
+import java.util.concurrent.atomic.AtomicBoolean;%0A
 import j
@@ -1248,16 +1248,16 @@
 nteger;%0A
-
 import j
@@ -2002,24 +2002,101 @@
 Integer(0);%0A
+        AtomicBoolean isDefaultConnectionPresent = new AtomicBoolean(false);%0A
         deta
@@ -2234,24 +2234,166 @@
 ntAndGet();%0A
+                if (FLOW_MAIN.getName().equals(getName(input))) %7B%0A                    isDefaultConnectionPresent.set(true);%0A                %7D%0A
             
@@ -2722,21 +2722,48 @@
 etInput(
-false
+isDefaultConnectionPresent.get()
 );%0A     

fix lookup connector naming