import backtype.storm.generated.Bolt; //导入依赖的package包/类
public static ComponentCommon getComponentCommon(StormTopology topology,
		String componentId) {
	for (StormTopology._Fields f : StormTopology.metaDataMap.keySet()) {
		Map<String, Object> componentMap = (Map<String, Object>) topology
				.getFieldValue(f);
		if (componentMap.containsKey(componentId)) {
			Object component = componentMap.get(componentId);
			if (component instanceof Bolt) {
				return ((Bolt) component).get_common();
			}
			if (component instanceof SpoutSpec) {
				return ((SpoutSpec) component).get_common();
			}
			if (component instanceof StateSpoutSpec) {
				return ((StateSpoutSpec) component).get_common();
			}
			throw new RuntimeException(
					"Unreachable code! No get_common conversion for component "
							+ component);
		}
	}
	throw new IllegalArgumentException(
			"Could not find component common for " + componentId);
}
 

import backtype.storm.generated.Bolt; //导入依赖的package包/类
public static ComponentCommon getComponentCommon(StormTopology topology, String componentId) {
    for(StormTopology._Fields f: StormTopology.metaDataMap.keySet()) {
        Map<String, Object> componentMap = (Map<String, Object>) topology.getFieldValue(f);
        if(componentMap.containsKey(componentId)) {
            Object component = componentMap.get(componentId);
            if(component instanceof Bolt) {
                return ((Bolt) component).get_common();
            }
            if(component instanceof SpoutSpec) {
                return ((SpoutSpec) component).get_common();
            }
            if(component instanceof StateSpoutSpec) {
                return ((StateSpoutSpec) component).get_common();
            }
            throw new RuntimeException("Unreachable code! No get_common conversion for component " + component);
        }
    }
    throw new IllegalArgumentException("Could not find component common for " + componentId);
}
 

import backtype.storm.generated.Bolt; //导入依赖的package包/类
public static Set<String> getDownstreamComponents(String componentId, StormTopology topology) {
    Set<String> components = new HashSet<>();

    Map<String, Bolt> bolts = topology.get_bolts();
    for (Entry<String, Bolt> entry : bolts.entrySet()) {
        String downstreamComponentId = entry.getKey();
        Bolt bolt = entry.getValue();
        Set<GlobalStreamId> input = bolt.get_common().get_inputs().keySet();
        for (GlobalStreamId stream : input) {
            String upstreamComponentId = stream.get_componentId();
            if (upstreamComponentId.equals(componentId)) {
                components.add(downstreamComponentId);
                break;
            }
        }
    }

    return components;
}
 

import backtype.storm.generated.Bolt; //导入依赖的package包/类
public static Set<String> getAllDownstreamComponents(String componentId, StormTopology topology,
                                                     Set<String> traversedComponents) {
    Set<String> components = new HashSet<>();
    traversedComponents.add(componentId);

    Map<String, Bolt> bolts = topology.get_bolts();
    for (Entry<String, Bolt> entry : bolts.entrySet()) {
        String downstreamComponentId = entry.getKey();
        Bolt bolt = entry.getValue();
        Set<GlobalStreamId> input = bolt.get_common().get_inputs().keySet();
        for (GlobalStreamId stream : input) {
            String upstreamComponentId = stream.get_componentId();
            if (upstreamComponentId.equals(componentId) && !traversedComponents.contains(downstreamComponentId)) {
                components.add(downstreamComponentId);
                components.addAll(getAllDownstreamComponents(downstreamComponentId, topology, traversedComponents));
                break;
            }
        }
    }

    return components;
}
 

