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