X-Git-Url: https://git.pterodactylus.net/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fnet%2Fpterodactylus%2Ffcp%2Fquelaton%2FClientPutCommandImpl.java;h=06d54e0b4ef2f4811eca52b3cc3b85bc445f5642;hb=60bf94d38d9c17dfba81e20564e98337714ae332;hp=02d020c78dee9cbce579bf59bc8220e7a42907aa;hpb=bbe77c263f9f37b2407a6b5ce2f5e68e11bd9e7c;p=jFCPlib.git diff --git a/src/main/java/net/pterodactylus/fcp/quelaton/ClientPutCommandImpl.java b/src/main/java/net/pterodactylus/fcp/quelaton/ClientPutCommandImpl.java index 02d020c..06d54e0 100644 --- a/src/main/java/net/pterodactylus/fcp/quelaton/ClientPutCommandImpl.java +++ b/src/main/java/net/pterodactylus/fcp/quelaton/ClientPutCommandImpl.java @@ -4,12 +4,16 @@ import java.io.File; import java.io.IOException; import java.io.InputStream; import java.nio.file.Files; +import java.util.List; import java.util.Objects; import java.util.Optional; +import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicLong; import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Consumer; import net.pterodactylus.fcp.ClientPut; import net.pterodactylus.fcp.FcpMessage; @@ -21,6 +25,7 @@ import net.pterodactylus.fcp.TestDDAComplete; import net.pterodactylus.fcp.TestDDAReply; import net.pterodactylus.fcp.TestDDARequest; import net.pterodactylus.fcp.TestDDAResponse; +import net.pterodactylus.fcp.URIGenerated; import net.pterodactylus.fcp.UploadFrom; import com.google.common.util.concurrent.ListenableFuture; @@ -28,7 +33,7 @@ import com.google.common.util.concurrent.ListeningExecutorService; import com.google.common.util.concurrent.MoreExecutors; /** - * Default {@link ClientPutCommand} implemented based on {@link FcpReplySequence}. + * Default {@link ClientPutCommand} implemented based on {@link FcpDialog}. * * @author David ‘Bombe’ Roden */ @@ -41,6 +46,7 @@ class ClientPutCommandImpl implements ClientPutCommand { private final AtomicReference payload = new AtomicReference<>(); private final AtomicLong length = new AtomicLong(); private final AtomicReference targetFilename = new AtomicReference<>(); + private final List> keyGenerateds = new CopyOnWriteArrayList<>(); public ClientPutCommandImpl(ExecutorService threadPool, ConnectionSupplier connectionSupplier) { this.threadPool = MoreExecutors.listeningDecorator(threadPool); @@ -48,38 +54,50 @@ class ClientPutCommandImpl implements ClientPutCommand { } @Override + public ClientPutCommand onKeyGenerated(Consumer keyGenerated) { + keyGenerateds.add(keyGenerated); + return this; + } + + @Override public ClientPutCommand named(String targetFilename) { this.targetFilename.set(targetFilename); return this; } @Override - public WithUri>> redirectTo(String uri) { + public WithUri redirectTo(String uri) { this.redirectUri.set(Objects.requireNonNull(uri, "uri must not be null")); return this::key; } @Override - public WithUri>> from(File file) { + public WithUri from(File file) { this.file.set(Objects.requireNonNull(file, "file must not be null")); return this::key; } @Override - public WithLength>>> from(InputStream inputStream) { + public WithLength from(InputStream inputStream) { payload.set(Objects.requireNonNull(inputStream, "inputStream must not be null")); return this::length; } - private WithUri>> length(long length) { + private WithUri length(long length) { this.length.set(length); return this::key; } private Executable> key(String uri) { + return () -> threadPool.submit(() -> execute(uri)); + } + + private Optional execute(String uri) throws InterruptedException, ExecutionException, IOException { String identifier = new RandomIdentifierGenerator().generate(); ClientPut clientPut = createClientPutCommand(uri, identifier); - return () -> threadPool.submit(() -> new ClientPutReplySequence().send(clientPut).get()); + try (ClientPutDialog clientPutDialog = new ClientPutDialog()) { + return clientPutDialog.send(clientPut).get(); + } } private ClientPut createClientPutCommand(String uri, String identifier) { @@ -116,14 +134,14 @@ class ClientPutCommandImpl implements ClientPutCommand { return clientPut; } - private class ClientPutReplySequence extends FcpReplySequence> { + private class ClientPutDialog extends FcpDialog> { private final AtomicReference originalClientPut = new AtomicReference<>(); private final AtomicReference directory = new AtomicReference<>(); private final AtomicReference finalKey = new AtomicReference<>(); private final AtomicBoolean putFinished = new AtomicBoolean(); - public ClientPutReplySequence() throws IOException { + public ClientPutDialog() throws IOException { super(ClientPutCommandImpl.this.threadPool, ClientPutCommandImpl.this.connectionSupplier.get()); } @@ -148,6 +166,13 @@ class ClientPutCommandImpl implements ClientPutCommand { } @Override + protected void consumeURIGenerated(URIGenerated uriGenerated) { + for (Consumer keyGenerated : keyGenerateds) { + keyGenerated.accept(uriGenerated.getURI()); + } + } + + @Override protected void consumePutSuccessful(PutSuccessful putSuccessful) { finalKey.set(new Key(putSuccessful.getURI())); putFinished.set(true);