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