chore: refactoring and cleanups

signum-network · Jan 10, 2024 · 8340754 · 8340754
1 parent 6d4c97d
commit 8340754
Show file tree

Hide file tree

Showing 4 changed files with 26 additions and 93 deletions.
diff --git a/src/brs/web/api/ws/BlockchainEventNotifier.java b/src/brs/web/api/ws/BlockchainEventNotifier.java
@@ -2,10 +2,9 @@
 
 import brs.*;
 import brs.props.Props;
-import brs.web.api.ws.common.Debouncer2;
+import brs.web.api.ws.common.Debouncer;
 import brs.web.api.ws.common.SimpleScheduler;
 import brs.web.api.ws.emitter.data.ConnectedEventData;
-import brs.web.api.ws.common.Debouncer;
 import brs.web.api.ws.emitter.*;
 import brs.web.server.WebServerContext;
 import org.slf4j.Logger;
@@ -17,16 +16,15 @@
 public class BlockchainEventNotifier {
 
   private final static int SHUTDOWN_TIMEOUT_SECS = 5;
-  private final static int HEARTBEAT_INTERVAL_SECS = 10; // make this configurable
   private final static int BLOCK_PUSHED_DEBOUNCE_SECS = 1;
   private final static int IO_THREAD_COUNT = 10;
   private static BlockchainEventNotifier instance;
   private final ExecutorService notifyExecutor;
-//  private final Debouncer blockPushedDebouncer;
+  //  private final Debouncer blockPushedDebouncer;
   private final Logger logger = LoggerFactory.getLogger(BlockchainEventNotifier.class);
   private final WebServerContext context;
   private final ConcurrentHashMap<String, WebSocketConnection> connections = new ConcurrentHashMap<>();
-  private final Debouncer2 blockPushedDebouncer2;
+  private final Debouncer blockPushedDebouncer = new Debouncer(BLOCK_PUSHED_DEBOUNCE_SECS * 1000);
   private SimpleScheduler heartbeat;
 
   public static BlockchainEventNotifier getInstance(WebServerContext context) {
@@ -42,22 +40,20 @@ private BlockchainEventNotifier(WebServerContext context) {
     this.context.getBlockchainProcessor().addListener(this::onBlockGeneratedEvent, BlockchainProcessor.Event.BLOCK_GENERATED);
     this.context.getBlockchainProcessor().addListener(this::onBlockPushedEvent, BlockchainProcessor.Event.BLOCK_PUSHED);
     this.context.getTransactionProcessor().addListener(this::onPendingTransactionEvent, TransactionProcessor.Event.ADDED_UNCONFIRMED_TRANSACTIONS);
-//    this.blockPushedDebouncer = new Debouncer(BLOCK_PUSHED_DEBOUNCE_SECS * 1000);
-    this.blockPushedDebouncer2 = new Debouncer2();
     initializeHeartBeat();
   }
 
 
   private void withActiveConnectionsOnly(Runnable fn) {
-    if (!this.connections.isEmpty()) {
+    if (!connections.isEmpty()) {
       fn.run();
     }
   }
 
   public void addConnection(WebSocketConnection connection) {
     connections.put(connection.getId(), connection);
 
-    this.notifyExecutor.submit(() -> {
+    notifyExecutor.submit(() -> {
       ConnectedEventData data = new ConnectedEventData();
       data.version = Burst.VERSION.toString();
       data.networkName = context.getPropertyService().getString(Props.NETWORK_NAME);
@@ -87,47 +83,33 @@ private void initializeHeartBeat() {
     );
   }
 
-
   public void removeConnection(WebSocketConnection connection) {
     connections.remove(connection.getId());
   }
 
   public void onBlockGeneratedEvent(Block block) {
-    withActiveConnectionsOnly(() -> {
-      notifyExecutor.submit(() -> {
-        connections.values().forEach(connection -> new BlockGeneratedEventEmitter(connection).emit(block));
-      });
-    });
+    withActiveConnectionsOnly(() -> notifyExecutor.submit(() -> {
+      connections.values().forEach(connection -> new BlockGeneratedEventEmitter(connection).emit(block));
+    }));
   }
 
   public void onBlockPushedEvent(Block block) {
-    withActiveConnectionsOnly(() -> {
-      blockPushedDebouncer2.debounce(
-        BlockchainEventNotifier.class,
-        () -> {
-          notifyExecutor.submit(() -> {
-            int currentHeight = context.getBlockchainProcessor().getLastBlockchainFeederHeight();
-            connections.values().forEach(connection -> new BlockPushedEventEmitter(connection, currentHeight).emit(block));
-          });
-        },
-        BLOCK_PUSHED_DEBOUNCE_SECS, TimeUnit.SECONDS);
-    });
+    withActiveConnectionsOnly(() -> blockPushedDebouncer.debounce(() -> notifyExecutor.submit(() -> {
+      int currentHeight = context.getBlockchainProcessor().getLastBlockchainFeederHeight();
+      connections.values().forEach(connection -> new BlockPushedEventEmitter(connection, currentHeight).emit(block));
+    })));
   }
 
   private void onPendingTransactionEvent(List<? extends Transaction> transactions) {
-    withActiveConnectionsOnly(() -> {
-      notifyExecutor.submit(() -> {
-        connections.values().forEach(connection -> new PendingTransactionsAddedEventEmitter(connection).emit(transactions));
-      });
-    });
+    withActiveConnectionsOnly(() -> notifyExecutor.submit(() -> {
+      connections.values().forEach(connection -> new PendingTransactionsAddedEventEmitter(connection).emit(transactions));
+    }));
   }
 
   private void shutdownNotifyExecutor() {
     if (!notifyExecutor.isTerminated()) {
       logger.info("Closing {} websocket connection(s)...", connections.size());
-      notifyExecutor.submit(() -> {
-        connections.values().forEach(WebSocketConnection::close);
-      });
+      notifyExecutor.submit(() -> connections.values().forEach(WebSocketConnection::close));
       notifyExecutor.shutdown();
       try {
         if (!notifyExecutor.awaitTermination(SHUTDOWN_TIMEOUT_SECS, TimeUnit.SECONDS)) {
@@ -146,7 +128,7 @@ public void shutdown() {
       if (heartbeat != null) {
         heartbeat.shutdown();
       }
-      blockPushedDebouncer2.shutdown(SHUTDOWN_TIMEOUT_SECS);
+      blockPushedDebouncer.shutdown();
       shutdownNotifyExecutor();
     } catch (Exception e) {
       logger.warn("Graceful WebSocket shutdown not successful: {}", e.getMessage());

diff --git a/src/brs/web/api/ws/common/Debouncer.java b/src/brs/web/api/ws/common/Debouncer.java
@@ -1,6 +1,5 @@
 package brs.web.api.ws.common;
 
-import brs.web.api.ws.WebSocketConnection;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -13,7 +12,7 @@ public class Debouncer {
   private static final Logger logger = LoggerFactory.getLogger(Debouncer.class);
 
   private final int delay;
-  private Timer timer;
+  private final Timer timer;
   private final AtomicBoolean isRunning = new AtomicBoolean(false);
 
   public Debouncer(int delayMillies) {

diff --git a/src/brs/web/api/ws/common/Debouncer2.java b/src/brs/web/api/ws/common/Debouncer2.java
diff --git a/src/brs/web/server/ServerConnectorWebsocketBuilderImpl.java b/src/brs/web/server/ServerConnectorWebsocketBuilderImpl.java
@@ -3,14 +3,17 @@
 import brs.props.PropertyService;
 import brs.props.Props;
 import brs.web.api.ws.WebSocketConnectionCreator;
-import org.eclipse.jetty.server.Server;
-import org.eclipse.jetty.server.ServerConnector;
+import org.eclipse.jetty.server.*;
 import org.eclipse.jetty.servlet.ServletContextHandler;
+import org.eclipse.jetty.util.ssl.SslContextFactory;
 import org.eclipse.jetty.websocket.server.config.JettyWebSocketServletContainerInitializer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.time.Duration;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 
 public class ServerConnectorWebsocketBuilderImpl implements ServerConnectorBuilder {
 
@@ -24,8 +27,9 @@ public ServerConnectorWebsocketBuilderImpl(WebServerContext context, ServletCont
   }
   @Override
   public ServerConnector build(Server server) {
-    ServerConnector connector = new ServerConnector(server);
     PropertyService propertyService = context.getPropertyService();
+    // TODO: WSS Support
+    ServerConnector connector =  new ServerConnector(server);
     connector.setHost(propertyService.getString(Props.API_LISTEN));
     connector.setPort(propertyService.getInt(Props.API_WEBSOCKET_PORT));
     connector.setReuseAddress(true);
@@ -37,4 +41,5 @@ public ServerConnector build(Server server) {
     });
     return connector;
   }
+
 }