- () -> ref.set(buildRestServer(dmaapName, restsim)),
- () -> ref.get().shutdown());
- }
-
- // NOTE: topics must be started AFTER the (dmaap) rest servers
-
- // topic sinks
- AtomicReference<List<TopicSink>> sinkRef = new AtomicReference<>();
- addAction("topic sinks", () -> sinkRef.set(buildSinks(params.getTopicSinks())),
- () -> shutdownSinks(sinkRef.get()));
-
- // topic sources
- AtomicReference<List<TopicSource>> sourceRef = new AtomicReference<>();
- addAction("topic sources", () -> sourceRef.set(buildSources(params.getTopicSources())),
- () -> shutdownSources(sourceRef.get()));
-
- // topic server simulators
- for (TopicServerParameters topicsim : params.getTopicServers()) {
- AtomicReference<TopicServer<?>> ref = new AtomicReference<>();
- addAction(topicsim.getName(),
- () -> ref.set(buildTopicServer(topicsim, sinkRef.get(), sourceRef.get())),