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