import storm.trident.planner.Node; //导入依赖的package包/类
public Collection<Group> outgoingGroups(Group g) {
    Set<Group> ret = new HashSet();
    for(Node n: g.outgoingNodes()) {
        Group other = nodeGroup(n);
        if(other==null || !other.equals(g)) {
            ret.add(other);                
        }
    }
    return ret;
}
 

import storm.trident.planner.Node; //导入依赖的package包/类
private static void checkValidJoins(Collection<Node> g) {
    boolean hasDRPCSpout = false;
    boolean hasBatchSpout = false;
    for(Node n: g) {
        if(n instanceof SpoutNode) {
            SpoutNode.SpoutType type = ((SpoutNode) n).type;
            if(type==SpoutNode.SpoutType.BATCH) {
                hasBatchSpout = true;
            } else if(type==SpoutNode.SpoutType.DRPC) {
                hasDRPCSpout = true;
            }
        }
    }
    if(hasBatchSpout && hasDRPCSpout) {
        throw new RuntimeException("Cannot join DRPC stream with streams originating from other spouts");
    }
}
 

import storm.trident.planner.Node; //导入依赖的package包/类
public Stream stateQuery(TridentState state, Fields inputFields, QueryFunction function, Fields functionFields) {
    projectionValidation(inputFields);
    String stateId = state._node.stateInfo.id;
    Node n = new ProcessorNode(_topology.getUniqueStreamId(),
                    _name,
                    TridentUtils.fieldsConcat(getOutputFields(), functionFields),
                    functionFields,
                    new StateQueryProcessor(stateId, inputFields, function));
    _topology._colocate.get(stateId).add(n);
    return _topology.addSourcedNode(this, n);
}
 

import storm.trident.planner.Node; //导入依赖的package包/类
public void reindex() {
    groupIndex.clear();
    for(Group g: currGroups) {
        for(Node n: g.nodes) {
            groupIndex.put(n, g);
        }
    }  
}
 

import storm.trident.planner.Node; //导入依赖的package包/类
private void merge(Group g1, Group g2) {
    Group newGroup = new Group(g1, g2);
    currGroups.remove(g1);
    currGroups.remove(g2);
    currGroups.add(newGroup);
    for(Node n: newGroup.nodes) {
        groupIndex.put(n, newGroup);
    }
}
 

import storm.trident.planner.Node; //导入依赖的package包/类
public Collection<Group> incomingGroups(Group g) {
    Set<Group> ret = new HashSet();
    for(Node n: g.incomingNodes()) {
        Group other = nodeGroup(n);
        if(other==null || !other.equals(g)) {
            ret.add(other);                
        }
    }
    return ret;        
}
 

import storm.trident.planner.Node; //导入依赖的package包/类
protected void registerSourcedNode(List<Stream> sources, Node newNode) {
    registerNode(newNode);
    int streamIndex = 0;
    for(Stream s: sources) {
        _graph.addEdge(s._node, newNode, new IndexedEdge(s._node, newNode, streamIndex));
        streamIndex++;
    }        
}
 

import storm.trident.planner.Node; //导入依赖的package包/类
public Set<Node> outgoingNodes() {
    Set<Node> ret = new HashSet<Node>();
    for(Node n: nodes) {
        ret.addAll(TridentUtils.getChildren(graph, n));
    }
    return ret;
}
 

import storm.trident.planner.Node; //导入依赖的package包/类
public Set<Node> incomingNodes() {
    Set<Node> ret = new HashSet<Node>();
    for(Node n: nodes) {
        ret.addAll(TridentUtils.getParents(graph, n));
    }        
    return ret;        
}
 

import storm.trident.planner.Node; //导入依赖的package包/类
private Stream newDRPCStream(DRPCSpout spout) {
    // : consider adding a shuffle grouping after the spout to avoid so much routing of the args/return-info all over the place
    // (at least until its possible to just pack bolt logic into the spout itself)

    Node n = new SpoutNode(getUniqueStreamId(), TridentUtils.getSingleOutputStreamFields(spout), null, spout, SpoutNode.SpoutType.DRPC);
    Stream nextStream = addNode(n);
    // later on, this will be joined back with return-info and all the results
    return nextStream.project(new Fields("args"));
}
 

