import java.io.IOException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
-import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Supplier;
-import net.pterodactylus.fcp.ClientHello;
-import net.pterodactylus.fcp.CloseConnectionDuplicateClientName;
import net.pterodactylus.fcp.FcpConnection;
-import net.pterodactylus.fcp.NodeHello;
import com.google.common.util.concurrent.ListeningExecutorService;
import com.google.common.util.concurrent.MoreExecutors;
private FcpConnection connect() throws IOException {
FcpConnection fcpConnection = this.fcpConnection.get();
- if (fcpConnection != null) {
+ if ((fcpConnection != null) && !fcpConnection.isClosed()) {
return fcpConnection;
}
fcpConnection = createConnection();
- this.fcpConnection.compareAndSet(null, fcpConnection);
+ this.fcpConnection.set(fcpConnection);
return fcpConnection;
}
private FcpConnection createConnection() throws IOException {
- FcpConnection connection = new FcpConnection(hostname, port);
- connection.connect();
- FcpReplySequence<?> nodeHelloSequence = new ClientHelloReplySequence(connection);
- ClientHello clientHello = new ClientHello(clientName.get(), "2.0");
try {
- nodeHelloSequence.send(clientHello).get();
+ return new ClientHelloImpl(threadPool, hostname, port).withName(clientName.get()).execute().get();
} catch (InterruptedException | ExecutionException e) {
- connection.close();
- throw new IOException(String.format("Could not connect to %s:%d.", hostname, port), e);
+ throw new IOException(e);
}
- return connection;
+ }
+
+ @Override
+ public GetNodeCommand getNode() {
+ return new GetNodeCommandImpl(threadPool, this::connect);
}
@Override
}
@Override
+ public ListPeerCommand listPeer() {
+ return new ListPeerCommandImpl(threadPool, this::connect);
+ }
+
+ @Override
public ListPeersCommand listPeers() {
return new ListPeersCommandImpl(threadPool, this::connect);
}
- private class ClientHelloReplySequence extends FcpReplySequence<Void> {
-
- private final AtomicReference<NodeHello> receivedNodeHello;
- private final AtomicBoolean receivedClosed;
-
- public ClientHelloReplySequence(FcpConnection connection) {
- super(DefaultFcpClient.this.threadPool, connection);
- receivedNodeHello = new AtomicReference<>();
- 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);
- }
+ @Override
+ public AddPeerCommand addPeer() {
+ return new AddPeerCommandImpl(threadPool, this::connect);
+ }
+ @Override
+ public ListPeerNotesCommand listPeerNotes() {
+ return new ListPeerNotesCommandImpl(threadPool, this::connect);
}
}