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.concurrent.atomic.AtomicBoolean;
   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 static final RandomIdentifierGenerator IDENTIFIER = new RandomIdentifierGenerator();
  25         private final ListeningExecutorService threadPool;
  26         private final ConnectionSupplier connectionSupplier;
  27         private final SubscribeUSK subscribeUSK = new SubscribeUSK(IDENTIFIER.generate());
  28         private final ActiveSubscriptions activeSubscriptions;
  29
  30         public SubscribeUskCommandImpl(
  31                 ExecutorService threadPool, ConnectionSupplier connectionSupplier,
  32                 ActiveSubscriptions activeSubscriptions) {
  33                 this.activeSubscriptions = activeSubscriptions;
  34                 this.threadPool = MoreExecutors.listeningDecorator(threadPool);
  35                 this.connectionSupplier = connectionSupplier;
  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                 private final AtomicBoolean finished = new AtomicBoolean();
  61                 private final AtomicBoolean success = new AtomicBoolean();
  62
  63                 public SubscribeUskDialog() throws IOException {
  64                         super(threadPool, connectionSupplier.get());
  65                 }
  66
  67                 @Override
  68                 protected boolean isFinished() {
  69                         return finished.get();
  70                 }
  71
  72                 @Override
  73                 protected Boolean getResult() {
  74                         return success.get();
  75                 }
  76
  77                 @Override
  78                 protected void consumeSubscribedUSK(SubscribedUSK subscribedUSK) {
  79                         success.set(true);
  80                         finished.set(true);
  81                 }
  82
  83                 @Override
  84                 protected void consumeIdentifierCollision(IdentifierCollision identifierCollision) {
  85                         finished.set(true);
  86                 }
  87
  88         }
  89
  90 }