import org.ldk.structs.*;
import java.io.IOException;
+import java.util.LinkedList;
import java.net.SocketAddress;
import java.net.StandardSocketOptions;
+import java.nio.Buffer;
import java.nio.ByteBuffer;
import java.nio.channels.*;
public class NioPeerHandler {
private static class Peer {
SocketDescriptor descriptor;
- // When we are told by LDK to disconnect, we can't return to LDK until we are sure
- // won't call any more read/write PeerManager functions with the same connection.
- // This is set to true if we're in such a condition (with disconnect checked
- // before with the Peer monitor lock held) and false when we can return.
- boolean block_disconnect_socket = false;
- // Indicates LDK told us to disconnect this peer, and thus we should not call socket_disconnected.
- boolean disconnect_requested = false;
SelectionKey key;
}
@Override
public void disconnect_socket() {
- synchronized (peer) {
- peer.disconnect_requested = true;
- }
try {
do_selector_action(() -> {
peer.key.cancel();
peer.key.channel().close();
});
} catch (IOException ignored) { }
- synchronized (peer) {
- while (peer.block_disconnect_socket) {
- try {
- peer.wait();
- } catch (InterruptedException ignored) { }
- }
- }
}
@Override public boolean eq(SocketDescriptor other_arg) { return other_arg.hash() == our_id; }
@Override public long hash() { return our_id; }
/**
* Constructs a new peer handler, spawning a thread to monitor for socket events.
- * The background thread will call the PeerManager's timer_tick_occured() function for you on an appropriate schedule.
*
* @param manager The LDK PeerManager which connection data will be provided to.
* @throws IOException If an internal java.nio error occurs.
this.selector = Selector.open();
io_thread = new Thread(() -> {
ByteBuffer buf = ByteBuffer.allocate(8192);
- long lastTimerTick = System.currentTimeMillis();
while (true) {
try {
if (IS_ANDROID) {
continue; // There is no attachment so the rest of the loop is useless
}
Peer peer = (Peer) key.attachment();
- synchronized (peer) {
- if (peer.disconnect_requested)
- continue;
- peer.block_disconnect_socket = true;
- }
try {
if (key.isValid() && (key.interestOps() & SelectionKey.OP_WRITE) != 0 && key.isWritable()) {
Result_NonePeerHandleErrorZ res = this.peer_manager.write_buffer_space_avail(peer.descriptor);
}
}
if (key.isValid() && (key.interestOps() & SelectionKey.OP_READ) != 0 && key.isReadable()) {
- buf.clear();
+ ((Buffer)buf).clear();
int read = ((SocketChannel) key.channel()).read(buf);
if (read == -1) {
this.peer_manager.socket_disconnected(peer.descriptor);
key.cancel();
} else if (read > 0) {
- buf.flip();
+ ((Buffer)buf).flip();
byte[] read_bytes = new byte[read];
buf.get(read_bytes, 0, read);
Result_boolPeerHandleErrorZ res = this.peer_manager.read_event(peer.descriptor, read_bytes);
key.cancel();
peer_manager.socket_disconnected(peer.descriptor);
}
- synchronized (peer) {
- peer.block_disconnect_socket = false;
- peer.notifyAll();
- }
} catch (CancelledKeyException e) {
try { key.channel().close(); } catch (IOException ignored) { }
// The key is only cancelled when we have notified the PeerManager that the socket is closed, so
// no need to do anything here with the PeerManager.
}
}
- if (lastTimerTick < System.currentTimeMillis() - 30 * 1000) {
- peer_manager.timer_tick_occurred();
- lastTimerTick = System.currentTimeMillis();
- }
peer_manager.process_events();
}
}, "NioPeerHandler NIO Thread");
}
}
+ /**
+ * Disconnects any connections currently open with the peer with the given node id.
+ *
+ * @param their_node_id must be a valid 33-byte public key
+ */
+ public void disconnect(byte[] their_node_id) {
+ this.peer_manager.disconnect_by_node_id(their_node_id, false);
+ }
+
+ /**
+ * Before shutdown, we have to ensure all of our listening sockets are closed manually, as they appear
+ * to otherwise remain open and lying around on OSX (though no other platform).
+ */
+ private LinkedList<ServerSocketChannel> listening_sockets = new LinkedList();
/**
* Binds a listening socket to the given address, accepting incoming connections and handling them on the background
* thread.
*/
public void bind_listener(SocketAddress socket_address) throws IOException {
ServerSocketChannel listen_channel = ServerSocketChannel.open();
+ listen_channel.setOption(StandardSocketOptions.SO_REUSEADDR, true);
listen_channel.bind(socket_address);
listen_channel.configureBlocking(false);
do_selector_action(() -> listen_channel.register(this.selector, SelectionKey.OP_ACCEPT));
+ synchronized(listening_sockets) {
+ listening_sockets.add(listen_channel);
+ }
}
/**
try {
io_thread.join();
} catch (InterruptedException ignored) { }
+ synchronized(listening_sockets) {
+ try {
+ selector.close();
+ for (ServerSocketChannel chan : listening_sockets) {
+ chan.close();
+ }
+ } catch (IOException ignored) {}
+ }
}
/**
public void check_events() {
selector.wakeup();
}
-}
\ No newline at end of file
+}
projects / ldk-java / blobdiff