final ChannelSelectorConfiguration selectorConfig = config.getSelectorConfiguration();
final ChannelSelector selector = ChannelSelectorFactory.create(channels, selectorConfig);
final ChannelProcessor channelProcessor = new ChannelProcessor(selector);
Configurables.configure(channelProcessor, config);
source.setChannelProcessor(channelProcessor);
conf.getSourceRunners().put(comp.getComponentName(), SourceRunner.forSource(source));
}
}
final Map<String, Context> sourceContexts = agentConf.getSourceContext();
for (final String src : sources) {
final Context context = sourceContexts.get(src);
if (context != null) {
final Source source = sourceFactory.create(src,
context.getString(BasicConfigurationConstants.CONFIG_TYPE));
final List<Channel> channels = new ArrayList<Channel>();
Configurables.configure(source, context);
final String[] channelNames =
context.getString(BasicConfigurationConstants.CONFIG_CHANNELS).split("\\s+");
for (final String chName : channelNames) {
channels.add(conf.getChannels().get(chName));
}
final Map<String, String> selectorConfig = context.getSubProperties(
BasicConfigurationConstants.CONFIG_SOURCE_CHANNELSELECTOR_PREFIX);
final ChannelSelector selector = ChannelSelectorFactory.create(channels, selectorConfig);
final ChannelProcessor channelProcessor = new ChannelProcessor(selector);
Configurables.configure(channelProcessor, context);
source.setChannelProcessor(channelProcessor);
conf.getSourceRunners().put(src, SourceRunner.forSource(source));
}