import backtype.storm.generated.Bolt; //导入依赖的package包/类
public static Set<Integer> getDownstreamTasks(String componentId, TopologyContext context) {
    Set<Integer> tasks = new HashSet<>();
    StormTopology topology = context.getRawTopology();

    Map<String, Bolt> bolts = topology.get_bolts();
    for (Entry<String, Bolt> entry : bolts.entrySet()) {
        String downstreamComponentId = entry.getKey();
        Bolt bolt = entry.getValue();
        Set<GlobalStreamId> input = bolt.get_common().get_inputs().keySet();
        for (GlobalStreamId stream : input) {
            String upstreamComponentId = stream.get_componentId();
            if (upstreamComponentId.equals(componentId)) {
                tasks.addAll(context.getComponentTasks(downstreamComponentId));
                break;
            }
        }
    }

    return tasks;
}
 

import backtype.storm.generated.Bolt; //导入依赖的package包/类
public static Set<String> getEndBolts(StormTopology topology) {
    Set<String> endBolts = new HashSet<>();

    Map<String, Bolt> bolts = topology.get_bolts();
    for (Entry<String, Bolt> entry : bolts.entrySet()) {
        String componentId = entry.getKey();
        Bolt bolt = entry.getValue();
        Map<String, StreamInfo> outputStreams = bolt.get_common().get_streams();
        if (outputStreams.size() == 0) {
            endBolts.add(entry.getKey());  
        } else {
            // Try to check if there is any "real" downstream for the output stream
            boolean found = false;
            for (String streamId : outputStreams.keySet()) {
                if (hasDownstreamComponent(topology, componentId, streamId)) {
                    found = true;
                    break;
                }
            }
            if (!found)
                endBolts.add(entry.getKey()); 
        }
    }

    return endBolts;
}
 

import backtype.storm.generated.Bolt; //导入依赖的package包/类
@Test
public void aggregate() throws Exception {
	beansXml = "<breeze:topology id='aggregate'>" +
			"<breeze:spout id='iterator' beanType='java.util.Iterator' signature='next()' outputFields='x'/>" +
			"<breeze:spout id='queue' beanType='java.util.Queue' signature='poll()' outputFields='x'/>" +
			"<breeze:bolt id='collector' beanType='org.slf4j.Logger' signature='info(x)'/>" +
			"</breeze:topology>";
	refresh();

	StormTopology topology = getBean("aggregate", StormTopology.class);

	Bolt collector = topology.get_bolts().get("collector");
	Map<GlobalStreamId, Grouping> inputs = collector.get_common().get_inputs();
	assertEquals("input count", 2, inputs.size());
	assertNotNull("iterator grouping", inputs.get(new GlobalStreamId("iterator", "default")));
	assertNotNull("queue grouping", inputs.get(new GlobalStreamId("queue", "default")));
}
 

import backtype.storm.generated.Bolt; //导入依赖的package包/类
public static Bolt mkBolt(Map<GlobalStreamId, Grouping> inputs,
		IBolt bolt, HashMap<String, StreamInfo> output, Integer p) {
	ComponentCommon common = mkComponentcommon(inputs, output, p);
	byte[] boltSer = Utils.serialize(bolt);
	ComponentObject component = ComponentObject.serialized_java(boltSer);
	return new Bolt(component, common);
}
 

import backtype.storm.generated.Bolt; //导入依赖的package包/类
@SuppressWarnings("unused")
private int computeWorkerNum() {
	Integer settingNum = JStormUtils.parseInt(stormConf
			.get(Config.TOPOLOGY_WORKERS));

	int hintSum = 0;

	Map<String, Object> components = ThriftTopologyUtils
			.getComponents(sysTopology);
	for (Entry<String, Object> entry : components.entrySet()) {
		String componentName = entry.getKey();
		Object component = entry.getValue();

		ComponentCommon common = null;
		if (component instanceof Bolt) {
			common = ((Bolt) component).get_common();
		}
		if (component instanceof SpoutSpec) {
			common = ((SpoutSpec) component).get_common();
		}
		if (component instanceof StateSpoutSpec) {
			common = ((StateSpoutSpec) component).get_common();
		}

		int hint = common.get_parallelism_hint();
		hintSum += hint;
	}

	if (settingNum == null) {
		return hintSum;
	} else {
		return Math.min(settingNum, hintSum);
	}
}
 

