import storm.trident.planner.ProcessorNode; //导入依赖的package包/类
@Override
public Stream each(Fields inputFields, Function function, Fields functionFields) {
    projectionValidation(inputFields);
    return _topology.addSourcedNode(this,
            new ProcessorNode(_topology.getUniqueStreamId(),
                _name,
                TridentUtils.fieldsConcat(getOutputFields(), functionFields),
                functionFields,
                new EachProcessor(inputFields, function)));
}
 

import storm.trident.planner.ProcessorNode; //导入依赖的package包/类
@Override
public Stream partitionAggregate(Fields inputFields, Aggregator agg, Fields functionFields) {
    projectionValidation(inputFields);
    return _topology.addSourcedNode(this,
            new ProcessorNode(_topology.getUniqueStreamId(),
                _name,
                functionFields,
                functionFields,
                new AggregateProcessor(inputFields, agg)));
}
 

import storm.trident.planner.ProcessorNode; //导入依赖的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.ProcessorNode; //导入依赖的package包/类
public TridentState partitionPersist(StateSpec stateSpec, Fields inputFields, StateUpdater updater, Fields functionFields) {
    projectionValidation(inputFields);
    String id = _topology.getUniqueStateId();
    ProcessorNode n = new ProcessorNode(_topology.getUniqueStreamId(),
                _name,
                functionFields,
                functionFields,
                new PartitionPersistProcessor(id, inputFields, updater));
    n.committer = true;
    n.stateInfo = new NodeStateInfo(id, stateSpec);
    return _topology.addSourcedStateNode(this, n);
}
 

import storm.trident.planner.ProcessorNode; //导入依赖的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.ProcessorNode; //导入依赖的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.ProcessorNode; //导入依赖的package包/类
@Override
public Stream each(Fields inputFields, Function function, Fields functionFields) {
    projectionValidation(inputFields);
    return _topology.addSourcedNode(this,
            new ProcessorNode(_topology.getUniqueStreamId(), _name, TridentUtils.fieldsConcat(getOutputFields(), functionFields), functionFields,
                    new EachProcessor(inputFields, function)));
}
 

import storm.trident.planner.ProcessorNode; //导入依赖的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.ProcessorNode; //导入依赖的package包/类
public TridentState partitionPersist(StateSpec stateSpec, Fields inputFields, StateUpdater updater, Fields functionFields) {
    projectionValidation(inputFields);
    String id = _topology.getUniqueStateId();
    ProcessorNode n =
            new ProcessorNode(_topology.getUniqueStreamId(), _name, functionFields, functionFields, new PartitionPersistProcessor(id, inputFields, updater));
    n.committer = true;
    n.stateInfo = new NodeStateInfo(id, stateSpec);
    return _topology.addSourcedStateNode(this, n);
}
 

import storm.trident.planner.ProcessorNode; //导入依赖的package包/类
@Override
public Stream each(Fields inputFields, Function function, Fields functionFields) {
    projectionValidation(inputFields);
    return _topology.addSourcedNode(this,
            new ProcessorNode(_topology.getUniqueStreamId(),
                    _name,
                    TridentUtils.fieldsConcat(getOutputFields(), functionFields),
                    functionFields,
                    new EachProcessor(inputFields, function)));
}
 

import storm.trident.planner.ProcessorNode; //导入依赖的package包/类
/**
 * Returns a stream consisting of the result of applying the given mapping function to the values of this stream.
 *
 * @param function a mapping function to be applied to each value in this stream.
 * @return the new stream
 */
public Stream map(MapFunction function) {
    projectionValidation(getOutputFields());
    return _topology.addSourcedNode(this,
                                    new ProcessorNode(
                                            _topology.getUniqueStreamId(),
                                            _name,
                                            getOutputFields(),
                                            getOutputFields(),
                                            new MapProcessor(getOutputFields(), new MapFunctionExecutor(function))));
}
 

import storm.trident.planner.ProcessorNode; //导入依赖的package包/类
/**
 * Returns a stream consisting of the results of replacing each value of this stream with the contents
 * produced by applying the provided mapping function to each value. This has the effect of applying
 * a one-to-many transformation to the values of the stream, and then flattening the resulting elements into a new stream.
 *
 * @param function a mapping function to be applied to each value in this stream which produces new values.
 * @return the new stream
 */
public Stream flatMap(FlatMapFunction function) {
    projectionValidation(getOutputFields());
    return _topology.addSourcedNode(this,
                                    new ProcessorNode(
                                            _topology.getUniqueStreamId(),
                                            _name,
                                            getOutputFields(),
                                            getOutputFields(),
                                            new MapProcessor(getOutputFields(), new FlatMapFunctionExecutor(function))));
}
 

import storm.trident.planner.ProcessorNode; //导入依赖的package包/类
/**
 * Returns a stream consisting of the trident tuples of this stream, additionally performing the provided action on
 * each trident tuple as they are consumed from the resulting stream. This is mostly useful for debugging
 * to see the tuples as they flow past a certain point in a pipeline.
 *
 * @param action the action to perform on the trident tuple as they are consumed from the stream
 * @return the new stream
 */
public Stream peek(Consumer action) {
    projectionValidation(getOutputFields());
    return _topology.addSourcedNode(this,
                                    new ProcessorNode(
                                            _topology.getUniqueStreamId(),
                                            _name,
                                            getOutputFields(),
                                            getOutputFields(),
                                            new MapProcessor(getOutputFields(), new ConsumerExecutor(action))));
}