src/main/java/net/pterodactylus/fcp/quelaton/ClientPutCommandImpl.java

   1 package net.pterodactylus.fcp.quelaton;
   2
   3 import java.io.File;
   4 import java.io.IOException;
   5 import java.io.InputStream;
   6 import java.nio.file.Files;
   7 import java.util.List;
   8 import java.util.Objects;
   9 import java.util.Optional;
  10 import java.util.concurrent.CopyOnWriteArrayList;
  11 import java.util.concurrent.ExecutionException;
  12 import java.util.concurrent.ExecutorService;
  13 import java.util.concurrent.atomic.AtomicLong;
  14 import java.util.concurrent.atomic.AtomicReference;
  15 import java.util.function.Consumer;
  16 import java.util.function.Supplier;
  17
  18 import net.pterodactylus.fcp.ClientPut;
  19 import net.pterodactylus.fcp.FcpMessage;
  20 import net.pterodactylus.fcp.Key;
  21 import net.pterodactylus.fcp.ProtocolError;
  22 import net.pterodactylus.fcp.PutFailed;
  23 import net.pterodactylus.fcp.PutSuccessful;
  24 import net.pterodactylus.fcp.TestDDAComplete;
  25 import net.pterodactylus.fcp.TestDDAReply;
  26 import net.pterodactylus.fcp.TestDDARequest;
  27 import net.pterodactylus.fcp.TestDDAResponse;
  28 import net.pterodactylus.fcp.URIGenerated;
  29 import net.pterodactylus.fcp.UploadFrom;
  30
  31 import com.google.common.util.concurrent.ListenableFuture;
  32 import com.google.common.util.concurrent.ListeningExecutorService;
  33 import com.google.common.util.concurrent.MoreExecutors;
  34
  35 /**
  36  * Default {@link ClientPutCommand} implemented based on {@link FcpDialog}.
  37  *
  38  * @author <a href="mailto:bombe@pterodactylus.net">David ‘Bombe’ Roden</a>
  39  */
  40 class ClientPutCommandImpl implements ClientPutCommand {
  41
  42         private final ListeningExecutorService threadPool;
  43         private final ConnectionSupplier connectionSupplier;
  44         private final Supplier<String> identifierGenerator;
  45         private final AtomicReference<String> redirectUri = new AtomicReference<>();
  46         private final AtomicReference<File> file = new AtomicReference<>();
  47         private final AtomicReference<InputStream> payload = new AtomicReference<>();
  48         private final AtomicLong length = new AtomicLong();
  49         private final AtomicReference<String> targetFilename = new AtomicReference<>();
  50         private final List<Consumer<String>> keyGenerateds = new CopyOnWriteArrayList<>();
  51
  52         public ClientPutCommandImpl(ExecutorService threadPool, ConnectionSupplier connectionSupplier, Supplier<String> identifierGenerator) {
  53                 this.threadPool = MoreExecutors.listeningDecorator(threadPool);
  54                 this.connectionSupplier = connectionSupplier;
  55                 this.identifierGenerator = identifierGenerator;
  56         }
  57
  58         @Override
  59         public ClientPutCommand onKeyGenerated(Consumer<String> keyGenerated) {
  60                 keyGenerateds.add(keyGenerated);
  61                 return this;
  62         }
  63
  64         @Override
  65         public ClientPutCommand named(String targetFilename) {
  66                 this.targetFilename.set(targetFilename);
  67                 return this;
  68         }
  69
  70         @Override
  71         public WithUri redirectTo(String uri) {
  72                 this.redirectUri.set(Objects.requireNonNull(uri, "uri must not be null"));
  73                 return this::key;
  74         }
  75
  76         @Override
  77         public WithUri from(File file) {
  78                 this.file.set(Objects.requireNonNull(file, "file must not be null"));
  79                 return this::key;
  80         }
  81
  82         @Override
  83         public WithLength from(InputStream inputStream) {
  84                 payload.set(Objects.requireNonNull(inputStream, "inputStream must not be null"));
  85                 return this::length;
  86         }
  87
  88         private WithUri length(long length) {
  89                 this.length.set(length);
  90                 return this::key;
  91         }
  92
  93         private Executable<Optional<Key>> key(String uri) {
  94                 return () -> threadPool.submit(() -> execute(uri));
  95         }
  96
  97         private Optional<Key> execute(String uri) throws InterruptedException, ExecutionException, IOException {
  98                 ClientPut clientPut = createClientPutCommand(uri, identifierGenerator.get());
  99                 try (ClientPutDialog clientPutDialog = new ClientPutDialog()) {
 100                         return clientPutDialog.send(clientPut).get();
 101                 }
 102         }
 103
 104         private ClientPut createClientPutCommand(String uri, String identifier) {
 105                 ClientPut clientPut;
 106                 if (file.get() != null) {
 107                         clientPut = createClientPutFromDisk(uri, identifier, file.get());
 108                 } else if (redirectUri.get() != null) {
 109                         clientPut = createClientPutRedirect(uri, identifier, redirectUri.get());
 110                 } else {
 111                         clientPut = createClientPutDirect(uri, identifier, length.get(), payload.get());
 112                 }
 113                 if (targetFilename.get() != null) {
 114                         clientPut.setTargetFilename(targetFilename.get());
 115                 }
 116                 return clientPut;
 117         }
 118
 119         private ClientPut createClientPutFromDisk(String uri, String identifier, File file) {
 120                 ClientPut clientPut = new ClientPut(uri, identifier, UploadFrom.disk);
 121                 clientPut.setFilename(file.getAbsolutePath());
 122                 return clientPut;
 123         }
 124
 125         private ClientPut createClientPutRedirect(String uri, String identifier, String redirectUri) {
 126                 ClientPut clientPut = new ClientPut(uri, identifier, UploadFrom.redirect);
 127                 clientPut.setTargetURI(redirectUri);
 128                 return clientPut;
 129         }
 130
 131         private ClientPut createClientPutDirect(String uri, String identifier, long length, InputStream payload) {
 132                 ClientPut clientPut = new ClientPut(uri, identifier, UploadFrom.direct);
 133                 clientPut.setDataLength(length);
 134                 clientPut.setPayloadInputStream(payload);
 135                 return clientPut;
 136         }
 137
 138         private class ClientPutDialog extends FcpDialog<Optional<Key>> {
 139
 140                 private final AtomicReference<FcpMessage> originalClientPut = new AtomicReference<>();
 141                 private final AtomicReference<String> directory = new AtomicReference<>();
 142
 143                 public ClientPutDialog() throws IOException {
 144                         super(ClientPutCommandImpl.this.threadPool, ClientPutCommandImpl.this.connectionSupplier.get(), Optional.<Key>empty());
 145                 }
 146
 147                 @Override
 148                 public ListenableFuture<Optional<Key>> send(FcpMessage fcpMessage) throws IOException {
 149                         originalClientPut.set(fcpMessage);
 150                         String filename = fcpMessage.getField("Filename");
 151                         if (filename != null) {
 152                                 directory.set(new File(filename).getParent());
 153                         }
 154                         return super.send(fcpMessage);
 155                 }
 156
 157                 @Override
 158                 protected void consumeURIGenerated(URIGenerated uriGenerated) {
 159                         for (Consumer<String> keyGenerated : keyGenerateds) {
 160                                 keyGenerated.accept(uriGenerated.getURI());
 161                         }
 162                 }
 163
 164                 @Override
 165                 protected void consumePutSuccessful(PutSuccessful putSuccessful) {
 166                         setResult(Optional.of(new Key(putSuccessful.getURI())));
 167                 }
 168
 169                 @Override
 170                 protected void consumePutFailed(PutFailed putFailed) {
 171                         finish();
 172                 }
 173
 174                 @Override
 175                 protected void consumeProtocolError(ProtocolError protocolError) {
 176                         if (protocolError.getCode() == 25) {
 177                                 setIdentifier(directory.get());
 178                                 sendMessage(new TestDDARequest(directory.get(), true, false));
 179                         } else {
 180                                 finish();
 181                         }
 182                 }
 183
 184                 @Override
 185                 protected void consumeTestDDAReply(TestDDAReply testDDAReply) {
 186                         try {
 187                                 String readContent = Files.readAllLines(new File(testDDAReply.getReadFilename()).toPath()).get(0);
 188                                 sendMessage(new TestDDAResponse(directory.get(), readContent));
 189                         } catch (IOException e) {
 190                                 sendMessage(new TestDDAResponse(directory.get(), "failed-to-read"));
 191                         }
 192                 }
 193
 194                 @Override
 195                 protected void consumeTestDDAComplete(TestDDAComplete testDDAComplete) {
 196                         setIdentifier(originalClientPut.get().getField("Identifier"));
 197                         sendMessage(originalClientPut.get());
 198                 }
 199
 200         }
 201
 202 }