ProxyService { static void pumpOneToMany(ReadStream<Buffer> readStream, List<WriteStream<Buffer>> writeStreams) { readStream.handler(data -> { for (WriteStream<Buffer> w : writeStreams) { w.write(data); } pauseAndResume(readStream, writeStreams); }); readStream.endHandler(v -> { for (WriteStream<Buffer> w : writeStreams) { w.end(); } }); } ProxyService(Vertx vertx, ModuleManager modules, TenantManager tm,
                      DiscoveryManager dm, InternalModule im, String okapiUrl, JsonObject config); void proxy(RoutingContext ctx); Future<OkapiClient> callSystemInterface(Tenant tenant, ModuleInstance inst,
                                                 String request, ProxyContext pc); void redirectProxy(RoutingContext ctx); Future<Void> autoDeploy(ModuleDescriptor md); Future<Void> autoUndeploy(ModuleDescriptor md);  }
@Test public void pumpOneToManySimple() { MyReadStream readStream = new MyReadStream(); MyWriteStream writeStream1 = new MyWriteStream(); MyWriteStream writeStream2 = new MyWriteStream(); ProxyService.pumpOneToMany(readStream, Arrays.asList(writeStream1, writeStream2)); readStream.handle(4); assertThat(writeStream1.overflow).isTrue(); assertThat(writeStream2.overflow).isTrue(); assertThat(readStream.pause).isTrue(); writeStream1.processQueue(4); assertThat(writeStream1.overflow).isFalse(); assertThat(writeStream2.overflow).isTrue(); assertThat(readStream.pause).isTrue(); writeStream2.processQueue(4); assertThat(writeStream1.overflow).isFalse(); assertThat(writeStream2.overflow).isFalse(); assertThat(readStream.pause).isFalse(); readStream.handle(1); assertThat(writeStream1.end).isFalse(); assertThat(writeStream2.end).isFalse(); readStream.endHandler.handle(null); assertThat(writeStream1.end).isTrue(); assertThat(writeStream2.end).isTrue(); }
@Test public void pumpOneToManyPiledUp() { MyReadStream readStream = new MyReadStream(); MyWriteStream writeStream1 = new MyWriteStream(); MyWriteStream writeStream2 = new MyWriteStream(); ProxyService.pumpOneToMany(readStream, Arrays.asList(writeStream1, writeStream2)); readStream.handle(4); writeStream1.processQueue(4); writeStream1.write(null); writeStream1.write(null); writeStream1.write(null); writeStream1.write(null); writeStream1.processQueue(4); assertThat(writeStream2.writeQueueFull()).isTrue(); assertThat(writeStream2.overflow).isTrue(); assertThat(readStream.pause).isTrue(); }