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