import storm.trident.planner.ProcessorContext; //导入依赖的package包/类
@Override
public void finishBatch(ProcessorContext processorContext) {
    BatchState state = (BatchState) processorContext.state[_context.getStateIndex()];
    if(!state.tuples.isEmpty()) {
        List<Object> results = _function.batchRetrieve(_state, state.args);
        if(results.size()!=state.tuples.size()) {
            throw new RuntimeException("Results size is different than argument size: " + results.size() + " vs " + state.tuples.size());
        }
        for(int i=0; i<state.tuples.size(); i++) {
            TridentTuple tuple = state.tuples.get(i);
            Object result = results.get(i);
            _collector.setContext(processorContext, tuple);
            _function.execute(_projection.create(tuple), result, _collector);            
        }
    }
}
 

import storm.trident.planner.ProcessorContext; //导入依赖的package包/类
@Override
public void finishBatch(ProcessorContext processorContext) {
    _collector.setContext(processorContext);
    Object batchId = processorContext.batchId;
    // since this processor type is a committer, this occurs in the commit phase
    List<TridentTuple> buffer = (List) processorContext.state[_context.getStateIndex()];
    
    // don't update unless there are tuples
    // this helps out with things like global partition persist, where multiple tasks may still
    // exist for this processor. Only want the global one to do anything
    // this is also a helpful optimization that state implementations don't need to manually do
    if(buffer.size() > 0) {
        Long txid = null;
        // this is to support things like persisting off of drpc stream, which is inherently unreliable
        // and won't have a tx attempt
        if(batchId instanceof TransactionAttempt) {
            txid = ((TransactionAttempt) batchId).getTransactionId();
        }
        _state.beginCommit(txid);
        _updater.updateState(_state, buffer, _collector);
        _state.commit(txid);
    }
}
 

import storm.trident.planner.ProcessorContext; //导入依赖的package包/类
@Override
public void finishBatch(ProcessorContext processorContext) {
    BatchState state = (BatchState) processorContext.state[_context.getStateIndex()];
    if (!state.tuples.isEmpty()) {
        List<Object> results = _function.batchRetrieve(_state, state.args);
        if (results.size() != state.tuples.size()) {
            throw new RuntimeException("Results size is different than argument size: " + results.size() + " vs " + state.tuples.size());
        }
        for (int i = 0; i < state.tuples.size(); i++) {
            TridentTuple tuple = state.tuples.get(i);
            Object result = results.get(i);
            _collector.setContext(processorContext, tuple);
            _function.execute(_projection.create(tuple), result, _collector);
        }
    }
}
 

import storm.trident.planner.ProcessorContext; //导入依赖的package包/类
@Override
public void finishBatch(ProcessorContext processorContext) {
    _collector.setContext(processorContext);
    Object batchId = processorContext.batchId;
    // since this processor type is a committer, this occurs in the commit phase
    List<TridentTuple> buffer = (List) processorContext.state[_context.getStateIndex()];

    // don't update unless there are tuples
    // this helps out with things like global partition persist, where multiple tasks may still
    // exist for this processor. Only want the global one to do anything
    // this is also a helpful optimization that state implementations don't need to manually do
    if (buffer.size() > 0) {
        Long txid = null;
        // this is to support things like persisting off of drpc stream, which is inherently unreliable
        // and won't have a tx attempt
        if (batchId instanceof TransactionAttempt) {
            txid = ((TransactionAttempt) batchId).getTransactionId();
        }
        _state.beginCommit(txid);
        _updater.updateState(_state, buffer, _collector);
        _state.commit(txid);
    }
}
 

import storm.trident.planner.ProcessorContext; //导入依赖的package包/类
@Override
public void finishBatch(ProcessorContext processorContext) {
    BatchState state = (BatchState) processorContext.state[_context.getStateIndex()];
    if(!state.tuples.isEmpty()) {
        List<Object> results = _function.batchRetrieve(_state, Collections.unmodifiableList(state.args));
        if(results.size()!=state.tuples.size()) {
            throw new RuntimeException("Results size is different than argument size: " + results.size() + " vs " + state.tuples.size());
        }
        for (int i = 0; i < state.tuples.size(); i++) {
            TridentTuple tuple = state.tuples.get(i);
            Object result = results.get(i);
            _collector.setContext(processorContext, tuple);
            _function.execute(state.args.get(i), result, _collector);
        }
    }
}
 

