- AtomicReference<List<TopicSource>> sourceRef = new AtomicReference<>();
- addAction("topic sources", () -> sourceRef.set(buildSources(params.getTopicSources())),
- () -> shutdownSources(sourceRef.get()));
+ Map<String, TopicSource> sources = new HashMap<>();
+ for (TopicParameters topicParams : params.getTopicSources()) {
+ String topic = topicParams.getTopic();
+ addAction("Source " + topic,
+ () -> sources.put(topic, startSource(topicParams)),
+ () -> sources.get(topic).shutdown());
+ }