import org.apache.storm.topology.IComponent; //导入依赖的package包/类
@Test(expected = IllegalArgumentException.class)
public void testRawType() throws Exception {
	IComponent boltOrSpout;

	if (this.r.nextBoolean()) {
		boltOrSpout = mock(IRichSpout.class);
	} else {
		boltOrSpout = mock(IRichBolt.class);
	}

	final SetupOutputFieldsDeclarer declarer = new SetupOutputFieldsDeclarer();
	declarer.declare(new Fields("dummy1", "dummy2"));
	PowerMockito.whenNew(SetupOutputFieldsDeclarer.class).withNoArguments().thenReturn(declarer);

	WrapperSetupHelper.getNumberOfAttributes(boltOrSpout,
			new HashSet<String>(singleton(Utils.DEFAULT_STREAM_ID)));
}
 

import org.apache.storm.topology.IComponent; //导入依赖的package包/类
@Test(expected = IllegalArgumentException.class)
public void testToManyAttributes() throws Exception {
	IComponent boltOrSpout;

	if (this.r.nextBoolean()) {
		boltOrSpout = mock(IRichSpout.class);
	} else {
		boltOrSpout = mock(IRichBolt.class);
	}

	final SetupOutputFieldsDeclarer declarer = new SetupOutputFieldsDeclarer();
	final String[] schema = new String[26];
	for (int i = 0; i < schema.length; ++i) {
		schema[i] = "a" + i;
	}
	declarer.declare(new Fields(schema));
	PowerMockito.whenNew(SetupOutputFieldsDeclarer.class).withNoArguments().thenReturn(declarer);

	WrapperSetupHelper.getNumberOfAttributes(boltOrSpout, null);
}
 

import org.apache.storm.topology.IComponent; //导入依赖的package包/类
/**
 * Computes the number of output attributes used by a {@link SpoutWrapper} or {@link BoltWrapper}
 * per declared output stream. The number is {@code -1} for raw output type or a value within range [0;25] for
 * output type {@link org.apache.flink.api.java.tuple.Tuple0 Tuple0} to
 * {@link org.apache.flink.api.java.tuple.Tuple25 Tuple25}.
 *
 * @param spoutOrBolt
 *            The Storm {@link IRichSpout spout} or {@link IRichBolt bolt} to be used.
 * @param rawOutputs
 *            Contains stream names if a single attribute output stream, should not be of type
 *            {@link org.apache.flink.api.java.tuple.Tuple1 Tuple1} but be of a raw type. (Can be {@code null}.)
 * @return The number of attributes to be used for each stream.
 * @throws IllegalArgumentException
 *             If {@code rawOutput} is {@code true} and the number of declared output attributes is not 1 or if
 *             {@code rawOutput} is {@code false} and the number of declared output attributes is not with range
 *             [0;25].
 */
static HashMap<String, Integer> getNumberOfAttributes(final IComponent spoutOrBolt,
		final Collection<String> rawOutputs)
				throws IllegalArgumentException {
	final SetupOutputFieldsDeclarer declarer = new SetupOutputFieldsDeclarer();
	spoutOrBolt.declareOutputFields(declarer);

	for (Entry<String, Integer> schema : declarer.outputSchemas.entrySet()) {
		int declaredNumberOfAttributes = schema.getValue();
		if ((declaredNumberOfAttributes < 0) || (declaredNumberOfAttributes > 25)) {
			throw new IllegalArgumentException(
					"Provided bolt declares non supported number of output attributes. Must be in range [0;25] but "
							+ "was " + declaredNumberOfAttributes);
		}

		if (rawOutputs != null && rawOutputs.contains(schema.getKey())) {
			if (declaredNumberOfAttributes != 1) {
				throw new IllegalArgumentException(
						"Ouput type is requested to be raw type, but provided bolt declares more then one output "
								+ "attribute.");
			}
			schema.setValue(-1);
		}
	}

	return declarer.outputSchemas;
}
 

import org.apache.storm.topology.IComponent; //导入依赖的package包/类
@Test
public void testEmptyDeclarerBolt() {
	IComponent boltOrSpout;

	if (this.r.nextBoolean()) {
		boltOrSpout = mock(IRichSpout.class);
	} else {
		boltOrSpout = mock(IRichBolt.class);
	}

	Assert.assertEquals(new HashMap<String, Integer>(),
			WrapperSetupHelper.getNumberOfAttributes(boltOrSpout, null));
}
 

import org.apache.storm.topology.IComponent; //导入依赖的package包/类
private void testTupleTypes(final int numberOfAttributes) throws Exception {
	String[] schema;
	if (numberOfAttributes == -1) {
		schema = new String[1];
	} else {
		schema = new String[numberOfAttributes];
	}
	for (int i = 0; i < schema.length; ++i) {
		schema[i] = "a" + i;
	}

	IComponent boltOrSpout;
	if (this.r.nextBoolean()) {
		boltOrSpout = mock(IRichSpout.class);
	} else {
		boltOrSpout = mock(IRichBolt.class);
	}

	final SetupOutputFieldsDeclarer declarer = new SetupOutputFieldsDeclarer();
	declarer.declare(new Fields(schema));
	PowerMockito.whenNew(SetupOutputFieldsDeclarer.class).withNoArguments().thenReturn(declarer);

	HashMap<String, Integer> attributes = new HashMap<String, Integer>();
	attributes.put(Utils.DEFAULT_STREAM_ID, numberOfAttributes);

	Assert.assertEquals(attributes, WrapperSetupHelper.getNumberOfAttributes(
			boltOrSpout,
			numberOfAttributes == -1 ? new HashSet<String>(singleton(Utils.DEFAULT_STREAM_ID)) : null));
}
 

import org.apache.storm.topology.IComponent; //导入依赖的package包/类
public TestSourceSpout(IComponent sourceComponent, String streamName, String streamId){
    this.sourceComponent = sourceComponent;
    this.streamName = streamName;
    this.messageQueue = MessageQueue.getInstance();
    this.streamId = streamId;
}
 

import org.apache.storm.topology.IComponent; //导入依赖的package包/类
abstract public IComponent buildBolt();