RemoteFsClusterClient implements FsClient, Initializable<RemoteFsClusterClient>, EventloopService, EventloopJmxBeanEx { @Override public Promise<ChannelSupplier<ByteBuf>> download(@NotNull String name, long offset, long length) { if (deadClients.size() >= replicationCount) { return ofFailure("There are more dead partitions than replication count(" + deadClients.size() + " dead, replication count is " + replicationCount + "), aborting", emptyList()); } return Promises.toList( aliveClients.entrySet().stream() .map(entry -> { Object partitionId = entry.getKey(); return entry.getValue().getMetadata(name) .map(res -> res != null ? new Tuple2<>(partitionId, res) : null) .thenEx(wrapDeath(partitionId)) .toTry(); })) .then(tries -> { List<Tuple2<Object, FileMetadata>> successes = tries.stream() .filter(Try::isSuccess) .map(Try::get) .collect(toList()); if (deadClients.size() >= replicationCount) { return ofFailure("There are more dead partitions than replication count(" + deadClients.size() + " dead, replication count is " + replicationCount + "), aborting", tries); } List<Tuple2<Object, FileMetadata>> found = successes.stream().filter(Objects::nonNull).collect(toList()); Optional<Tuple2<Object, FileMetadata>> maybeBest = found.stream() .max(Comparator.comparing(Tuple2::getValue2, FileMetadata.COMPARATOR)); if (!maybeBest.isPresent()) { return ofFailure("File not found: " + name, tries); } Tuple2<Object, FileMetadata> best = maybeBest.get(); return Promises.any(found.stream() .filter(piwfs -> piwfs.getValue2().getRevision() == best.getValue2().getRevision()) .map(piwfs -> { FsClient client = aliveClients.get(piwfs.getValue1()); if (client == null) { return Promise.ofException(new StacklessException(RemoteFsClusterClient.class, "Client " + piwfs.getValue1() + " is not alive")); } logger.trace("downloading file {} from {}", name, piwfs.getValue1()); return client.download(name, offset, length) .whenException(e -> logger.warn("Failed to connect to server with key " + piwfs.getValue1() + " to download file " + name, e)) .thenEx(wrapDeath(piwfs.getValue1())) .map(supplier -> supplier .withEndOfStream(eos -> eos .whenException(e -> markIfDead(piwfs.getValue1(), e)) .whenComplete(downloadFinishPromise.recordStats()))); }), AsyncCloseable::close); }) .whenComplete(downloadStartPromise.recordStats()); } private  RemoteFsClusterClient(Eventloop eventloop, Map<Object, FsClient> clients); static RemoteFsClusterClient create(Eventloop eventloop); static RemoteFsClusterClient create(Eventloop eventloop, Map<Object, FsClient> clients); RemoteFsClusterClient withPartition(Object id, FsClient client); RemoteFsClusterClient withReplicationCount(int replicationCount); RemoteFsClusterClient withServerSelector(@NotNull ServerSelector serverSelector); @NotNull @Override Eventloop getEventloop(); Map<Object, FsClient> getClients(); Map<Object, FsClient> getAliveClients(); Map<Object, FsClient> getDeadClients(); ServerSelector getServerSelector(); Promise<Void> checkAllPartitions(); Promise<Void> checkDeadPartitions(); boolean markDead(Object partitionId, @Nullable Throwable e); @Override Promise<ChannelConsumer<ByteBuf>> upload(@NotNull String name, long offset); @Override Promise<ChannelConsumer<ByteBuf>> upload(@NotNull String name, long offset, long revision); @Override Promise<ChannelSupplier<ByteBuf>> download(@NotNull String name, long offset, long length); @Override Promise<Void> move(@NotNull String name, @NotNull String target, long targetRevision, long tombstoneRevision); @Override Promise<Void> copy(@NotNull String name, @NotNull String target, long targetRevision); @Override Promise<Void> delete(@NotNull String name, long revision); @Override Promise<List<FileMetadata>> listEntities(@NotNull String glob); @Override Promise<List<FileMetadata>> list(@NotNull String glob); @Override Promise<Void> ping(); @NotNull @Override Promise<Void> start(); @NotNull @Override Promise<Void> stop(); @Override String toString(); @JmxAttribute int getReplicationCount(); @JmxAttribute void setReplicationCount(int replicationCount); @JmxAttribute int getAlivePartitionCount(); @JmxAttribute int getDeadPartitionCount(); @JmxAttribute String[] getAlivePartitions(); @JmxAttribute String[] getDeadPartitions(); @JmxAttribute PromiseStats getConnectPromise(); @JmxAttribute PromiseStats getUploadStartPromise(); @JmxAttribute PromiseStats getUploadFinishPromise(); @JmxAttribute PromiseStats getDownloadStartPromise(); @JmxAttribute PromiseStats getDownloadFinishPromise(); @JmxAttribute PromiseStats getMovePromise(); @JmxAttribute PromiseStats getCopyPromise(); @JmxAttribute PromiseStats getListPromise(); @JmxAttribute PromiseStats getDeletePromise();  }
@Test public void testDownload() throws IOException { int numOfServer = 3; String file = "the_file.txt"; String content = "another test content of the file"; Files.write(serverStorages[numOfServer].resolve(file), content.getBytes(UTF_8)); await(ChannelSupplier.ofPromise(client.download(file, 0)) .streamTo(ChannelFileWriter.open(newCachedThreadPool(), clientStorage.resolve(file))) .whenComplete(() -> servers.forEach(AbstractServer::close))); assertEquals(new String(readAllBytes(clientStorage.resolve(file)), UTF_8), content); }
@Test public void downloadNonExisting() { String fileName = "i_dont_exist.txt"; Throwable exception = awaitException(ChannelSupplier.ofPromise(client.download(fileName)) .streamTo(ChannelConsumer.of(AsyncConsumer.of(ByteBuf::recycle))) .whenComplete(() -> servers.forEach(AbstractServer::close))); assertThat(exception, instanceOf(StacklessException.class)); assertThat(exception.getMessage(), containsString(fileName)); }