import storm.trident.planner.Node; //导入依赖的package包/类
public TridentState newStaticState(StateSpec spec) {
    String stateId = getUniqueStateId();
    Node n = new Node(getUniqueStreamId(), null, new Fields());
    n.stateInfo = new NodeStateInfo(stateId, spec);
    registerNode(n);
    return new TridentState(this, n);
}
 

import storm.trident.planner.Node; //导入依赖的package包/类
public Stream multiReduce(List<Fields> inputFields, List<Stream> streams, MultiReducer function, Fields outputFields) {
    List<String> names = new ArrayList<String>();
    for(Stream s: streams) {
        if(s._name!=null) {
            names.add(s._name);
        }
    }
    Node n = new ProcessorNode(getUniqueStreamId(), Utils.join(names, "-"), outputFields, outputFields, new MultiReducerProcessor(inputFields, function));
    return addSourcedNode(streams, n);
}
 

import storm.trident.planner.Node; //导入依赖的package包/类
private static Set<PartitionNode> externalGroupInputs(Group g) {
    Set<PartitionNode> ret = new HashSet();
    for(Node n: g.incomingNodes()) {
        if(n instanceof PartitionNode) {
            ret.add((PartitionNode) n);
        }
    }
    return ret;
}
 

import storm.trident.planner.Node; //导入依赖的package包/类
private static SpoutNode getDRPCSpoutNode(Collection<Node> g) {
    for(Node n: g) {
        if(n instanceof SpoutNode) {
            SpoutNode.SpoutType type = ((SpoutNode) n).type;
            if(type==SpoutNode.SpoutType.DRPC) {
                return (SpoutNode) n;
            }
        }
    }
    return null;
}
 

import storm.trident.planner.Node; //导入依赖的package包/类
private static Map<Node, String> genSpoutIds(Collection<SpoutNode> spoutNodes) {
    Map<Node, String> ret = new HashMap();
    int ctr = 0;
    for(SpoutNode n: spoutNodes) {
        ret.put(n, "spout" + ctr);
        ctr++;
    }
    return ret;
}
 

import storm.trident.planner.Node; //导入依赖的package包/类
private static Map<String, String> getOutputStreamBatchGroups(Group g, Map<Node, String> batchGroupMap) {
    Map<String, String> ret = new HashMap();
    Set<PartitionNode> externalGroupOutputs = externalGroupOutputs(g);
    for(PartitionNode n: externalGroupOutputs) {
        ret.put(n.streamId, batchGroupMap.get(n));
    }        
    return ret;
}
 

import storm.trident.planner.Node; //导入依赖的package包/类
private static Set<String> committerBatches(Group g, Map<Node, String> batchGroupMap) {
    Set<String> ret = new HashSet();
    for(Node n: g.nodes) {
       if(n instanceof ProcessorNode) {
           if(((ProcessorNode) n).committer) {
               ret.add(batchGroupMap.get(n));
           }
       } 
    }
    return ret;
}
 

import storm.trident.planner.Node; //导入依赖的package包/类
private static Integer getFixedParallelism(Set<Group> groups) {
    Integer ret = null;
    for(Group g: groups) {
        for(Node n: g.nodes) {
            if(n.stateInfo != null && n.stateInfo.spec.requiredNumPartitions!=null) {
                int reqPartitions = n.stateInfo.spec.requiredNumPartitions;
                if(ret!=null && ret!=reqPartitions) {
                    throw new RuntimeException("Cannot have one group have fixed parallelism of two different values");
                }
                ret = reqPartitions;
            }
        }
    }
    return ret;
}
 

import storm.trident.planner.Node; //导入依赖的package包/类
private static Set<PartitionNode> externalGroupOutputs(Group g) {
    Set<PartitionNode> ret = new HashSet();
    for(Node n: g.outgoingNodes()) {
        if(n instanceof PartitionNode) {
            ret.add((PartitionNode) n);
        }
    }
    return ret;
}
 

import storm.trident.planner.Node; //导入依赖的package包/类
protected void registerNode(Node n) {
    _graph.addVertex(n);
    if(n.stateInfo!=null) {
        String id = n.stateInfo.id;
        if(!_colocate.containsKey(id)) {
            _colocate.put(id, new ArrayList());
        }
        _colocate.get(id).add(n);
    }
}