import backtype.storm.generated.Bolt; //导入依赖的package包/类
@SuppressWarnings("unused")
public static StormTopology add_system_components(StormTopology topology) {
	// generate inputs
	Map<GlobalStreamId, Grouping> inputs = new HashMap<GlobalStreamId, Grouping>();

	// generate outputs
	HashMap<String, StreamInfo> outputs = new HashMap<String, StreamInfo>();
	ArrayList<String> fields = new ArrayList<String>();

	outputs.put(Constants.SYSTEM_TICK_STREAM_ID,
			Thrift.outputFields(JStormUtils.mk_list("rate_secs")));
	outputs.put(Constants.METRICS_TICK_STREAM_ID,
			Thrift.outputFields(JStormUtils.mk_list("interval")));
	outputs.put(Constants.CREDENTIALS_CHANGED_STREAM_ID,
			Thrift.outputFields(JStormUtils.mk_list("creds")));

	ComponentCommon common = new ComponentCommon(inputs, outputs);

	IBolt ackerbolt = new SystemBolt();

	Bolt bolt = Thrift.mkBolt(inputs, ackerbolt, outputs,
			Integer.valueOf(0));

	topology.put_to_bolts(Constants.SYSTEM_COMPONENT_ID, bolt);

	add_system_streams(topology);

	return topology;

}
 

import backtype.storm.generated.Bolt; //导入依赖的package包/类
/**
 * get all bolts' inputs and spouts' outputs <Bolt_name, <Input_name>>
 * <Spout_name, <Output_name>>
 * 
 * @param topology_context
 * @return all bolts' inputs and spouts' outputs
 */
public static Map<String, Set<String>> buildSpoutOutoputAndBoltInputMap(
		DefaultTopologyAssignContext context) {
	Set<String> bolts = context.getRawTopology().get_bolts().keySet();
	Set<String> spouts = context.getRawTopology().get_spouts().keySet();
	Map<String, Set<String>> relationship = new HashMap<String, Set<String>>();
	for (Entry<String, Bolt> entry : context.getRawTopology().get_bolts()
			.entrySet()) {
		Map<GlobalStreamId, Grouping> inputs = entry.getValue()
				.get_common().get_inputs();
		Set<String> input = new HashSet<String>();
		relationship.put(entry.getKey(), input);
		for (Entry<GlobalStreamId, Grouping> inEntry : inputs.entrySet()) {
			String component = inEntry.getKey().get_componentId();
			input.add(component);
			if (!bolts.contains(component)) {
				// spout
				Set<String> spoutOutput = relationship.get(component);
				if (spoutOutput == null) {
					spoutOutput = new HashSet<String>();
					relationship.put(component, spoutOutput);
				}
				spoutOutput.add(entry.getKey());
			}
		}
	}
	for (String spout : spouts) {
		if (relationship.get(spout) == null)
			relationship.put(spout, new HashSet<String>());
	}
	for (String bolt : bolts) {
		if (relationship.get(bolt) == null)
			relationship.put(bolt, new HashSet<String>());
	}
	return relationship;
}
 

import backtype.storm.generated.Bolt; //导入依赖的package包/类
public static String componentType(StormTopology topology, String id) {
	Map<String, Bolt> bolts = topology.get_bolts();
	Map<String, SpoutSpec> spouts = topology.get_spouts();
	String type = "";
	if (bolts.containsKey(id)) {
		type = BOLT_STR;
	} else if (spouts.containsKey(id)) {
		type = SPOUT_STR;
	}
	return type;
}
 

