import storm.trident.operation.impl.JoinerMultiReducer.JoinState; //导入依赖的package包/类
private void emitCrossJoin(JoinState state, TridentCollector collector, int overrideIndex, TridentTuple overrideTuple) {
    List<List>[] sides = state.sides;
    int[] indices = state.indices;
    for(int i=0; i<indices.length; i++) {
        indices[i] = 0;
    }
    
    boolean keepGoing = true;
    //emit cross-join of all emitted tuples
    while(keepGoing) {
        List[] combined = new List[sides.length+1];
        combined[0] = state.group;
        for(int i=0; i<sides.length; i++) {
            if(i==overrideIndex) {
                combined[i+1] = overrideTuple;
            } else {
                combined[i+1] = sides[i].get(indices[i]);                
            }
        }
        collector.emit(_factory.create(combined));
        keepGoing = increment(sides, indices, indices.length - 1, overrideIndex);
    }
}
 

import storm.trident.operation.impl.JoinerMultiReducer.JoinState; //导入依赖的package包/类
private void emitCrossJoin(JoinState state, TridentCollector collector, int overrideIndex, TridentTuple overrideTuple) {
    List<List>[] sides = state.sides;
    int[] indices = state.indices;
    for (int i = 0; i < indices.length; i++) {
        indices[i] = 0;
    }

    boolean keepGoing = true;
    // emit cross-join of all emitted tuples
    while (keepGoing) {
        List[] combined = new List[sides.length + 1];
        combined[0] = state.group;
        for (int i = 0; i < sides.length; i++) {
            if (i == overrideIndex) {
                combined[i + 1] = overrideTuple;
            } else {
                combined[i + 1] = sides[i].get(indices[i]);
            }
        }
        collector.emit(_factory.create(combined));
        keepGoing = increment(sides, indices, indices.length - 1, overrideIndex);
    }
}
 

import storm.trident.operation.impl.JoinerMultiReducer.JoinState; //导入依赖的package包/类
@Override
public void execute(JoinState state, int streamIndex, TridentTuple group, TridentTuple input, TridentCollector collector) {
    //: do the inner join incrementally, emitting the cross join with this tuple, against all other sides
    //: only do cross join if at least one tuple in each side
    List<List> side = state.sides[streamIndex];
    if(side.isEmpty()) {
        state.numSidesReceived++;
    }
            
    side.add(input);
    if(state.numSidesReceived == state.sides.length) {
        emitCrossJoin(state, collector, streamIndex, input);
    }        
}
 

import storm.trident.operation.impl.JoinerMultiReducer.JoinState; //导入依赖的package包/类
@Override
public void complete(JoinState state, TridentTuple group, TridentCollector collector) {
    List<List>[] sides = state.sides;
    boolean wasEmpty = state.numSidesReceived < sides.length;
    for(int i=0; i<sides.length; i++) {
        if(sides[i].isEmpty() && _types.get(i) == JoinType.OUTER) {
            state.numSidesReceived++;
            sides[i].add(makeNullList(_sideFields.get(i).size()));
        }
    }
    if(wasEmpty && state.numSidesReceived == sides.length) {
        emitCrossJoin(state, collector, -1, null);
    }
}
 

import storm.trident.operation.impl.JoinerMultiReducer.JoinState; //导入依赖的package包/类
public JoinState(int numSides, TridentTuple group) {
    sides = new List[numSides];
    indices = new int[numSides];
    this.group = group;
    for(int i=0; i<numSides; i++) {
        sides[i] = new ArrayList<List>();
    }            
}
 

import storm.trident.operation.impl.JoinerMultiReducer.JoinState; //导入依赖的package包/类
@Override
public void execute(JoinState state, int streamIndex, TridentTuple group, TridentTuple input, TridentCollector collector) {
    //TODO: do the inner join incrementally, emitting the cross join with this tuple, against all other sides
    //TODO: only do cross join if at least one tuple in each side
    List<List> side = state.sides[streamIndex];
    if(side.isEmpty()) {
        state.numSidesReceived++;
    }
            
    side.add(input);
    if(state.numSidesReceived == state.sides.length) {
        emitCrossJoin(state, collector, streamIndex, input);
    }        
}
 

import storm.trident.operation.impl.JoinerMultiReducer.JoinState; //导入依赖的package包/类
@Override
public void execute(JoinState state, int streamIndex, TridentTuple group, TridentTuple input, TridentCollector collector) {
    // TODO: do the inner join incrementally, emitting the cross join with this tuple, against all other sides
    // TODO: only do cross join if at least one tuple in each side
    List<List> side = state.sides[streamIndex];
    if (side.isEmpty()) {
        state.numSidesReceived++;
    }

    side.add(input);
    if (state.numSidesReceived == state.sides.length) {
        emitCrossJoin(state, collector, streamIndex, input);
    }
}
 

import storm.trident.operation.impl.JoinerMultiReducer.JoinState; //导入依赖的package包/类
@Override
public void complete(JoinState state, TridentTuple group, TridentCollector collector) {
    List<List>[] sides = state.sides;
    boolean wasEmpty = state.numSidesReceived < sides.length;
    for (int i = 0; i < sides.length; i++) {
        if (sides[i].isEmpty() && _types.get(i) == JoinType.OUTER) {
            state.numSidesReceived++;
            sides[i].add(makeNullList(_sideFields.get(i).size()));
        }
    }
    if (wasEmpty && state.numSidesReceived == sides.length) {
        emitCrossJoin(state, collector, -1, null);
    }
}
 

import storm.trident.operation.impl.JoinerMultiReducer.JoinState; //导入依赖的package包/类
public JoinState(int numSides, TridentTuple group) {
    sides = new List[numSides];
    indices = new int[numSides];
    this.group = group;
    for (int i = 0; i < numSides; i++) {
        sides[i] = new ArrayList<List>();
    }
}
 

import storm.trident.operation.impl.JoinerMultiReducer.JoinState; //导入依赖的package包/类
@Override
public JoinState init(TridentCollector collector, TridentTuple group) {
    return new JoinState(_types.size(), group);
}