@SuppressWarnings("unchecked") @Test public void loadSinkTest() { Map<String, Sink> sinks = SourceAndSinkBuilder.createSinksFromConfiguration(vertx, fluid.getConfig()); assertThat(sinks).hasSize(2); Sink<String> sink1 = sinks.get("sink1"); assertThat(sink1).isNotNull().isInstanceOf(ListSink.class); Sink<Integer> sink2 = sinks.get("sink2"); assertThat(sink2).isNotNull().isInstanceOf(ListSink.class); Source.from("1", "2", "3").to(sink1); Source.from(4, 5).to(sink2); assertThat(((ListSink) sink1).values()).containsExactly("1", "2", "3"); assertThat(((ListSink) sink2).values()).containsExactly(4, 5); }
public static Map<String, Sink> createSinksFromConfiguration(Vertx vertx, FluidConfig config) { Map<String, Sink> map = new HashMap<>(); Optional<Config> sinks = config.getConfig("sinks"); if (sinks.isPresent()) { Iterator<String> names = sinks.get().names(); while (names.hasNext()) { String name = names.next(); LOGGER.info("Creating sink from configuration `" + name + "`"); Optional<Config> conf = sinks.get().getConfig(name); Sink<?> sink = buildSink(vertx, name, conf.orElseThrow(() -> new IllegalStateException("Illegal configuration for source `" + name + "`"))); map.put(name, sink); } } else { LOGGER.warn("No sinks configured from the fluid configuration"); } return map; }
SourceAndSinkBuilder { public static Map<String, Sink> createSinksFromConfiguration(Vertx vertx, FluidConfig config) { Map<String, Sink> map = new HashMap<>(); Optional<Config> sinks = config.getConfig("sinks"); if (sinks.isPresent()) { Iterator<String> names = sinks.get().names(); while (names.hasNext()) { String name = names.next(); LOGGER.info("Creating sink from configuration `" + name + "`"); Optional<Config> conf = sinks.get().getConfig(name); Sink<?> sink = buildSink(vertx, name, conf.orElseThrow(() -> new IllegalStateException("Illegal configuration for source `" + name + "`"))); map.put(name, sink); } } else { LOGGER.warn("No sinks configured from the fluid configuration"); } return map; } }
SourceAndSinkBuilder { public static Map<String, Sink> createSinksFromConfiguration(Vertx vertx, FluidConfig config) { Map<String, Sink> map = new HashMap<>(); Optional<Config> sinks = config.getConfig("sinks"); if (sinks.isPresent()) { Iterator<String> names = sinks.get().names(); while (names.hasNext()) { String name = names.next(); LOGGER.info("Creating sink from configuration `" + name + "`"); Optional<Config> conf = sinks.get().getConfig(name); Sink<?> sink = buildSink(vertx, name, conf.orElseThrow(() -> new IllegalStateException("Illegal configuration for source `" + name + "`"))); map.put(name, sink); } } else { LOGGER.warn("No sinks configured from the fluid configuration"); } return map; }  }
SourceAndSinkBuilder { public static Map<String, Sink> createSinksFromConfiguration(Vertx vertx, FluidConfig config) { Map<String, Sink> map = new HashMap<>(); Optional<Config> sinks = config.getConfig("sinks"); if (sinks.isPresent()) { Iterator<String> names = sinks.get().names(); while (names.hasNext()) { String name = names.next(); LOGGER.info("Creating sink from configuration `" + name + "`"); Optional<Config> conf = sinks.get().getConfig(name); Sink<?> sink = buildSink(vertx, name, conf.orElseThrow(() -> new IllegalStateException("Illegal configuration for source `" + name + "`"))); map.put(name, sink); } } else { LOGGER.warn("No sinks 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, Sink> createSinksFromConfiguration(Vertx vertx, FluidConfig config) { Map<String, Sink> map = new HashMap<>(); Optional<Config> sinks = config.getConfig("sinks"); if (sinks.isPresent()) { Iterator<String> names = sinks.get().names(); while (names.hasNext()) { String name = names.next(); LOGGER.info("Creating sink from configuration `" + name + "`"); Optional<Config> conf = sinks.get().getConfig(name); Sink<?> sink = buildSink(vertx, name, conf.orElseThrow(() -> new IllegalStateException("Illegal configuration for source `" + name + "`"))); map.put(name, sink); } } else { LOGGER.warn("No sinks 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);  }