6edce63cadb471aee6745a3edef2f415e7663ff6
[jFCPlib.git] / src / main / java / net / pterodactylus / fcp / quelaton / DefaultFcpClient.java
1 package net.pterodactylus.fcp.quelaton;
2
3 import java.io.IOException;
4 import java.io.InputStream;
5 import java.util.Optional;
6 import java.util.concurrent.ExecutionException;
7 import java.util.concurrent.ExecutorService;
8 import java.util.concurrent.Future;
9 import java.util.concurrent.atomic.AtomicBoolean;
10 import java.util.concurrent.atomic.AtomicReference;
11 import java.util.function.Supplier;
12
13 import net.pterodactylus.fcp.AllData;
14 import net.pterodactylus.fcp.ClientGet;
15 import net.pterodactylus.fcp.ClientHello;
16 import net.pterodactylus.fcp.CloseConnectionDuplicateClientName;
17 import net.pterodactylus.fcp.FcpConnection;
18 import net.pterodactylus.fcp.FcpKeyPair;
19 import net.pterodactylus.fcp.FcpUtils.TempInputStream;
20 import net.pterodactylus.fcp.GenerateSSK;
21 import net.pterodactylus.fcp.GetFailed;
22 import net.pterodactylus.fcp.NodeHello;
23 import net.pterodactylus.fcp.Priority;
24 import net.pterodactylus.fcp.ReturnType;
25 import net.pterodactylus.fcp.SSKKeypair;
26
27 /**
28  * Default {@link FcpClient} implementation.
29  *
30  * @author <a href="bombe@freenetproject.org">David ‘Bombe’ Roden</a>
31  */
32 public class DefaultFcpClient implements FcpClient {
33
34         private final ExecutorService threadPool;
35         private final String hostname;
36         private final int port;
37         private final AtomicReference<FcpConnection> fcpConnection = new AtomicReference<>();
38         private final Supplier<String> clientName;
39         private final Supplier<String> expectedVersion;
40
41         public DefaultFcpClient(ExecutorService threadPool, String hostname, int port, Supplier<String> clientName,
42                         Supplier<String> expectedVersion) {
43                 this.threadPool = threadPool;
44                 this.hostname = hostname;
45                 this.port = port;
46                 this.clientName = clientName;
47                 this.expectedVersion = expectedVersion;
48         }
49
50         private FcpConnection connect() throws IOException {
51                 FcpConnection fcpConnection = this.fcpConnection.get();
52                 if (fcpConnection != null) {
53                         return fcpConnection;
54                 }
55                 fcpConnection = createConnection();
56                 this.fcpConnection.compareAndSet(null, fcpConnection);
57                 return fcpConnection;
58         }
59
60         private FcpConnection createConnection() throws IOException {
61                 FcpConnection connection = new FcpConnection(hostname, port);
62                 connection.connect();
63                 FcpReplySequence<?> nodeHelloSequence = new FcpReplySequence<Void>(threadPool, connection) {
64                         private final AtomicReference<NodeHello> receivedNodeHello = new AtomicReference<>();
65                         private final AtomicBoolean receivedClosed = new AtomicBoolean();
66                         @Override
67                         protected boolean isFinished() {
68                                 return receivedNodeHello.get() != null || receivedClosed.get();
69                         }
70
71                         @Override
72                         protected void consumeNodeHello(NodeHello nodeHello) {
73                                 receivedNodeHello.set(nodeHello);
74                         }
75
76                         @Override
77                         protected void consumeCloseConnectionDuplicateClientName(
78                                 CloseConnectionDuplicateClientName closeConnectionDuplicateClientName) {
79                                 receivedClosed.set(true);
80                         }
81                 };
82                 ClientHello clientHello = new ClientHello(clientName.get(), expectedVersion.get());
83                 try {
84                         nodeHelloSequence.send(clientHello).get();
85                 } catch (InterruptedException | ExecutionException e) {
86                         connection.close();
87                         throw new IOException(String.format("Could not connect to %s:%d.", hostname, port), e);
88                 }
89                 return connection;
90         }
91
92         @Override
93         public GenerateKeypairCommand generateKeypair() {
94                 return new GenerateKeypairCommandImpl();
95         }
96
97         private class GenerateKeypairCommandImpl implements GenerateKeypairCommand {
98
99                 @Override
100                 public Future<FcpKeyPair> execute() {
101                         return threadPool.submit(() -> {
102                                 connect();
103                                 return new FcpReplySequence<FcpKeyPair>(threadPool, connect()) {
104                                         private AtomicReference<FcpKeyPair> keyPair = new AtomicReference<>();
105
106                                         @Override
107                                         protected boolean isFinished() {
108                                                 return keyPair.get() != null;
109                                         }
110
111                                         @Override
112                                         protected FcpKeyPair getResult() {
113                                                 return keyPair.get();
114                                         }
115
116                                         @Override
117                                         protected void consumeSSKKeypair(SSKKeypair sskKeypair) {
118                                                 keyPair.set(new FcpKeyPair(sskKeypair.getRequestURI(), sskKeypair.getInsertURI()));
119                                         }
120                                 }.send(new GenerateSSK()).get();
121                         });
122                 }
123
124         }
125
126         @Override
127         public ClientGetCommand clientGet() {
128                 return new ClientGetCommandImpl();
129         }
130
131         private class ClientGetCommandImpl implements ClientGetCommand {
132
133                 private String identifier;
134                 private boolean ignoreDataStore;
135                 private boolean dataStoreOnly;
136                 private Long maxSize;
137                 private Priority priority;
138                 private boolean realTime;
139                 private boolean global;
140
141                 @Override
142                 public ClientGetCommand identifier(String identifier) {
143                         this.identifier = identifier;
144                         return this;
145                 }
146
147                 @Override
148                 public ClientGetCommand ignoreDataStore() {
149                         ignoreDataStore = true;
150                         return this;
151                 }
152
153                 @Override
154                 public ClientGetCommand dataStoreOnly() {
155                         dataStoreOnly = true;
156                         return this;
157                 }
158
159                 @Override
160                 public ClientGetCommand maxSize(long maxSize) {
161                         this.maxSize = maxSize;
162                         return this;
163                 }
164
165                 @Override
166                 public ClientGetCommand priority(Priority priority) {
167                         this.priority = priority;
168                         return this;
169                 }
170
171                 @Override
172                 public ClientGetCommand realTime() {
173                         realTime = true;
174                         return this;
175                 }
176
177                 @Override
178                 public ClientGetCommand global() {
179                         global = true;
180                         return this;
181                 }
182
183                 @Override
184                 public Future<Optional<Data>> uri(String uri) {
185                         ClientGet clientGet = new ClientGet(uri, identifier, ReturnType.direct);
186                         if (ignoreDataStore) {
187                                 clientGet.setIgnoreDataStore(true);
188                         }
189                         if (dataStoreOnly) {
190                                 clientGet.setDataStoreOnly(true);
191                         }
192                         if (maxSize != null) {
193                                 clientGet.setMaxSize(maxSize);
194                         }
195                         if (priority != null) {
196                                 clientGet.setPriority(priority);
197                         }
198                         if (realTime) {
199                                 clientGet.setRealTimeFlag(true);
200                         }
201                         if (global) {
202                                 clientGet.setGlobal(true);
203                         }
204                         return threadPool.submit(() -> {
205                                 FcpReplySequence<Optional<Data>> replySequence = new FcpReplySequence<Optional<Data>>(threadPool, connect()) {
206                                         private final AtomicBoolean finished = new AtomicBoolean();
207                                         private final AtomicBoolean failed = new AtomicBoolean();
208
209                                         private final String identifier = ClientGetCommandImpl.this.identifier;
210
211                                         private String contentType;
212                                         private long dataLength;
213                                         private InputStream payload;
214
215                                         @Override
216                                         protected boolean isFinished() {
217                                                 return finished.get() || failed.get();
218                                         }
219
220                                         @Override
221                                         protected Optional<Data> getResult() {
222                                                 return failed.get() ? Optional.empty() : Optional.of(new Data() {
223                                                         @Override
224                                                         public String getMimeType() {
225                                                                 return contentType;
226                                                         }
227
228                                                         @Override
229                                                         public long size() {
230                                                                 return dataLength;
231                                                         }
232
233                                                         @Override
234                                                         public InputStream getInputStream() {
235                                                                 return payload;
236                                                         }
237                                                 });
238                                         }
239
240                                         @Override
241                                         protected void consumeAllData(AllData allData) {
242                                                 if (allData.getIdentifier().equals(identifier)) {
243                                                         synchronized (this) {
244                                                                 contentType = allData.getContentType();
245                                                                 dataLength = allData.getDataLength();
246                                                                 try {
247                                                                         payload = new TempInputStream(allData.getPayloadInputStream(), dataLength);
248                                                                         finished.set(true);
249                                                                 } catch (IOException e) {
250                                                                         // TODO – logging
251                                                                         failed.set(true);
252                                                                 }
253                                                         }
254                                                 }
255                                         }
256
257                                         @Override
258                                         protected void consumeGetFailed(GetFailed getFailed) {
259                                                 if (getFailed.getIdentifier().equals(identifier)) {
260                                                         failed.set(true);
261                                                 }
262                                         }
263
264                                         @Override
265                                         protected void consumeConnectionClosed(Throwable throwable) {
266                                                 failed.set(true);
267                                         }
268                                 };
269                                 return replySequence.send(clientGet).get();
270                         });
271                 }
272
273         }
274
275 }
276