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

   1 package net.pterodactylus.fcp.quelaton;
   2
   3 import java.io.IOException;
   4 import java.util.Objects;
   5 import java.util.Queue;
   6 import java.util.concurrent.ConcurrentLinkedQueue;
   7 import java.util.concurrent.ExecutionException;
   8 import java.util.concurrent.ExecutorService;
   9 import java.util.concurrent.atomic.AtomicBoolean;
  10 import java.util.concurrent.atomic.AtomicReference;
  11 import java.util.function.Consumer;
  12
  13 import net.pterodactylus.fcp.AllData;
  14 import net.pterodactylus.fcp.BaseMessage;
  15 import net.pterodactylus.fcp.CloseConnectionDuplicateClientName;
  16 import net.pterodactylus.fcp.ConfigData;
  17 import net.pterodactylus.fcp.DataFound;
  18 import net.pterodactylus.fcp.EndListPeerNotes;
  19 import net.pterodactylus.fcp.EndListPeers;
  20 import net.pterodactylus.fcp.EndListPersistentRequests;
  21 import net.pterodactylus.fcp.FCPPluginReply;
  22 import net.pterodactylus.fcp.FcpConnection;
  23 import net.pterodactylus.fcp.FcpListener;
  24 import net.pterodactylus.fcp.FcpMessage;
  25 import net.pterodactylus.fcp.FinishedCompression;
  26 import net.pterodactylus.fcp.GetFailed;
  27 import net.pterodactylus.fcp.IdentifierCollision;
  28 import net.pterodactylus.fcp.NodeData;
  29 import net.pterodactylus.fcp.NodeHello;
  30 import net.pterodactylus.fcp.Peer;
  31 import net.pterodactylus.fcp.PeerNote;
  32 import net.pterodactylus.fcp.PeerRemoved;
  33 import net.pterodactylus.fcp.PersistentGet;
  34 import net.pterodactylus.fcp.PersistentPut;
  35 import net.pterodactylus.fcp.PersistentPutDir;
  36 import net.pterodactylus.fcp.PersistentRequestModified;
  37 import net.pterodactylus.fcp.PersistentRequestRemoved;
  38 import net.pterodactylus.fcp.PluginInfo;
  39 import net.pterodactylus.fcp.PluginRemoved;
  40 import net.pterodactylus.fcp.ProtocolError;
  41 import net.pterodactylus.fcp.PutFailed;
  42 import net.pterodactylus.fcp.PutFetchable;
  43 import net.pterodactylus.fcp.PutSuccessful;
  44 import net.pterodactylus.fcp.ReceivedBookmarkFeed;
  45 import net.pterodactylus.fcp.SSKKeypair;
  46 import net.pterodactylus.fcp.SentFeed;
  47 import net.pterodactylus.fcp.SimpleProgress;
  48 import net.pterodactylus.fcp.StartedCompression;
  49 import net.pterodactylus.fcp.SubscribedUSK;
  50 import net.pterodactylus.fcp.SubscribedUSKUpdate;
  51 import net.pterodactylus.fcp.TestDDAComplete;
  52 import net.pterodactylus.fcp.TestDDAReply;
  53 import net.pterodactylus.fcp.URIGenerated;
  54 import net.pterodactylus.fcp.UnknownNodeIdentifier;
  55 import net.pterodactylus.fcp.UnknownPeerNoteType;
  56
  57 import com.google.common.util.concurrent.ListenableFuture;
  58 import com.google.common.util.concurrent.ListeningExecutorService;
  59 import com.google.common.util.concurrent.MoreExecutors;
  60
  61 /**
  62  * An FCP dialog enables you to conveniently wait for a specific set of FCP replies.
  63  *
  64  * @author <a href="bombe@freenetproject.org">David ‘Bombe’ Roden</a>
  65  */
  66 public abstract class FcpDialog<R> implements AutoCloseable, FcpListener {
  67
  68         private final Object syncObject = new Object();
  69         private final ListeningExecutorService executorService;
  70         private final FcpConnection fcpConnection;
  71         private final Queue<FcpMessage> messages = new ConcurrentLinkedQueue<>();
  72         private final AtomicReference<String> identifier = new AtomicReference<>();
  73         private final AtomicBoolean connectionClosed = new AtomicBoolean();
  74         private final AtomicReference<Throwable> connectionFailureReason = new AtomicReference<>();
  75
  76         public FcpDialog(ExecutorService executorService, FcpConnection fcpConnection) {
  77                 this.executorService = MoreExecutors.listeningDecorator(executorService);
  78                 this.fcpConnection = fcpConnection;
  79         }
  80
  81         protected void setIdentifier(String identifier) {
  82                 this.identifier.set(identifier);
  83         }
  84
  85         protected abstract boolean isFinished();
  86
  87         public ListenableFuture<R> send(FcpMessage fcpMessage) throws IOException {
  88                 setIdentifier(fcpMessage.getField("Identifier"));
  89                 fcpConnection.addFcpListener(this);
  90                 messages.add(fcpMessage);
  91                 return executorService.submit(() -> {
  92                         synchronized (syncObject) {
  93                                 while (!connectionClosed.get() && (!isFinished() || !messages.isEmpty())) {
  94                                         while (messages.peek() != null) {
  95                                                 FcpMessage message = messages.poll();
  96                                                 fcpConnection.sendMessage(message);
  97                                         }
  98                                         if (isFinished() || connectionClosed.get()) {
  99                                                 continue;
 100                                         }
 101                                         syncObject.wait();
 102                                 }
 103                         }
 104                         Throwable throwable = connectionFailureReason.get();
 105                         if (throwable != null) {
 106                                 throw new ExecutionException(throwable);
 107                         }
 108                         return getResult();
 109                 });
 110         }
 111
 112         protected void sendMessage(FcpMessage fcpMessage) {
 113                 messages.add(fcpMessage);
 114                 notifySyncObject();
 115         }
 116
 117         private void notifySyncObject() {
 118                 synchronized (syncObject) {
 119                         syncObject.notifyAll();
 120                 }
 121         }
 122
 123         protected R getResult() {
 124                 return null;
 125         }
 126
 127         @Override
 128         public void close() {
 129                 fcpConnection.removeFcpListener(this);
 130         }
 131
 132         private <M extends BaseMessage> void consume(Consumer<M> consumer, M message) {
 133                 consume(consumer, message, "Identifier");
 134         }
 135
 136         private <M extends BaseMessage> void consume(Consumer<M> consumer, M message,
 137                         String identifier) {
 138                 if (Objects.equals(message.getField(identifier), this.identifier.get())) {
 139                         consumeAlways(consumer, message);
 140                 }
 141         }
 142
 143         private <M extends BaseMessage> void consumeAlways(Consumer<M> consumer, M message) {
 144                 consumer.accept(message);
 145                 notifySyncObject();
 146         }
 147
 148         private void consumeUnknown(FcpMessage fcpMessage) {
 149                 consumeUnknownMessage(fcpMessage);
 150                 notifySyncObject();
 151         }
 152
 153         private void consumeClose(Throwable throwable) {
 154                 connectionFailureReason.set(throwable);
 155                 connectionClosed.set(true);
 156                 notifySyncObject();
 157         }
 158
 159         @Override
 160         public final void receivedNodeHello(FcpConnection fcpConnection, NodeHello nodeHello) {
 161                 consume(this::consumeNodeHello, nodeHello);
 162         }
 163
 164         protected void consumeNodeHello(NodeHello nodeHello) { }
 165
 166         @Override
 167         public final void receivedCloseConnectionDuplicateClientName(FcpConnection fcpConnection,
 168                         CloseConnectionDuplicateClientName closeConnectionDuplicateClientName) {
 169                 connectionFailureReason.set(new IOException("duplicate client name"));
 170                 connectionClosed.set(true);
 171                 notifySyncObject();
 172         }
 173
 174         @Override
 175         public final void receivedSSKKeypair(FcpConnection fcpConnection, SSKKeypair sskKeypair) {
 176                 consume(this::consumeSSKKeypair, sskKeypair);
 177         }
 178
 179         protected void consumeSSKKeypair(SSKKeypair sskKeypair) { }
 180
 181         @Override
 182         public final void receivedPeer(FcpConnection fcpConnection, Peer peer) {
 183                 consume(this::consumePeer, peer);
 184         }
 185
 186         protected void consumePeer(Peer peer) { }
 187
 188         @Override
 189         public final void receivedEndListPeers(FcpConnection fcpConnection, EndListPeers endListPeers) {
 190                 consume(this::consumeEndListPeers, endListPeers);
 191         }
 192
 193         protected void consumeEndListPeers(EndListPeers endListPeers) { }
 194
 195         @Override
 196         public final void receivedPeerNote(FcpConnection fcpConnection, PeerNote peerNote) {
 197                 consume(this::consumePeerNote, peerNote);
 198         }
 199
 200         protected void consumePeerNote(PeerNote peerNote) { }
 201
 202         @Override
 203         public final void receivedEndListPeerNotes(FcpConnection fcpConnection, EndListPeerNotes endListPeerNotes) {
 204                 consume(this::consumeEndListPeerNotes, endListPeerNotes);
 205         }
 206
 207         protected void consumeEndListPeerNotes(EndListPeerNotes endListPeerNotes) { }
 208
 209         @Override
 210         public final void receivedPeerRemoved(FcpConnection fcpConnection, PeerRemoved peerRemoved) {
 211                 consume(this::consumePeerRemoved, peerRemoved);
 212         }
 213
 214         protected void consumePeerRemoved(PeerRemoved peerRemoved) { }
 215
 216         @Override
 217         public final void receivedNodeData(FcpConnection fcpConnection, NodeData nodeData) {
 218                 consume(this::consumeNodeData, nodeData);
 219         }
 220
 221         protected void consumeNodeData(NodeData nodeData) { }
 222
 223         @Override
 224         public final void receivedTestDDAReply(FcpConnection fcpConnection, TestDDAReply testDDAReply) {
 225                 consume(this::consumeTestDDAReply, testDDAReply, "Directory");
 226         }
 227
 228         protected void consumeTestDDAReply(TestDDAReply testDDAReply) { }
 229
 230         @Override
 231         public final void receivedTestDDAComplete(FcpConnection fcpConnection, TestDDAComplete testDDAComplete) {
 232                 consume(this::consumeTestDDAComplete, testDDAComplete, "Directory");
 233         }
 234
 235         protected void consumeTestDDAComplete(TestDDAComplete testDDAComplete) { }
 236
 237         @Override
 238         public final void receivedPersistentGet(FcpConnection fcpConnection, PersistentGet persistentGet) {
 239                 consume(this::consumePersistentGet, persistentGet);
 240         }
 241
 242         protected void consumePersistentGet(PersistentGet persistentGet) { }
 243
 244         @Override
 245         public final void receivedPersistentPut(FcpConnection fcpConnection, PersistentPut persistentPut) {
 246                 consume(this::consumePersistentPut, persistentPut);
 247         }
 248
 249         protected void consumePersistentPut(PersistentPut persistentPut) { }
 250
 251         @Override
 252         public final void receivedEndListPersistentRequests(FcpConnection fcpConnection,
 253                         EndListPersistentRequests endListPersistentRequests) {
 254                 consume(this::consumeEndListPersistentRequests, endListPersistentRequests);
 255         }
 256
 257         protected void consumeEndListPersistentRequests(EndListPersistentRequests endListPersistentRequests) { }
 258
 259         @Override
 260         public final void receivedURIGenerated(FcpConnection fcpConnection, URIGenerated uriGenerated) {
 261                 consume(this::consumeURIGenerated, uriGenerated);
 262         }
 263
 264         protected void consumeURIGenerated(URIGenerated uriGenerated) { }
 265
 266         @Override
 267         public final void receivedDataFound(FcpConnection fcpConnection, DataFound dataFound) {
 268                 consume(this::consumeDataFound, dataFound);
 269         }
 270
 271         protected void consumeDataFound(DataFound dataFound) { }
 272
 273         @Override
 274         public final void receivedAllData(FcpConnection fcpConnection, AllData allData) {
 275                 consume(this::consumeAllData, allData);
 276         }
 277
 278         protected void consumeAllData(AllData allData) { }
 279
 280         @Override
 281         public final void receivedSimpleProgress(FcpConnection fcpConnection, SimpleProgress simpleProgress) {
 282                 consume(this::consumeSimpleProgress, simpleProgress);
 283         }
 284
 285         protected void consumeSimpleProgress(SimpleProgress simpleProgress) { }
 286
 287         @Override
 288         public final void receivedStartedCompression(FcpConnection fcpConnection, StartedCompression startedCompression) {
 289                 consume(this::consumeStartedCompression, startedCompression);
 290         }
 291
 292         protected void consumeStartedCompression(StartedCompression startedCompression) { }
 293
 294         @Override
 295         public final void receivedFinishedCompression(FcpConnection fcpConnection, FinishedCompression finishedCompression) {
 296                 consume(this::consumeFinishedCompression, finishedCompression);
 297         }
 298
 299         protected void consumeFinishedCompression(FinishedCompression finishedCompression) { }
 300
 301         @Override
 302         public final void receivedUnknownPeerNoteType(FcpConnection fcpConnection, UnknownPeerNoteType unknownPeerNoteType) {
 303                 consume(this::consumeUnknownPeerNoteType, unknownPeerNoteType);
 304         }
 305
 306         protected void consumeUnknownPeerNoteType(UnknownPeerNoteType unknownPeerNoteType) { }
 307
 308         @Override
 309         public final void receivedUnknownNodeIdentifier(FcpConnection fcpConnection,
 310                         UnknownNodeIdentifier unknownNodeIdentifier) {
 311                 consume(this::consumeUnknownNodeIdentifier, unknownNodeIdentifier);
 312         }
 313
 314         protected void consumeUnknownNodeIdentifier(UnknownNodeIdentifier unknownNodeIdentifier) { }
 315
 316         @Override
 317         public final void receivedConfigData(FcpConnection fcpConnection, ConfigData configData) {
 318                 consume(this::consumeConfigData, configData);
 319         }
 320
 321         protected void consumeConfigData(ConfigData configData) { }
 322
 323         @Override
 324         public final void receivedGetFailed(FcpConnection fcpConnection, GetFailed getFailed) {
 325                 consume(this::consumeGetFailed, getFailed);
 326         }
 327
 328         protected void consumeGetFailed(GetFailed getFailed) { }
 329
 330         @Override
 331         public final void receivedPutFailed(FcpConnection fcpConnection, PutFailed putFailed) {
 332                 consume(this::consumePutFailed, putFailed);
 333         }
 334
 335         protected void consumePutFailed(PutFailed putFailed) { }
 336
 337         @Override
 338         public final void receivedIdentifierCollision(FcpConnection fcpConnection, IdentifierCollision identifierCollision) {
 339                 consume(this::consumeIdentifierCollision, identifierCollision);
 340         }
 341
 342         protected void consumeIdentifierCollision(IdentifierCollision identifierCollision) { }
 343
 344         @Override
 345         public final void receivedPersistentPutDir(FcpConnection fcpConnection, PersistentPutDir persistentPutDir) {
 346                 consume(this::consumePersistentPutDir, persistentPutDir);
 347         }
 348
 349         protected void consumePersistentPutDir(PersistentPutDir persistentPutDir) { }
 350
 351         @Override
 352         public final void receivedPersistentRequestRemoved(FcpConnection fcpConnection,
 353                         PersistentRequestRemoved persistentRequestRemoved) {
 354                 consume(this::consumePersistentRequestRemoved, persistentRequestRemoved);
 355         }
 356
 357         protected void consumePersistentRequestRemoved(PersistentRequestRemoved persistentRequestRemoved) { }
 358
 359         @Override
 360         public final void receivedSubscribedUSK(FcpConnection fcpConnection, SubscribedUSK subscribedUSK) {
 361                 consume(this::consumeSubscribedUSK, subscribedUSK);
 362         }
 363
 364         protected void consumeSubscribedUSK(SubscribedUSK subscribedUSK) { }
 365
 366         @Override
 367         public final void receivedSubscribedUSKUpdate(FcpConnection fcpConnection, SubscribedUSKUpdate subscribedUSKUpdate) {
 368                 consume(this::consumeSubscribedUSKUpdate, subscribedUSKUpdate);
 369         }
 370
 371         protected void consumeSubscribedUSKUpdate(SubscribedUSKUpdate subscribedUSKUpdate) { }
 372
 373         @Override
 374         public final void receivedPluginInfo(FcpConnection fcpConnection, PluginInfo pluginInfo) {
 375                 consume(this::consumePluginInfo, pluginInfo);
 376         }
 377
 378         protected void consumePluginInfo(PluginInfo pluginInfo) { }
 379
 380         @Override
 381         public final void receivedPluginRemoved(FcpConnection fcpConnection, PluginRemoved pluginRemoved) {
 382                 consume(this::consumePluginRemoved, pluginRemoved);
 383         }
 384
 385         protected void consumePluginRemoved(PluginRemoved pluginRemoved) { }
 386
 387         @Override
 388         public final void receivedFCPPluginReply(FcpConnection fcpConnection, FCPPluginReply fcpPluginReply) {
 389                 consume(this::consumeFCPPluginReply, fcpPluginReply);
 390         }
 391
 392         protected void consumeFCPPluginReply(FCPPluginReply fcpPluginReply) { }
 393
 394         @Override
 395         public final void receivedPersistentRequestModified(FcpConnection fcpConnection,
 396                         PersistentRequestModified persistentRequestModified) {
 397                 consume(this::consumePersistentRequestModified, persistentRequestModified);
 398         }
 399
 400         protected void consumePersistentRequestModified(PersistentRequestModified persistentRequestModified) { }
 401
 402         @Override
 403         public final void receivedPutSuccessful(FcpConnection fcpConnection, PutSuccessful putSuccessful) {
 404                 consume(this::consumePutSuccessful, putSuccessful);
 405         }
 406
 407         protected void consumePutSuccessful(PutSuccessful putSuccessful) { }
 408
 409         @Override
 410         public final void receivedPutFetchable(FcpConnection fcpConnection, PutFetchable putFetchable) {
 411                 consume(this::consumePutFetchable, putFetchable);
 412         }
 413
 414         protected void consumePutFetchable(PutFetchable putFetchable) { }
 415
 416         @Override
 417         public final void receivedSentFeed(FcpConnection source, SentFeed sentFeed) {
 418                 consume(this::consumeSentFeed, sentFeed);
 419         }
 420
 421         protected void consumeSentFeed(SentFeed sentFeed) { }
 422
 423         @Override
 424         public final void receivedBookmarkFeed(FcpConnection fcpConnection, ReceivedBookmarkFeed receivedBookmarkFeed) {
 425                 consume(this::consumeReceivedBookmarkFeed, receivedBookmarkFeed);
 426         }
 427
 428         protected void consumeReceivedBookmarkFeed(ReceivedBookmarkFeed receivedBookmarkFeed) { }
 429
 430         @Override
 431         public final void receivedProtocolError(FcpConnection fcpConnection, ProtocolError protocolError) {
 432                 consume(this::consumeProtocolError, protocolError);
 433         }
 434
 435         protected void consumeProtocolError(ProtocolError protocolError) { }
 436
 437         @Override
 438         public final void receivedMessage(FcpConnection fcpConnection, FcpMessage fcpMessage) {
 439                 consumeUnknown(fcpMessage);
 440         }
 441
 442         protected void consumeUnknownMessage(FcpMessage fcpMessage) { }
 443
 444         @Override
 445         public final void connectionClosed(FcpConnection fcpConnection, Throwable throwable) {
 446                 consumeClose(throwable);
 447         }
 448
 449 }