import backtype.storm.generated.NullStruct; //导入依赖的package包/类
/**
 * Add watermark stream to source components of window bolts
 */
private void maybeAddWatermarkInputs(ComponentCommon common, IRichBolt bolt) {
    if (bolt instanceof WindowedBoltExecutor) {
        Set<String> comps = new HashSet<>();
        for (GlobalStreamId globalStreamId : common.get_inputs().keySet()) {
            comps.add(globalStreamId.get_componentId());
        }

        for (String comp : comps) {
            common.put_to_inputs(
                    new GlobalStreamId(comp, Common.WATERMARK_STREAM_ID),
                    Grouping.all(new NullStruct()));
        }
    }
}
 

import backtype.storm.generated.NullStruct; //导入依赖的package包/类
/**
 * For bolts that has incoming streams from spouts (the root bolts),
 * add checkpoint stream from checkpoint spout to its input. For other bolts,
 * add checkpoint stream from the previous bolt to its input.
 */
private void addCheckPointInputs(ComponentCommon component) {
    Set<GlobalStreamId> checkPointInputs = new HashSet<>();
    for (GlobalStreamId inputStream : component.get_inputs().keySet()) {
        String sourceId = inputStream.get_componentId();
        if (_spouts.containsKey(sourceId)) {
            checkPointInputs.add(new GlobalStreamId(CheckpointSpout.CHECKPOINT_COMPONENT_ID, CheckpointSpout.CHECKPOINT_STREAM_ID));
        } else {
            checkPointInputs.add(new GlobalStreamId(sourceId, CheckpointSpout.CHECKPOINT_STREAM_ID));
        }
    }
    for (GlobalStreamId streamId : checkPointInputs) {
        component.put_to_inputs(streamId, Grouping.all(new NullStruct()));
    }
}
 

import backtype.storm.generated.NullStruct; //导入依赖的package包/类
public static Grouping mkDirectGrouping() {
	return Grouping.direct(new NullStruct());
}
 

import backtype.storm.generated.NullStruct; //导入依赖的package包/类
public Stream shuffle() {
    return partition(Grouping.shuffle(new NullStruct()));
}
 

import backtype.storm.generated.NullStruct; //导入依赖的package包/类
public Stream localOrShuffle() {
    return partition(Grouping.local_or_shuffle(new NullStruct()));
}
 

import backtype.storm.generated.NullStruct; //导入依赖的package包/类
public Stream broadcast() {
    return partition(Grouping.all(new NullStruct()));
}
 

import backtype.storm.generated.NullStruct; //导入依赖的package包/类
@Override
public BoltDeclarer shuffleGrouping(String componentId, String streamId) {
	return grouping(componentId, streamId,
			Grouping.shuffle(new NullStruct()));
}
 

import backtype.storm.generated.NullStruct; //导入依赖的package包/类
@Override
public BoltDeclarer localOrShuffleGrouping(String componentId,
		String streamId) {
	return grouping(componentId, streamId,
			Grouping.local_or_shuffle(new NullStruct()));
}
 

import backtype.storm.generated.NullStruct; //导入依赖的package包/类
@Override
public BoltDeclarer localFirstGrouping(String componentId,
		String streamId) {
	return grouping(componentId, streamId,
			Grouping.localFirst(new NullStruct()));
}
 

import backtype.storm.generated.NullStruct; //导入依赖的package包/类
@Override
public BoltDeclarer noneGrouping(String componentId, String streamId) {
	return grouping(componentId, streamId,
			Grouping.none(new NullStruct()));
}
 

import backtype.storm.generated.NullStruct; //导入依赖的package包/类
@Override
public BoltDeclarer allGrouping(String componentId, String streamId) {
	return grouping(componentId, streamId,
			Grouping.all(new NullStruct()));
}
 

import backtype.storm.generated.NullStruct; //导入依赖的package包/类
@Override
public BoltDeclarer directGrouping(String componentId, String streamId) {
	return grouping(componentId, streamId,
			Grouping.direct(new NullStruct()));
}
 

import backtype.storm.generated.NullStruct; //导入依赖的package包/类
public BoltDeclarer shuffleGrouping(String componentId, String streamId) {
    return grouping(componentId, streamId, Grouping.shuffle(new NullStruct()));
}
 

import backtype.storm.generated.NullStruct; //导入依赖的package包/类
public BoltDeclarer localOrShuffleGrouping(String componentId, String streamId) {
    return grouping(componentId, streamId, Grouping.local_or_shuffle(new NullStruct()));
}
 

import backtype.storm.generated.NullStruct; //导入依赖的package包/类
@Override
public BoltDeclarer localFirstGrouping(String componentId, String streamId) {
    return grouping(componentId, streamId, Grouping.localFirst(new NullStruct()));
}
 

import backtype.storm.generated.NullStruct; //导入依赖的package包/类
public BoltDeclarer noneGrouping(String componentId, String streamId) {
    return grouping(componentId, streamId, Grouping.none(new NullStruct()));
}
 

import backtype.storm.generated.NullStruct; //导入依赖的package包/类
public BoltDeclarer allGrouping(String componentId, String streamId) {
    return grouping(componentId, streamId, Grouping.all(new NullStruct()));
}
 

import backtype.storm.generated.NullStruct; //导入依赖的package包/类
public BoltDeclarer directGrouping(String componentId, String streamId) {
    return grouping(componentId, streamId, Grouping.direct(new NullStruct()));
}
 

import backtype.storm.generated.NullStruct; //导入依赖的package包/类
public BoltDeclarer shuffleGrouping(String componentId, String streamId) {
	return grouping(componentId, streamId,
			Grouping.shuffle(new NullStruct()));
}
 

import backtype.storm.generated.NullStruct; //导入依赖的package包/类
public BoltDeclarer localOrShuffleGrouping(String componentId,
		String streamId) {
	return grouping(componentId, streamId,
			Grouping.local_or_shuffle(new NullStruct()));
}