/*
# Licensed Materials - Property of IBM
# Copyright IBM Corp. 2015  
 */
package com.ibm.streamsx.topology.generator.spl;

import static com.ibm.streamsx.topology.builder.BVirtualMarker.END_PARALLEL;
import static com.ibm.streamsx.topology.builder.BVirtualMarker.ISOLATE;
import static com.ibm.streamsx.topology.builder.BVirtualMarker.PARALLEL;
import static com.ibm.streamsx.topology.generator.spl.GraphUtilities.addBefore;
import static com.ibm.streamsx.topology.generator.spl.GraphUtilities.findOperatorByKind;

import java.util.Arrays;
import java.util.List;

import com.google.gson.JsonArray;
import com.google.gson.JsonObject;
import com.ibm.streamsx.topology.builder.BVirtualMarker;

/**
 * Preprocessor modifies the passed in JSON to perform
 * logical graph transformations.
 */
class Preprocessor {
    
    private final SPLGenerator generator;
    private final JsonObject graph;
    
    Preprocessor(SPLGenerator generator, JsonObject graph) {
        this.generator = generator;
        this.graph = graph;
    }
    
    void preprocess() {
        
        GraphValidation graphValidationProcess = new GraphValidation();
        graphValidationProcess.validateGraph(graph);
        
        isolateParalleRegions();
        
        PEPlacement pePlacementPreprocess = new PEPlacement(generator, graph);

        // The hash adder operators need to be relocated to enable directly 
	// adjacent parallel regions
        // TODO: renable adjacent parallel regions optimization
        //relocateHashAdders();
        
        pePlacementPreprocess.tagIsolationRegions();
        pePlacementPreprocess.tagLowLatencyRegions();
        
        
        
        ThreadingModel.preProcessThreadedPorts(graph);
        
        removeRemainingVirtualMarkers();
        
        AutonomousRegions.preprocessAutonomousRegions(graph);
        
        pePlacementPreprocess.resolveColocationTags();

        // Optimize phase.
        new Optimizer(graph).optimize();       
    }
       
    /**
     * Isolate parallel regions to ensure that
     * we get parallelism through multiple PEs
     * (with the ability to have those PEs be distributed
     * across multiple hosts).
     * 
     * For 4.2 and later we achieve this using deploymentConfig
     * unless there are isolated regions.
     * 
     * Pre-4.2 we insert isolates prior to region and after the region.
     */
    private void isolateParalleRegions() {
        boolean needExplicitIsolates = !generator.versionAtLeast(4, 2);
        
        // TODO 4.2 checking
        
        if (!needExplicitIsolates)
            return;
        
        // Add isolate before the parallel and end parallel markers
        List<JsonObject> parallelOperators = findOperatorByKind(PARALLEL, graph);  
        parallelOperators.addAll(findOperatorByKind(END_PARALLEL, graph));
        for (JsonObject po : parallelOperators) {
            String schema = po.get("inputs").getAsJsonArray().get(0).getAsJsonObject().get("type").getAsString();
                        
            addBefore(po, newMarker(schema, ISOLATE), graph);         
        }       
    }
    
    
    private int ppMarkerCount;
    /**
     * Create a new marker operator that can be inserted into
     * the graph using addBefore.
     */
    private JsonObject newMarker(String schema, BVirtualMarker marker) {
        JsonObject op = new JsonObject();
        op.addProperty("marker", true);
        op.addProperty("kind", marker.kind());
        String name = "$$PreprocessorMarker_" + ppMarkerCount++;
        op.addProperty("name", name);
        
        {
            JsonArray inputs = new JsonArray();
            op.add("inputs", inputs);
            JsonObject input = new JsonObject();
            inputs.add(input);            
            input.addProperty("index", 0);
            input.addProperty("name", name + "_IN");
            input.addProperty("type", schema);
            input.add("connections", new JsonArray());
            
        }
        {
            JsonArray outputs = new JsonArray();
            op.add("outputs", outputs);
            JsonObject output = new JsonObject();
            outputs.add(output);
            output.addProperty("index", 0);
            output.addProperty("type", schema);
            output.add("connections", new JsonArray());
            output.addProperty("name", name + "_IN");
        }
        return op;
    }
    
