FlowNode edgeTargetFlowNode = nodeGraph.getEdgeTarget( processEdge );
FlowElement flowElement = processEdge.getFlowElement();
List<FlowNode> sourceNodes = nodeGraph.getElementSourceProcesses( flowElement );
EdgeProperty edgeProperty = createEdgeProperty( initializedConfig, processEdge );
Vertex targetVertex = vertexMap.get( edgeTargetFlowNode );
if( sourceNodes.size() == 1 || flowElement instanceof CoGroup || flowElement instanceof Boundary ) // todo: create group vertices around incoming ordinal
{
FlowNode edgeSourceFlowNode = nodeGraph.getEdgeSource( processEdge );
Vertex sourceVertex = vertexMap.get( edgeSourceFlowNode );
LOG.debug( "adding edge between: {} and {}", sourceVertex, targetVertex );
dag.addEdge( Edge.create( sourceVertex, targetVertex, edgeProperty ) );
}
else if( flowElement instanceof GroupBy || flowElement instanceof Merge ) // merge - source nodes > 1
{
List<String> sourceVerticesIDs = new ArrayList<>();
List<Vertex> sourceVertices = new ArrayList<>();
for( FlowNode edgeSourceFlowNode : sourceNodes )
{
sourceVerticesIDs.add( edgeSourceFlowNode.getID() );
sourceVertices.add( vertexMap.get( edgeSourceFlowNode ) );
processedEdges.add( nodeGraph.getEdge( edgeSourceFlowNode, edgeTargetFlowNode ) );
}
VertexGroup vertexGroup = dag.createVertexGroup( edgeTargetFlowNode.getID(), sourceVertices.toArray( new Vertex[ sourceVertices.size() ] ) );
String inputClassName = flowElement instanceof Group ? OrderedGroupedMergedKVInput.class.getName() : ConcatenatedMergedKeyValueInput.class.getName();
InputDescriptor inputDescriptor = InputDescriptor.create( inputClassName ).setUserPayload( edgeProperty.getEdgeDestination().getUserPayload() );
LOG.info( "adding grouped edge between: {} and {}", Util.join( sourceVerticesIDs, "," ), targetVertex.getName() );
dag.addEdge( GroupInputEdge.create( vertexGroup, targetVertex, edgeProperty, inputDescriptor ) );
}
else