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

   1 package net.pterodactylus.fcp.quelaton;
   2
   3 import java.io.IOException;
   4 import java.util.Optional;
   5 import java.util.concurrent.ExecutionException;
   6 import java.util.concurrent.ExecutorService;
   7 import java.util.function.Supplier;
   8
   9 import net.pterodactylus.fcp.IdentifierCollision;
  10 import net.pterodactylus.fcp.SubscribeUSK;
  11 import net.pterodactylus.fcp.SubscribedUSK;
  12
  13 import com.google.common.util.concurrent.ListenableFuture;
  14 import com.google.common.util.concurrent.ListeningExecutorService;
  15 import com.google.common.util.concurrent.MoreExecutors;
  16
  17 /**
  18  * Default {@link SubscribeUskCommand} implementation based on {@link FcpDialog}.
  19  *
  20  * @author <a href="mailto:bombe@freenetproject.org">David ‘Bombe’ Roden</a>
  21  */
  22 public class SubscribeUskCommandImpl implements SubscribeUskCommand {
  23
  24         private final ListeningExecutorService threadPool;
  25         private final ConnectionSupplier connectionSupplier;
  26         private final SubscribeUSK subscribeUSK;
  27         private final ActiveSubscriptions activeSubscriptions;
  28
  29         public SubscribeUskCommandImpl(
  30                 ExecutorService threadPool, ConnectionSupplier connectionSupplier, Supplier<String> identifierGenerator,
  31                 ActiveSubscriptions activeSubscriptions) {
  32                 this.activeSubscriptions = activeSubscriptions;
  33                 this.threadPool = MoreExecutors.listeningDecorator(threadPool);
  34                 this.connectionSupplier = connectionSupplier;
  35                 subscribeUSK = new SubscribeUSK(identifierGenerator.get());
  36         }
  37
  38         @Override
  39         public Executable<Optional<UskSubscription>> uri(String uri) {
  40                 subscribeUSK.setUri(uri);
  41                 return this::execute;
  42         }
  43
  44         private ListenableFuture<Optional<UskSubscription>> execute() {
  45                 return threadPool.submit(this::executeDialog);
  46         }
  47
  48         private Optional<UskSubscription> executeDialog() throws IOException, ExecutionException, InterruptedException {
  49                 try (SubscribeUskDialog subscribeUskDialog = new SubscribeUskDialog()) {
  50                         if (subscribeUskDialog.send(subscribeUSK).get()) {
  51                                 UskSubscription uskSubscription = activeSubscriptions.createUskSubscription(subscribeUSK);
  52                                 return Optional.of(uskSubscription);
  53                         }
  54                         return Optional.empty();
  55                 }
  56         }
  57
  58         private class SubscribeUskDialog extends FcpDialog<Boolean> {
  59
  60                 public SubscribeUskDialog() throws IOException {
  61                         super(threadPool, connectionSupplier.get(), false);
  62                 }
  63
  64                 @Override
  65                 protected void consumeSubscribedUSK(SubscribedUSK subscribedUSK) {
  66                         setResult(true);
  67                 }
  68
  69                 @Override
  70                 protected void consumeIdentifierCollision(IdentifierCollision identifierCollision) {
  71                         finish();
  72                 }
  73
  74         }
  75
  76 }