import backtype.storm.generated.Bolt; //导入依赖的package包/类
private int computeWorkerNum() {
	Integer settingNum = JStormUtils.parseInt(stormConf
			.get(Config.TOPOLOGY_WORKERS));

	int hintSum = 0;

	Map<String, Object> components = ThriftTopologyUtils
			.getComponents(sysTopology);
	for (Entry<String, Object> entry : components.entrySet()) {
		String componentName = entry.getKey();
		Object component = entry.getValue();

		ComponentCommon common = null;
		if (component instanceof Bolt) {
			common = ((Bolt) component).get_common();
		}
		if (component instanceof SpoutSpec) {
			common = ((SpoutSpec) component).get_common();
		}
		if (component instanceof StateSpoutSpec) {
			common = ((StateSpoutSpec) component).get_common();
		}

		int hint = common.get_parallelism_hint();
		hintSum += hint;
	}

	if (settingNum == null) {
		return hintSum;
	} else {
		return Math.min(settingNum, hintSum);
	}
}
 

import backtype.storm.generated.Bolt; //导入依赖的package包/类
public static StormTopology add_system_components(StormTopology topology) {
	// generate inputs
	Map<GlobalStreamId, Grouping> inputs = new HashMap<GlobalStreamId, Grouping>();

	// generate outputs
	HashMap<String, StreamInfo> outputs = new HashMap<String, StreamInfo>();
	ArrayList<String> fields = new ArrayList<String>();

	outputs.put(Constants.SYSTEM_TICK_STREAM_ID,
			Thrift.outputFields(JStormUtils.mk_list("rate_secs")));
	outputs.put(Constants.METRICS_TICK_STREAM_ID,
			Thrift.outputFields(JStormUtils.mk_list("interval")));
	outputs.put(Constants.CREDENTIALS_CHANGED_STREAM_ID,
			Thrift.outputFields(JStormUtils.mk_list("creds")));

	ComponentCommon common = new ComponentCommon(inputs, outputs);

	IBolt ackerbolt = new SystemBolt();

	Bolt bolt = Thrift.mkBolt(inputs, ackerbolt, outputs,
			Integer.valueOf(0));

	topology.put_to_bolts(Constants.SYSTEM_COMPONENT_ID, bolt);

	add_system_streams(topology);

	return topology;

}
 

import backtype.storm.generated.Bolt; //导入依赖的package包/类
private int setBoltInfo(StormTopology oldTopology, StormTopology newTopology, int cnt, StormClusterState clusterState) throws Exception {
    Map<String, Bolt> oldBolts = oldTopology.get_bolts();
    Map<String, Bolt> bolts = newTopology.get_bolts();
    for (Entry<String, Bolt> entry : oldBolts.entrySet()) {
        String boltName = entry.getKey();
        Bolt oldBolt = entry.getValue();
        Bolt bolt = bolts.get(boltName);
        if (oldBolt.get_common().get_parallelism_hint() > bolt.get_common().get_parallelism_hint()) {
            int removedTaskNum = oldBolt.get_common().get_parallelism_hint() - bolt.get_common().get_parallelism_hint();
            TreeSet<Integer> taskIds = new TreeSet<Integer>(clusterState.task_ids_by_componentId(topologyid, boltName));
            Iterator<Integer> descendIterator = taskIds.descendingIterator();
            while (--removedTaskNum >= 0) {
                int taskId = descendIterator.next();
                removeTask(topologyid, taskId, clusterState);
                LOG.info("Remove bolt task, taskId=" + taskId + " for " + boltName);
            }
        } else if (oldBolt.get_common().get_parallelism_hint() == bolt.get_common().get_parallelism_hint()) {
            continue;
        } else {
            int delta = bolt.get_common().get_parallelism_hint() - oldBolt.get_common().get_parallelism_hint();
            Map<Integer, TaskInfo> taskInfoMap = new HashMap<Integer, TaskInfo>();

            for (int i = 1; i <= delta; i++) {
                cnt++;
                TaskInfo taskInfo = new TaskInfo((String) entry.getKey(), "bolt");
                taskInfoMap.put(cnt, taskInfo);
                newTasks.add(cnt);
                LOG.info("Setup new bolt task, taskId=" + cnt + " for " + boltName);
            }
            clusterState.add_task(topologyid, taskInfoMap);
        }
    }

    return cnt;
}
 

