X-Git-Url: https://git.pterodactylus.net/?a=blobdiff_plain;f=src%2Fmain%2Fjava%2Fnet%2Fpterodactylus%2Ffcp%2Fquelaton%2FDefaultFcpClient.java;h=79c7f44f9d27672d06a3d5829293301f6219d6f9;hb=7980851d84be2f31d0db3837619eb1548c3847c7;hp=603b1508ca70b040b5453e08dfd5fcf9fea3ed55;hpb=4e2d66eee125e3d2136790619c07ab0ad5b25929;p=jFCPlib.git diff --git a/src/main/java/net/pterodactylus/fcp/quelaton/DefaultFcpClient.java b/src/main/java/net/pterodactylus/fcp/quelaton/DefaultFcpClient.java index 603b150..79c7f44 100644 --- a/src/main/java/net/pterodactylus/fcp/quelaton/DefaultFcpClient.java +++ b/src/main/java/net/pterodactylus/fcp/quelaton/DefaultFcpClient.java @@ -3,7 +3,6 @@ package net.pterodactylus.fcp.quelaton; import java.io.IOException; import java.io.InputStream; import java.util.Optional; -import java.util.concurrent.Callable; import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Future; @@ -16,16 +15,11 @@ import net.pterodactylus.fcp.ClientGet; import net.pterodactylus.fcp.ClientHello; import net.pterodactylus.fcp.CloseConnectionDuplicateClientName; import net.pterodactylus.fcp.FcpConnection; -import net.pterodactylus.fcp.FcpKeyPair; import net.pterodactylus.fcp.FcpUtils.TempInputStream; -import net.pterodactylus.fcp.GenerateSSK; import net.pterodactylus.fcp.GetFailed; import net.pterodactylus.fcp.NodeHello; import net.pterodactylus.fcp.Priority; import net.pterodactylus.fcp.ReturnType; -import net.pterodactylus.fcp.SSKKeypair; - -import com.google.common.io.ByteStreams; /** * Default {@link FcpClient} implementation. @@ -50,26 +44,38 @@ public class DefaultFcpClient implements FcpClient { this.expectedVersion = expectedVersion; } - private void connect() throws IOException { - if (fcpConnection.get() != null) { - return; + private FcpConnection connect() throws IOException { + FcpConnection fcpConnection = this.fcpConnection.get(); + if (fcpConnection != null) { + return fcpConnection; } - fcpConnection.compareAndSet(null, createConnection()); + fcpConnection = createConnection(); + this.fcpConnection.compareAndSet(null, fcpConnection); + return fcpConnection; } private FcpConnection createConnection() throws IOException { FcpConnection connection = new FcpConnection(hostname, port); connection.connect(); - AtomicReference receivedNodeHello = new AtomicReference<>(); - AtomicBoolean receivedClosed = new AtomicBoolean(); - FcpReplySequence nodeHelloSequence = new FcpReplySequence(threadPool, connection); - nodeHelloSequence - .handle(NodeHello.class) - .with((nodeHello) -> receivedNodeHello.set(nodeHello)); - nodeHelloSequence - .handle(CloseConnectionDuplicateClientName.class) - .with((closeConnection) -> receivedClosed.set(true)); - nodeHelloSequence.waitFor(() -> receivedNodeHello.get() != null || receivedClosed.get()); + FcpReplySequence nodeHelloSequence = new FcpReplySequence(threadPool, connection) { + private final AtomicReference receivedNodeHello = new AtomicReference<>(); + private final AtomicBoolean receivedClosed = new AtomicBoolean(); + @Override + protected boolean isFinished() { + return receivedNodeHello.get() != null || receivedClosed.get(); + } + + @Override + protected void consumeNodeHello(NodeHello nodeHello) { + receivedNodeHello.set(nodeHello); + } + + @Override + protected void consumeCloseConnectionDuplicateClientName( + CloseConnectionDuplicateClientName closeConnectionDuplicateClientName) { + receivedClosed.set(true); + } + }; ClientHello clientHello = new ClientHello(clientName.get(), expectedVersion.get()); try { nodeHelloSequence.send(clientHello).get(); @@ -82,42 +88,7 @@ public class DefaultFcpClient implements FcpClient { @Override public GenerateKeypairCommand generateKeypair() { - return new GenerateKeypairCommandImpl(); - } - - private class GenerateKeypairCommandImpl implements GenerateKeypairCommand { - - @Override - public Future execute() { - return threadPool.submit(() -> { - connect(); - Sequence sequence = new Sequence(); - FcpReplySequence replySequence = new FcpReplySequence(threadPool, fcpConnection.get()); - replySequence.handle(SSKKeypair.class).with(sequence::handleSSKKeypair); - replySequence.waitFor(sequence::isFinished); - replySequence.send(new GenerateSSK()).get(); - return sequence.getKeyPair(); - }); - } - - private class Sequence { - - private AtomicReference keyPair = new AtomicReference<>(); - - public void handleSSKKeypair(SSKKeypair sskKeypair) { - keyPair.set(new FcpKeyPair(sskKeypair.getRequestURI(), sskKeypair.getInsertURI())); - } - - public boolean isFinished() { - return keyPair.get() != null; - } - - public FcpKeyPair getKeyPair() { - return keyPair.get(); - } - - } - + return new GenerateKeypairCommandImpl(threadPool, this::connect); } @Override @@ -179,11 +150,6 @@ public class DefaultFcpClient implements FcpClient { @Override public Future> uri(String uri) { - return threadPool.submit(() -> execute(uri)); - } - - private Optional execute(String uri) throws IOException, ExecutionException, InterruptedException { - DefaultFcpClient.this.connect(); ClientGet clientGet = new ClientGet(uri, identifier, ReturnType.direct); if (ignoreDataStore) { clientGet.setIgnoreDataStore(true); @@ -203,91 +169,73 @@ public class DefaultFcpClient implements FcpClient { if (global) { clientGet.setGlobal(true); } - try (FcpReplySequence replySequence = new FcpReplySequence(threadPool, fcpConnection.get())) { - Sequence sequence = new Sequence(identifier); - replySequence.handle(AllData.class).with(sequence::allData); - replySequence.handle(GetFailed.class).with(sequence::getFailed); - replySequence.handleClose().with(sequence::disconnect); - replySequence.waitFor(sequence::isFinished); - replySequence.send(clientGet).get(); - return sequence.isSuccessful() ? Optional.of(sequence.getData()) : Optional.empty(); - } - } - - private class Sequence { - - private final AtomicBoolean finished = new AtomicBoolean(); - private final AtomicBoolean failed = new AtomicBoolean(); - - private final String identifier; - - private String contentType; - private long dataLength; - private InputStream payload; - - private Sequence(String identifier) { - this.identifier = identifier; - } + return threadPool.submit(() -> { + FcpReplySequence> replySequence = new FcpReplySequence>(threadPool, connect()) { + private final AtomicBoolean finished = new AtomicBoolean(); + private final AtomicBoolean failed = new AtomicBoolean(); - public boolean isFinished() { - return finished.get() || failed.get(); - } + private final String identifier = ClientGetCommandImpl.this.identifier; - public boolean isSuccessful() { - return !failed.get(); - } + private String contentType; + private long dataLength; + private InputStream payload; - public Data getData() { - return new Data() { @Override - public String getMimeType() { - synchronized (Sequence.this) { - return contentType; - } + protected boolean isFinished() { + return finished.get() || failed.get(); } @Override - public long size() { - synchronized (Sequence.this) { - return dataLength; - } + protected Optional getResult() { + return failed.get() ? Optional.empty() : Optional.of(new Data() { + @Override + public String getMimeType() { + return contentType; + } + + @Override + public long size() { + return dataLength; + } + + @Override + public InputStream getInputStream() { + return payload; + } + }); } @Override - public InputStream getInputStream() { - synchronized (Sequence.this) { - return payload; + protected void consumeAllData(AllData allData) { + if (allData.getIdentifier().equals(identifier)) { + synchronized (this) { + contentType = allData.getContentType(); + dataLength = allData.getDataLength(); + try { + payload = new TempInputStream(allData.getPayloadInputStream(), dataLength); + finished.set(true); + } catch (IOException e) { + // TODO – logging + failed.set(true); + } + } } } - }; - } - public void allData(AllData allData) { - if (allData.getIdentifier().equals(identifier)) { - synchronized (this) { - contentType = allData.getContentType(); - dataLength = allData.getDataLength(); - try { - payload = new TempInputStream(allData.getPayloadInputStream(), dataLength); - finished.set(true); - } catch (IOException e) { - // TODO – logging + @Override + protected void consumeGetFailed(GetFailed getFailed) { + if (getFailed.getIdentifier().equals(identifier)) { failed.set(true); } } - } - } - - public void getFailed(GetFailed getFailed) { - if (getFailed.getIdentifier().equals(identifier)) { - failed.set(true); - } - } - - public void disconnect(Throwable t) { - failed.set(true); - } + @Override + protected void consumeConnectionClosed(Throwable throwable) { + failed.set(true); + } + }; + return replySequence.send(clientGet).get(); + }); } }