import storm.trident.planner.ProcessorContext; //导入依赖的package包/类
@Override
public void execute(ProcessorContext processorContext, String streamId, TridentTuple tuple) {
    TridentTuple toEmit = _factory.create(tuple);
    for(TupleReceiver r: _context.getReceivers()) {
        r.execute(processorContext, _context.getOutStreamId(), toEmit);
    }
}
 

import storm.trident.planner.ProcessorContext; //导入依赖的package包/类
@Override
public void execute(ProcessorContext processorContext, String streamId, TridentTuple tuple) {
    TridentTuple toEmit = _factory.create(tuple);
    for (TupleReceiver r : _context.getReceivers()) {
        r.execute(processorContext, _context.getOutStreamId(), toEmit);
    }
}
 

import storm.trident.planner.ProcessorContext; //导入依赖的package包/类
@Override
public void startBatch(ProcessorContext processorContext) {
}
 

import storm.trident.planner.ProcessorContext; //导入依赖的package包/类
@Override
public void finishBatch(ProcessorContext processorContext) {
}
 

import storm.trident.planner.ProcessorContext; //导入依赖的package包/类
@Override
public void startBatch(ProcessorContext processorContext) {
    _collector.setContext(processorContext);
    processorContext.state[_context.getStateIndex()] = _reducer.init(_collector);
}
 

import storm.trident.planner.ProcessorContext; //导入依赖的package包/类
@Override
public void execute(ProcessorContext processorContext, String streamId, TridentTuple tuple) {
    _collector.setContext(processorContext);
    int i = _streamToIndex.get(streamId);
    _reducer.execute(processorContext.state[_context.getStateIndex()], i, _projectionFactories[i].create(tuple), _collector);
}
 

import storm.trident.planner.ProcessorContext; //导入依赖的package包/类
@Override
public void finishBatch(ProcessorContext processorContext) {
    _collector.setContext(processorContext);
    _reducer.complete(processorContext.state[_context.getStateIndex()], _collector);
}
 

import storm.trident.planner.ProcessorContext; //导入依赖的package包/类
public void setContext(ProcessorContext pc) {
    this.context = pc;
}
 

import storm.trident.planner.ProcessorContext; //导入依赖的package包/类
@Override
public void startBatch(ProcessorContext processorContext) {
    _collector.setContext(processorContext);
    processorContext.state[_context.getStateIndex()] = _agg.init(processorContext.batchId, _collector);
}
 

import storm.trident.planner.ProcessorContext; //导入依赖的package包/类
@Override
public void execute(ProcessorContext processorContext, String streamId, TridentTuple tuple) {
    _collector.setContext(processorContext);
    _agg.aggregate(processorContext.state[_context.getStateIndex()], _projection.create(tuple), _collector);
}
 

import storm.trident.planner.ProcessorContext; //导入依赖的package包/类
@Override
public void finishBatch(ProcessorContext processorContext) {
    _collector.setContext(processorContext);
    _agg.complete(processorContext.state[_context.getStateIndex()], _collector);
}
 

import storm.trident.planner.ProcessorContext; //导入依赖的package包/类
@Override
public void startBatch(ProcessorContext processorContext) {
    processorContext.state[_context.getStateIndex()] =  new BatchState();
}
 

import storm.trident.planner.ProcessorContext; //导入依赖的package包/类
@Override
public void execute(ProcessorContext processorContext, String streamId, TridentTuple tuple) {
    BatchState state = (BatchState) processorContext.state[_context.getStateIndex()];
    state.tuples.add(tuple);
    state.args.add(_projection.create(tuple));
}
 

import storm.trident.planner.ProcessorContext; //导入依赖的package包/类
@Override
public void startBatch(ProcessorContext processorContext) {
    processorContext.state[_context.getStateIndex()] = new ArrayList<TridentTuple>();        
}
 

import storm.trident.planner.ProcessorContext; //导入依赖的package包/类
@Override
public void execute(ProcessorContext processorContext, String streamId, TridentTuple tuple) {
    ((List) processorContext.state[_context.getStateIndex()]).add(_projection.create(tuple));
}