    private void removeRemainingVirtualMarkers(){
        for (BVirtualMarker marker : Arrays.asList(BVirtualMarker.UNION, BVirtualMarker.PENDING)) {
            List<JsonObject> unionOps = GraphUtilities.findOperatorByKind(marker, graph);
            GraphUtilities.removeOperators(unionOps, graph);
        }
    }
}

@@ -122,400 +122,8 @@
 l;%0A%0A
-import static com.ibm.streamsx.topology.builder.BVirtualMarker.END_PARALLEL;%0Aimport static com.ibm.streamsx.topology.builder.BVirtualMarker.ISOLATE;%0Aimport static com.ibm.streamsx.topology.builder.BVirtualMarker.PARALLEL;%0Aimport static com.ibm.streamsx.topology.generator.spl.GraphUtilities.addBefore;%0Aimport static com.ibm.streamsx.topology.generator.spl.GraphUtilities.findOperatorByKind;%0A%0A
 impo
@@ -143,16 +143,16 @@
 Arrays;%0A
+
 import j
@@ -806,42 +806,8 @@
     
-%0A        isolateParalleRegions();%0A
     
@@ -1505,16 +1505,16 @@
  phase.%0A
+
         
@@ -1563,2440 +1563,8 @@
   %7D%0A
-       %0A    /**%0A     * Isolate parallel regions to ensure that%0A     * we get parallelism through multiple PEs%0A     * (with the ability to have those PEs be distributed%0A     * across multiple hosts).%0A     * %0A     * For 4.2 and later we achieve this using deploymentConfig%0A     * unless there are isolated regions.%0A     * %0A     * Pre-4.2 we insert isolates prior to region and after the region.%0A     */%0A    private void isolateParalleRegions() %7B%0A        boolean needExplicitIsolates = !generator.versionAtLeast(4, 2);%0A        %0A        // TODO 4.2 checking%0A        %0A        if (!needExplicitIsolates)%0A            return;%0A        %0A        // Add isolate before the parallel and end parallel markers%0A        List%3CJsonObject%3E parallelOperators = findOperatorByKind(PARALLEL, graph);  %0A        parallelOperators.addAll(findOperatorByKind(END_PARALLEL, graph));%0A        for (JsonObject po : parallelOperators) %7B%0A            String schema = po.get(%22inputs%22).getAsJsonArray().get(0).getAsJsonObject().get(%22type%22).getAsString();%0A                        %0A            addBefore(po, newMarker(schema, ISOLATE), graph);         %0A        %7D       %0A    %7D%0A    %0A    %0A    private int ppMarkerCount;%0A    /**%0A     * Create a new marker operator that can be inserted into%0A     * the graph using addBefore.%0A     */%0A    private JsonObject newMarker(String schema, BVirtualMarker marker) %7B%0A        JsonObject op = new JsonObject();%0A        op.addProperty(%22marker%22, true);%0A        op.addProperty(%22kind%22, marker.kind());%0A        String name = %22$$PreprocessorMarker_%22 + ppMarkerCount++;%0A        op.addProperty(%22name%22, name);%0A        %0A        %7B%0A            JsonArray inputs = new JsonArray();%0A            op.add(%22inputs%22, inputs);%0A            JsonObject input = new JsonObject();%0A            inputs.add(input);            %0A            input.addProperty(%22index%22, 0);%0A            input.addProperty(%22name%22, name + %22_IN%22);%0A            input.addProperty(%22type%22, schema);%0A            input.add(%22connections%22, new JsonArray());%0A            %0A        %7D%0A        %7B%0A            JsonArray outputs = new JsonArray();%0A            op.add(%22outputs%22, outputs);%0A            JsonObject output = new JsonObject();%0A            outputs.add(output);%0A            output.addProperty(%22index%22, 0);%0A            output.addProperty(%22type%22, schema);%0A            output.add(%22connections%22, new JsonArray());%0A            output.addProperty(%22name%22, name + %22_IN%22);%0A        %7D%0A        return op;%0A    %7D%0A
     

Remove old pre-4.2 isolate/parallel code