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