mirrors/bisq
1
0
Fork 0
mirror of https://github.com/bisq-network/bisq.git synced 2024-11-19 09:52:23 +01:00
Code Issues Packages Projects Releases Wiki Activity

NetworkNode: Move server threading logic to server

Browse Source
This commit is contained in:
Alva Swanson 2023-02-11 20:40:55 +01:00
parent b1e2cb03ad
commit ee2157aa24
No known key found for this signature in database
GPG Key ID: 004760E77F753090
2 changed files with 14 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
p2p/src/main/java/bisq/network/p2p/network/NetworkNode.java
View File

@ -24,7 +24,6 @@ import bisq.common.UserThread;
import bisq.common.app.Capabilities; import bisq.common.app.Capabilities;
import bisq.common.proto.network.NetworkEnvelope; import bisq.common.proto.network.NetworkEnvelope;
import bisq.common.proto.network.NetworkProtoResolver; import bisq.common.proto.network.NetworkProtoResolver;
import bisq.common.util.SingleThreadExecutorUtils;
import bisq.common.util.Utilities; import bisq.common.util.Utilities;
import com.runjva.sourceforge.jsocks.protocol.Socks5Proxy; import com.runjva.sourceforge.jsocks.protocol.Socks5Proxy;
@ -50,7 +49,6 @@ import java.util.HashSet;
import java.util.Optional; import java.util.Optional;
import java.util.Set; import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet; import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException; import java.util.concurrent.TimeoutException;
@ -81,7 +79,6 @@ public abstract class NetworkNode implements MessageListener {
final CopyOnWriteArraySet<SetupListener> setupListeners = new CopyOnWriteArraySet<>(); final CopyOnWriteArraySet<SetupListener> setupListeners = new CopyOnWriteArraySet<>();
private final ListeningExecutorService connectionExecutor; private final ListeningExecutorService connectionExecutor;
private final ListeningExecutorService sendMessageExecutor; private final ListeningExecutorService sendMessageExecutor;
private final ExecutorService serverExecutor;
private Server server; private Server server;
private volatile boolean shutDownInProgress; private volatile boolean shutDownInProgress;
@ -112,7 +109,6 @@ public abstract class NetworkNode implements MessageListener {
maxConnections * 3, maxConnections * 3,
30, 30,
30); 30);
serverExecutor = SingleThreadExecutorUtils.getSingleThreadExecutor("NetworkNode.server-" + servicePort);
} }
/////////////////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////////////////
@ -373,7 +369,6 @@ public abstract class NetworkNode implements MessageListener {
if (server != null) { if (server != null) {
server.shutDown(); server.shutDown();
server = null; server = null;
serverExecutor.shutdownNow();
} }
Set<Connection> allConnections = getAllConnections(); Set<Connection> allConnections = getAllConnections();
@ -503,7 +498,7 @@ public abstract class NetworkNode implements MessageListener {
connectionListener, connectionListener,
networkProtoResolver, networkProtoResolver,
networkFilter); networkFilter);
serverExecutor.submit(server); server.start();
} }
private Optional<OutboundConnection> lookupOutBoundConnection(NodeAddress peersNodeAddress) { private Optional<OutboundConnection> lookupOutBoundConnection(NodeAddress peersNodeAddress) {

17
p2p/src/main/java/bisq/network/p2p/network/Server.java
View File

@ -42,9 +42,11 @@ class Server implements Runnable {
private final NetworkFilter networkFilter; private final NetworkFilter networkFilter;
private final ServerSocket serverSocket; private final ServerSocket serverSocket;
private final int localPort;
private final Set<Connection> connections = new CopyOnWriteArraySet<>(); private final Set<Connection> connections = new CopyOnWriteArraySet<>();
private volatile boolean stopped; private volatile boolean stopped;
private final NetworkProtoResolver networkProtoResolver; private final NetworkProtoResolver networkProtoResolver;
private final Thread serverThread = new Thread(this);
public Server(ServerSocket serverSocket, public Server(ServerSocket serverSocket,
@ -54,19 +56,23 @@ class Server implements Runnable {
@Nullable NetworkFilter networkFilter) { @Nullable NetworkFilter networkFilter) {
this.networkProtoResolver = networkProtoResolver; this.networkProtoResolver = networkProtoResolver;
this.serverSocket = serverSocket; this.serverSocket = serverSocket;
this.localPort = serverSocket.getLocalPort();
this.messageListener = messageListener; this.messageListener = messageListener;
this.connectionListener = connectionListener; this.connectionListener = connectionListener;
this.networkFilter = networkFilter; this.networkFilter = networkFilter;
} }
public void start() {
serverThread.setName("Server-" + localPort);
serverThread.start();
}
@Override @Override
public void run() { public void run() {
try { try {
// Thread created by NetworkNode
Thread.currentThread().setName("Server-" + serverSocket.getLocalPort());
try { try {
while (!stopped && !Thread.currentThread().isInterrupted()) { while (!stopped && !Thread.currentThread().isInterrupted()) {
log.debug("Ready to accept new clients on port " + serverSocket.getLocalPort()); log.debug("Ready to accept new clients on port " + localPort);
final Socket socket = serverSocket.accept(); final Socket socket = serverSocket.accept();
if (!stopped && !Thread.currentThread().isInterrupted()) { if (!stopped && !Thread.currentThread().isInterrupted()) {
log.debug("Accepted new client on localPort/port " + socket.getLocalPort() + "/" + socket.getPort()); log.debug("Accepted new client on localPort/port " + socket.getLocalPort() + "/" + socket.getPort());
@ -106,8 +112,11 @@ class Server implements Runnable {
connections.forEach(connection -> connection.shutDown(CloseConnectionReason.APP_SHUT_DOWN)); connections.forEach(connection -> connection.shutDown(CloseConnectionReason.APP_SHUT_DOWN));
try { try {
if (!serverSocket.isClosed()) if (!serverSocket.isClosed()) {
serverSocket.close(); serverSocket.close();
}
serverThread.interrupt();
} catch (SocketException e) { } catch (SocketException e) {
log.debug("SocketException at shutdown might be expected " + e.getMessage()); log.debug("SocketException at shutdown might be expected " + e.getMessage());
} catch (IOException e) { } catch (IOException e) {