Processing Streaming Data - Real-Time Analytics - page 159

def.fields = new Fields(fields);
filters.add(def);
return this ;

}

This bolt's prepare method captures the collector output as well as
preparing the filters for further use. In this case, the unique input
streams are inspected to determine the fields present in their output
tuples. If they match a particular filter, that filter is bound to that
particular stream. As a performance improvement, the index of the
filter element is retrieved as well:

public class FilterBinding {
public FilterDefinition filter;
public int fieldNdx;

}
transient OutputCollector
collector;
transient Map<GlobalStreamId,List<FilterBinding>>
bindings;
public void prepare(Map stormConf, TopologyContext
context,
OutputCollector collector) {
this .collector = collector;
bindings = new
HashMap<GlobalStreamId,List<FilterBinding>>();
for (GlobalStreamId id :
context.getThisSources().keySet()) {
Fields fromId =
context.getComponentOutputFields(id);
ArrayList<FilterBinding> bounds =
new ArrayList<FilterBinding>();
for (FilterDefinition def : filters) {
if (fromId.contains(def.fieldName)) {
FilterBinding bind = new FilterBinding();
bind.filter = def;