@Test @Ignore("this test uses lots of local ports (and all of them are in TIME_WAIT state after it for a minute) so HTTP tests after it may fail indefinitely") public void testUploadAlot() throws IOException { String content = "test content of the file"; ByteBuf data = ByteBuf.wrapForReading(content.getBytes(UTF_8)); await(Promises.sequence(IntStream.range(0, 1000) .mapToObj(i -> () -> ChannelSupplier.of(data.slice()) .streamTo(ChannelConsumer.ofPromise(client.upload("file_uploaded_" + i + ".txt"))))) .whenComplete(() -> servers.forEach(AbstractServer::close))); for (int i = 0; i < CLIENT_SERVER_PAIRS; i++) { for (int j = 0; j < 1000; j++) { assertEquals(new String(readAllBytes(serverStorages[i].resolve("file_uploaded_" + i + ".txt")), UTF_8), content); } } }
private Promise<ChannelConsumer<ByteBuf>> upload(@NotNull String filename, long offset, @Nullable Long revision) { List<Object> selected = serverSelector.selectFrom(filename, aliveClients.keySet(), replicationCount); checkState(!selected.isEmpty(), "Selected no servers to upload file " + filename); checkState(aliveClients.keySet().containsAll(selected), "Selected an id that is not one of client ids"); class ConsumerWithId { final Object id; final ChannelConsumer<ByteBuf> consumer; ConsumerWithId(Object id, ChannelConsumer<ByteBuf> consumer) { this.id = id; this.consumer = consumer; } } return Promises.toList(selected.stream() .map(id -> { FsClient client = aliveClients.get(id); return (revision == null ? client.upload(filename, offset) : client.upload(filename, offset, revision)) .thenEx(wrapDeath(id)) .map(consumer -> new ConsumerWithId(id, consumer.withAcknowledgement(ack -> ack.whenException(e -> markIfDead(id, e))))) .toTry(); })) .then(tries -> { List<ConsumerWithId> successes = tries.stream() .filter(Try::isSuccess) .map(Try::get) .collect(toList()); if (successes.isEmpty()) { return ofFailure("Couldn't connect to any partition to upload file " + filename, tries); } ChannelSplitter<ByteBuf> splitter = ChannelSplitter.<ByteBuf>create().lenient(); Promise<List<Try<Void>>> uploadResults = Promises.toList(successes.stream() .map(s1 -> getAcknowledgement(fn -> splitter.addOutput() .set(s1.consumer.withAcknowledgement(fn))) .toTry())); if (logger.isTraceEnabled()) { logger.trace("uploading file {} to {}, {}", filename, successes.stream().map(s -> s.id.toString()).collect(joining(", ", "[", "]")), this); } ChannelConsumer<ByteBuf> consumer = splitter.getInput().getConsumer(); return Promise.of(consumer.withAcknowledgement(ack -> ack .then(() -> uploadResults) .then(ackTries -> { long successCount = ackTries.stream().filter(Try::isSuccess).count(); if (ackTries.size() < replicationCount) { return ofFailure("Didn't connect to enough partitions uploading " + filename + ", only " + successCount + " finished uploads", ackTries); } if (successCount < replicationCount) { return ofFailure("Couldn't finish uploadind file " + filename + ", only " + successCount + " acknowlegdes received", ackTries); } return Promise.complete(); }) .whenComplete(uploadFinishPromise.recordStats()))); }) .whenComplete(uploadStartPromise.recordStats()); }
RemoteFsClusterClient implements FsClient, Initializable<RemoteFsClusterClient>, EventloopService, EventloopJmxBeanEx { private Promise<ChannelConsumer<ByteBuf>> upload(@NotNull String filename, long offset, @Nullable Long revision) { List<Object> selected = serverSelector.selectFrom(filename, aliveClients.keySet(), replicationCount); checkState(!selected.isEmpty(), "Selected no servers to upload file " + filename); checkState(aliveClients.keySet().containsAll(selected), "Selected an id that is not one of client ids"); class ConsumerWithId { final Object id; final ChannelConsumer<ByteBuf> consumer; ConsumerWithId(Object id, ChannelConsumer<ByteBuf> consumer) { this.id = id; this.consumer = consumer; } } return Promises.toList(selected.stream() .map(id -> { FsClient client = aliveClients.get(id); return (revision == null ? client.upload(filename, offset) : client.upload(filename, offset, revision)) .thenEx(wrapDeath(id)) .map(consumer -> new ConsumerWithId(id, consumer.withAcknowledgement(ack -> ack.whenException(e -> markIfDead(id, e))))) .toTry(); })) .then(tries -> { List<ConsumerWithId> successes = tries.stream() .filter(Try::isSuccess) .map(Try::get) .collect(toList()); if (successes.isEmpty()) { return ofFailure("Couldn't connect to any partition to upload file " + filename, tries); } ChannelSplitter<ByteBuf> splitter = ChannelSplitter.<ByteBuf>create().lenient(); Promise<List<Try<Void>>> uploadResults = Promises.toList(successes.stream() .map(s1 -> getAcknowledgement(fn -> splitter.addOutput() .set(s1.consumer.withAcknowledgement(fn))) .toTry())); if (logger.isTraceEnabled()) { logger.trace("uploading file {} to {}, {}", filename, successes.stream().map(s -> s.id.toString()).collect(joining(", ", "[", "]")), this); } ChannelConsumer<ByteBuf> consumer = splitter.getInput().getConsumer(); return Promise.of(consumer.withAcknowledgement(ack -> ack .then(() -> uploadResults) .then(ackTries -> { long successCount = ackTries.stream().filter(Try::isSuccess).count(); if (ackTries.size() < replicationCount) { return ofFailure("Didn't connect to enough partitions uploading " + filename + ", only " + successCount + " finished uploads", ackTries); } if (successCount < replicationCount) { return ofFailure("Couldn't finish uploadind file " + filename + ", only " + successCount + " acknowlegdes received", ackTries); } return Promise.complete(); }) .whenComplete(uploadFinishPromise.recordStats()))); }) .whenComplete(uploadStartPromise.recordStats()); } }
RemoteFsClusterClient implements FsClient, Initializable<RemoteFsClusterClient>, EventloopService, EventloopJmxBeanEx { private Promise<ChannelConsumer<ByteBuf>> upload(@NotNull String filename, long offset, @Nullable Long revision) { List<Object> selected = serverSelector.selectFrom(filename, aliveClients.keySet(), replicationCount); checkState(!selected.isEmpty(), "Selected no servers to upload file " + filename); checkState(aliveClients.keySet().containsAll(selected), "Selected an id that is not one of client ids"); class ConsumerWithId { final Object id; final ChannelConsumer<ByteBuf> consumer; ConsumerWithId(Object id, ChannelConsumer<ByteBuf> consumer) { this.id = id; this.consumer = consumer; } } return Promises.toList(selected.stream() .map(id -> { FsClient client = aliveClients.get(id); return (revision == null ? client.upload(filename, offset) : client.upload(filename, offset, revision)) .thenEx(wrapDeath(id)) .map(consumer -> new ConsumerWithId(id, consumer.withAcknowledgement(ack -> ack.whenException(e -> markIfDead(id, e))))) .toTry(); })) .then(tries -> { List<ConsumerWithId> successes = tries.stream() .filter(Try::isSuccess) .map(Try::get) .collect(toList()); if (successes.isEmpty()) { return ofFailure("Couldn't connect to any partition to upload file " + filename, tries); } ChannelSplitter<ByteBuf> splitter = ChannelSplitter.<ByteBuf>create().lenient(); Promise<List<Try<Void>>> uploadResults = Promises.toList(successes.stream() .map(s1 -> getAcknowledgement(fn -> splitter.addOutput() .set(s1.consumer.withAcknowledgement(fn))) .toTry())); if (logger.isTraceEnabled()) { logger.trace("uploading file {} to {}, {}", filename, successes.stream().map(s -> s.id.toString()).collect(joining(", ", "[", "]")), this); } ChannelConsumer<ByteBuf> consumer = splitter.getInput().getConsumer(); return Promise.of(consumer.withAcknowledgement(ack -> ack .then(() -> uploadResults) .then(ackTries -> { long successCount = ackTries.stream().filter(Try::isSuccess).count(); if (ackTries.size() < replicationCount) { return ofFailure("Didn't connect to enough partitions uploading " + filename + ", only " + successCount + " finished uploads", ackTries); } if (successCount < replicationCount) { return ofFailure("Couldn't finish uploadind file " + filename + ", only " + successCount + " acknowlegdes received", ackTries); } return Promise.complete(); }) .whenComplete(uploadFinishPromise.recordStats()))); }) .whenComplete(uploadStartPromise.recordStats()); } private  RemoteFsClusterClient(Eventloop eventloop, Map<Object, FsClient> clients); }
RemoteFsClusterClient implements FsClient, Initializable<RemoteFsClusterClient>, EventloopService, EventloopJmxBeanEx { private Promise<ChannelConsumer<ByteBuf>> upload(@NotNull String filename, long offset, @Nullable Long revision) { List<Object> selected = serverSelector.selectFrom(filename, aliveClients.keySet(), replicationCount); checkState(!selected.isEmpty(), "Selected no servers to upload file " + filename); checkState(aliveClients.keySet().containsAll(selected), "Selected an id that is not one of client ids"); class ConsumerWithId { final Object id; final ChannelConsumer<ByteBuf> consumer; ConsumerWithId(Object id, ChannelConsumer<ByteBuf> consumer) { this.id = id; this.consumer = consumer; } } return Promises.toList(selected.stream() .map(id -> { FsClient client = aliveClients.get(id); return (revision == null ? client.upload(filename, offset) : client.upload(filename, offset, revision)) .thenEx(wrapDeath(id)) .map(consumer -> new ConsumerWithId(id, consumer.withAcknowledgement(ack -> ack.whenException(e -> markIfDead(id, e))))) .toTry(); })) .then(tries -> { List<ConsumerWithId> successes = tries.stream() .filter(Try::isSuccess) .map(Try::get) .collect(toList()); if (successes.isEmpty()) { return ofFailure("Couldn't connect to any partition to upload file " + filename, tries); } ChannelSplitter<ByteBuf> splitter = ChannelSplitter.<ByteBuf>create().lenient(); Promise<List<Try<Void>>> uploadResults = Promises.toList(successes.stream() .map(s1 -> getAcknowledgement(fn -> splitter.addOutput() .set(s1.consumer.withAcknowledgement(fn))) .toTry())); if (logger.isTraceEnabled()) { logger.trace("uploading file {} to {}, {}", filename, successes.stream().map(s -> s.id.toString()).collect(joining(", ", "[", "]")), this); } ChannelConsumer<ByteBuf> consumer = splitter.getInput().getConsumer(); return Promise.of(consumer.withAcknowledgement(ack -> ack .then(() -> uploadResults) .then(ackTries -> { long successCount = ackTries.stream().filter(Try::isSuccess).count(); if (ackTries.size() < replicationCount) { return ofFailure("Didn't connect to enough partitions uploading " + filename + ", only " + successCount + " finished uploads", ackTries); } if (successCount < replicationCount) { return ofFailure("Couldn't finish uploadind file " + filename + ", only " + successCount + " acknowlegdes received", ackTries); } return Promise.complete(); }) .whenComplete(uploadFinishPromise.recordStats()))); }) .whenComplete(uploadStartPromise.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(); }
RemoteFsClusterClient implements FsClient, Initializable<RemoteFsClusterClient>, EventloopService, EventloopJmxBeanEx { private Promise<ChannelConsumer<ByteBuf>> upload(@NotNull String filename, long offset, @Nullable Long revision) { List<Object> selected = serverSelector.selectFrom(filename, aliveClients.keySet(), replicationCount); checkState(!selected.isEmpty(), "Selected no servers to upload file " + filename); checkState(aliveClients.keySet().containsAll(selected), "Selected an id that is not one of client ids"); class ConsumerWithId { final Object id; final ChannelConsumer<ByteBuf> consumer; ConsumerWithId(Object id, ChannelConsumer<ByteBuf> consumer) { this.id = id; this.consumer = consumer; } } return Promises.toList(selected.stream() .map(id -> { FsClient client = aliveClients.get(id); return (revision == null ? client.upload(filename, offset) : client.upload(filename, offset, revision)) .thenEx(wrapDeath(id)) .map(consumer -> new ConsumerWithId(id, consumer.withAcknowledgement(ack -> ack.whenException(e -> markIfDead(id, e))))) .toTry(); })) .then(tries -> { List<ConsumerWithId> successes = tries.stream() .filter(Try::isSuccess) .map(Try::get) .collect(toList()); if (successes.isEmpty()) { return ofFailure("Couldn't connect to any partition to upload file " + filename, tries); } ChannelSplitter<ByteBuf> splitter = ChannelSplitter.<ByteBuf>create().lenient(); Promise<List<Try<Void>>> uploadResults = Promises.toList(successes.stream() .map(s1 -> getAcknowledgement(fn -> splitter.addOutput() .set(s1.consumer.withAcknowledgement(fn))) .toTry())); if (logger.isTraceEnabled()) { logger.trace("uploading file {} to {}, {}", filename, successes.stream().map(s -> s.id.toString()).collect(joining(", ", "[", "]")), this); } ChannelConsumer<ByteBuf> consumer = splitter.getInput().getConsumer(); return Promise.of(consumer.withAcknowledgement(ack -> ack .then(() -> uploadResults) .then(ackTries -> { long successCount = ackTries.stream().filter(Try::isSuccess).count(); if (ackTries.size() < replicationCount) { return ofFailure("Didn't connect to enough partitions uploading " + filename + ", only " + successCount + " finished uploads", ackTries); } if (successCount < replicationCount) { return ofFailure("Couldn't finish uploadind file " + filename + ", only " + successCount + " acknowlegdes received", ackTries); } return Promise.complete(); }) .whenComplete(uploadFinishPromise.recordStats()))); }) .whenComplete(uploadStartPromise.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();  }