@SuppressWarnings("unchecked") @Test public void loadSourceTest() { Map<String, Source> sources = SourceAndSinkBuilder.createSourcesFromConfiguration(vertx, fluid.getConfig()); assertThat(sources).hasSize(2); Source<String> source1 = sources.get("source1"); assertThat(source1).isNotNull(); Source<Integer> source2 = sources.get("source2"); assertThat(source2).isNotNull(); ListSink<String> list = new ListSink<>(); source1.to(list); assertThat(list.values()).containsExactly("a", "b", "c"); assertThat(source1.name()).isEqualTo("source1"); ListSink<Integer> list2 = new ListSink<>(); source2.to(list2); assertThat(list2.values()).containsExactly(0, 1, 2, 3); assertThat(source2.name()).isEqualTo("source2"); }
public static Map<String, Source> createSourcesFromConfiguration(Vertx vertx, FluidConfig config) { Map<String, Source> map = new HashMap<>(); Optional<Config> sources = config.getConfig("sources"); if (sources.isPresent()) { Iterator<String> names = sources.get().names(); while (names.hasNext()) { String name = names.next(); LOGGER.info("Creating source from configuration `" + name + "`"); Optional<Config> conf = sources.get().getConfig(name); Source<?> source = buildSource(vertx, name, conf.orElseThrow(() -> new IllegalStateException("Illegal configuration for source `" + name + "`"))); map.put(name, source); } } else { LOGGER.warn("No sources configured from the fluid configuration"); } return map; }
SourceAndSinkBuilder { public static Map<String, Source> createSourcesFromConfiguration(Vertx vertx, FluidConfig config) { Map<String, Source> map = new HashMap<>(); Optional<Config> sources = config.getConfig("sources"); if (sources.isPresent()) { Iterator<String> names = sources.get().names(); while (names.hasNext()) { String name = names.next(); LOGGER.info("Creating source from configuration `" + name + "`"); Optional<Config> conf = sources.get().getConfig(name); Source<?> source = buildSource(vertx, name, conf.orElseThrow(() -> new IllegalStateException("Illegal configuration for source `" + name + "`"))); map.put(name, source); } } else { LOGGER.warn("No sources configured from the fluid configuration"); } return map; } }
SourceAndSinkBuilder { public static Map<String, Source> createSourcesFromConfiguration(Vertx vertx, FluidConfig config) { Map<String, Source> map = new HashMap<>(); Optional<Config> sources = config.getConfig("sources"); if (sources.isPresent()) { Iterator<String> names = sources.get().names(); while (names.hasNext()) { String name = names.next(); LOGGER.info("Creating source from configuration `" + name + "`"); Optional<Config> conf = sources.get().getConfig(name); Source<?> source = buildSource(vertx, name, conf.orElseThrow(() -> new IllegalStateException("Illegal configuration for source `" + name + "`"))); map.put(name, source); } } else { LOGGER.warn("No sources configured from the fluid configuration"); } return map; }  }
SourceAndSinkBuilder { public static Map<String, Source> createSourcesFromConfiguration(Vertx vertx, FluidConfig config) { Map<String, Source> map = new HashMap<>(); Optional<Config> sources = config.getConfig("sources"); if (sources.isPresent()) { Iterator<String> names = sources.get().names(); while (names.hasNext()) { String name = names.next(); LOGGER.info("Creating source from configuration `" + name + "`"); Optional<Config> conf = sources.get().getConfig(name); Source<?> source = buildSource(vertx, name, conf.orElseThrow(() -> new IllegalStateException("Illegal configuration for source `" + name + "`"))); map.put(name, source); } } else { LOGGER.warn("No sources configured from the fluid configuration"); } return map; }  static Map<String, Source> createSourcesFromConfiguration(Vertx vertx, FluidConfig config); static Map<String, Sink> createSinksFromConfiguration(Vertx vertx, FluidConfig config); }
SourceAndSinkBuilder { public static Map<String, Source> createSourcesFromConfiguration(Vertx vertx, FluidConfig config) { Map<String, Source> map = new HashMap<>(); Optional<Config> sources = config.getConfig("sources"); if (sources.isPresent()) { Iterator<String> names = sources.get().names(); while (names.hasNext()) { String name = names.next(); LOGGER.info("Creating source from configuration `" + name + "`"); Optional<Config> conf = sources.get().getConfig(name); Source<?> source = buildSource(vertx, name, conf.orElseThrow(() -> new IllegalStateException("Illegal configuration for source `" + name + "`"))); map.put(name, source); } } else { LOGGER.warn("No sources configured from the fluid configuration"); } return map; }  static Map<String, Source> createSourcesFromConfiguration(Vertx vertx, FluidConfig config); static Map<String, Sink> createSinksFromConfiguration(Vertx vertx, FluidConfig config);  }