Close FCP reply sequences after use
[jFCPlib.git] / src / main / java / net / pterodactylus / fcp / quelaton / ListPeersCommandImpl.java
1 package net.pterodactylus.fcp.quelaton;
2
3 import java.io.IOException;
4 import java.util.Collection;
5 import java.util.HashSet;
6 import java.util.concurrent.ExecutionException;
7 import java.util.concurrent.ExecutorService;
8 import java.util.concurrent.atomic.AtomicBoolean;
9
10 import net.pterodactylus.fcp.EndListPeers;
11 import net.pterodactylus.fcp.ListPeers;
12 import net.pterodactylus.fcp.Peer;
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 ListPeersCommand} implementation based on {@link FcpReplySequence}.
20  *
21  * @author <a href="mailto:bombe@pterodactylus.net">David ‘Bombe’ Roden</a>
22  */
23 public class ListPeersCommandImpl implements ListPeersCommand {
24
25         private final ListeningExecutorService threadPool;
26         private final ConnectionSupplier connectionSupplier;
27         private final AtomicBoolean includeMetadata = new AtomicBoolean(false);
28         private final AtomicBoolean includeVolatile = new AtomicBoolean(false);
29
30         public ListPeersCommandImpl(ExecutorService threadPool, ConnectionSupplier connectionSupplier) {
31                 this.threadPool = MoreExecutors.listeningDecorator(threadPool);
32                 this.connectionSupplier = connectionSupplier;
33         }
34
35         @Override
36         public ListPeersCommand includeMetadata() {
37                 includeMetadata.set(true);
38                 return this;
39         }
40
41         @Override
42         public ListPeersCommand includeVolatile() {
43                 includeVolatile.set(true);
44                 return this;
45         }
46
47         @Override
48         public ListenableFuture<Collection<Peer>> execute() {
49                 return threadPool.submit(this::executeSequence);
50         }
51
52         private Collection<Peer> executeSequence() throws InterruptedException, ExecutionException, IOException {
53                 String identifier = new RandomIdentifierGenerator().generate();
54                 ListPeers listPeers = new ListPeers(identifier, includeMetadata.get(), includeVolatile.get());
55                 try (ListPeersReplySequence listPeersReplySequence = new ListPeersReplySequence()) {
56                         return listPeersReplySequence.send(listPeers).get();
57                 }
58         }
59
60         private class ListPeersReplySequence extends FcpReplySequence<Collection<Peer>> {
61
62                 private final Collection<Peer> peers = new HashSet<>();
63                 private final AtomicBoolean finished = new AtomicBoolean(false);
64
65                 public ListPeersReplySequence() throws IOException {
66                         super(threadPool, connectionSupplier.get());
67                 }
68
69                 @Override
70                 protected boolean isFinished() {
71                         return finished.get();
72                 }
73
74                 @Override
75                 protected Collection<Peer> getResult() {
76                         return peers;
77                 }
78
79                 @Override
80                 protected void consumePeer(Peer peer) {
81                         peers.add(peer);
82                 }
83
84                 @Override
85                 protected void consumeEndListPeers(EndListPeers endListPeers) {
86                         finished.set(true);
87                 }
88
89         }
90
91 }