import backtype.storm.generated.Bolt; //导入依赖的package包/类
/**
 * Returns a component common from an object. This (casting) methods is required, as spouts and bolts do 
 * not have a common interface and currently TBase is not available.
 * 
 * @param object the object to take the component common from
 * @return the component common or <b>null</b>
 */
private static ComponentCommon getComponentCommon(Object object) {
    ComponentCommon result = null;
    if (object instanceof SpoutSpec) {
        result = ((SpoutSpec) object).get_common();
    } else if (object instanceof Bolt) {
        result = ((Bolt) object).get_common();
    }
    return result;
}
 

import backtype.storm.generated.Bolt; //导入依赖的package包/类
/**
 * Tests the recording topology builder.
 */
@Test
@SuppressWarnings("rawtypes")
public void testWithPipelineOptions() {
    final int numWorkers = 2;
    final String pipName = "TestPip";
    final int boltParallel = 3;
    final int boltTasks = 4;

    PipelineOptions options = new PipelineOptions();
    
    options.setNumberOfWorkers(numWorkers);
    options.setExecutorParallelism(TestPipeline.TEST_BOLT_IMPLNAME, boltParallel);
    options.setTaskParallelism(TestPipeline.TEST_BOLT_IMPLNAME, boltTasks);
    TestPipeline.main(options.toArgs(pipName));

    StormTopology topology = TestPipeline.getTopology();
    Assert.assertEquals(1, topology.get_bolts_size());
    Bolt bolt = topology.get_bolts().get(TestPipeline.TEST_BOLT_IMPLNAME);
    ComponentCommon common = bolt.get_common();
    Assert.assertEquals(boltParallel, common.get_parallelism_hint());
    Map jmap = (Map) JSONValue.parse(common.get_json_conf());
    Assert.assertEquals(Long.valueOf(boltTasks), jmap.get("topology.tasks"));
    
    Config conf = TestPipeline.getConfig();
    Assert.assertEquals(numWorkers, conf.get(Config.TOPOLOGY_WORKERS));
}
 

import backtype.storm.generated.Bolt; //导入依赖的package包/类
public static Bolt mkAckerBolt(Map<GlobalStreamId, Grouping> inputs,
		IBolt bolt, HashMap<String, StreamInfo> output, Integer p) {
	ComponentCommon common = mkAckerComponentcommon(inputs, output, p);
	byte[] boltSer = Utils.serialize(bolt);
	ComponentObject component = ComponentObject.serialized_java(boltSer);
	return new Bolt(component, common);
}
 

import backtype.storm.generated.Bolt; //导入依赖的package包/类
private ListenableDirectedGraph<NamedNode, DefaultEdge> _asGraph(StormTopology t) {
	final Map<String, Bolt> bolts = t.get_bolts();
	final Map<String, SpoutSpec> spouts = t.get_spouts();
	final ListenableDirectedGraph<NamedNode, DefaultEdge> ret = new ListenableDirectedGraph<NamedNode, DefaultEdge>(
			DefaultEdge.class);

	createSpouts(spouts, ret);
	createBolts(bolts, ret);
	createConnections(bolts, ret);
	return ret;
}
 

import backtype.storm.generated.Bolt; //导入依赖的package包/类
private void createConnections(Map<String, Bolt> bolts, ListenableDirectedGraph<NamedNode, DefaultEdge> ret) {
	for (final Entry<String, Bolt> boltspec : bolts.entrySet()) {
		final Bolt bolt = boltspec.getValue();
		final String id = boltspec.getKey();
		final Map<GlobalStreamId, Grouping> inputs = bolt.get_common().get_inputs();
		for (final Entry<GlobalStreamId, Grouping> input : inputs.entrySet()) {
			final GlobalStreamId from = input.getKey();
			// Grouping grouping = input.getValue();
			final String fromId = from.get_componentId();
			// String streamId = from.get_streamId();
			ret.addEdge(nns.get(fromId), nns.get(id));
		}

	}
}