NC-1361 Added configurable refresh delay for SyncingSubscriptionService on start up

acceptance-tests/src/test/java/tech/pegasys/pantheon/tests/acceptance/dsl/node/PantheonNodeFactory.java

@@ -61,12 +61,30 @@ public PantheonNode createMinerNode(final String name) throws IOException {
             name, createMiningParameters(true), createJsonRpcConfig(), createWebSocketConfig()));
   }
 
+  public PantheonNode createMinerNodeWithCustomRefreshDelay(
+      final String name, final Long refreshDelay) throws IOException {
+    WebSocketConfiguration wsConfig = createWebSocketConfig();
+    wsConfig.setRefreshDelay(refreshDelay);
+    return create(
+        new PantheonFactoryConfiguration(
+            name, createMiningParameters(true), createJsonRpcConfig(), wsConfig));
+  }
+
   public PantheonNode createArchiveNode(final String name) throws IOException {
     return create(
         new PantheonFactoryConfiguration(
             name, createMiningParameters(false), createJsonRpcConfig(), createWebSocketConfig()));
   }
 
+  public PantheonNode createArchiveNodeWithCustomRefreshDelay(
+      final String name, final Long refreshDelay) throws IOException {
+    WebSocketConfiguration wsConfig = createWebSocketConfig();
+    wsConfig.setRefreshDelay(refreshDelay);
+    return create(
+        new PantheonFactoryConfiguration(
+            name, createMiningParameters(false), createJsonRpcConfig(), wsConfig));
+  }
+
   public PantheonNode createArchiveNodeWithRpcDisabled(final String name) throws IOException {
     return create(
         new PantheonFactoryConfiguration(

ethereum/jsonrpc/src/main/java/tech/pegasys/pantheon/ethereum/jsonrpc/websocket/WebSocketConfiguration.java

@@ -27,18 +27,21 @@ public class WebSocketConfiguration {
   public static final int DEFAULT_WEBSOCKET_PORT = 8546;
   public static final Collection<RpcApi> DEFAULT_WEBSOCKET_APIS =
       Arrays.asList(RpcApis.ETH, RpcApis.NET, RpcApis.WEB3);
+  public static final long DEFAULT_WEBSOCKET_REFRESH_DELAY = 5000;
 
   private boolean enabled;
   private int port;
   private String host;
   private Collection<RpcApi> rpcApis;
+  private long refreshDelay;
 
   public static WebSocketConfiguration createDefault() {
     final WebSocketConfiguration config = new WebSocketConfiguration();
     config.setEnabled(false);
     config.setHost(DEFAULT_WEBSOCKET_HOST);
     config.setPort(DEFAULT_WEBSOCKET_PORT);
     config.setRpcApis(DEFAULT_WEBSOCKET_APIS);
+    config.setRefreshDelay(DEFAULT_WEBSOCKET_REFRESH_DELAY);
     return config;
   }
 
@@ -110,4 +113,12 @@ public boolean equals(final Object o) {
   public int hashCode() {
     return Objects.hashCode(enabled, port, host, rpcApis);
   }
+
+  public void setRefreshDelay(final long refreshDelay) {
+    this.refreshDelay = refreshDelay;
+  }
+
+  public long getRefreshDelay() {
+    return refreshDelay;
+  }
 }

ethereum/jsonrpc/src/main/java/tech/pegasys/pantheon/ethereum/jsonrpc/websocket/subscription/SubscriptionManager.java

@@ -13,6 +13,7 @@
 package tech.pegasys.pantheon.ethereum.jsonrpc.websocket.subscription;
 
 import tech.pegasys.pantheon.ethereum.jsonrpc.internal.results.JsonRpcResult;
+import tech.pegasys.pantheon.ethereum.jsonrpc.websocket.WebSocketConfiguration;
 import tech.pegasys.pantheon.ethereum.jsonrpc.websocket.subscription.request.SubscribeRequest;
 import tech.pegasys.pantheon.ethereum.jsonrpc.websocket.subscription.request.SubscriptionType;
 import tech.pegasys.pantheon.ethereum.jsonrpc.websocket.subscription.request.UnsubscribeRequest;
@@ -49,6 +50,15 @@ public class SubscriptionManager extends AbstractVerticle {
   private final Map<Long, Subscription> subscriptions = new HashMap<>();
   private final Map<String, List<Long>> connectionSubscriptionsMap = new HashMap<>();
   private final SubscriptionBuilder subscriptionBuilder = new SubscriptionBuilder();
+  private final long refreshDelay;
+
+  public SubscriptionManager(final long refreshDelay) {
+    this.refreshDelay = refreshDelay;
+  }
+
+  public SubscriptionManager() {
+    this.refreshDelay = WebSocketConfiguration.DEFAULT_WEBSOCKET_REFRESH_DELAY;
+  }
 
   @Override
   public void start() {
@@ -162,4 +172,8 @@ public void sendMessage(final Long subscriptionId, final JsonRpcResult msg) {
         .findFirst()
         .ifPresent(connectionId -> vertx.eventBus().send(connectionId, Json.encode(response)));
   }
+
+  public long getRefreshDelay() {
+    return refreshDelay;
+  }
 }

ethereum/jsonrpc/src/main/java/tech/pegasys/pantheon/ethereum/jsonrpc/websocket/subscription/syncing/SyncingSubscriptionService.java

@@ -27,7 +27,6 @@ public class SyncingSubscriptionService {
 
   private final SubscriptionManager subscriptionManager;
   private final Synchronizer synchronizer;
-  private final long currentRefreshDelay = 5000;
 
   private Optional<SyncStatus> previousSyncStatus;
   private long timerId;
@@ -76,7 +75,7 @@ public void engageNextTimerTick() {
           subscriptionManager
               .getVertx()
               .setTimer(
-                  currentRefreshDelay,
+                  subscriptionManager.getRefreshDelay(),
                   (id) -> {
                     sendSyncingToMatchingSubscriptions();
                     engageNextTimerTick();

pantheon/src/main/java/tech/pegasys/pantheon/RunnerBuilder.java

@@ -260,7 +260,8 @@ public Runner build() {
               filterManager);
 
       final SubscriptionManager subscriptionManager =
-          createSubscriptionManager(vertx, transactionPool);
+          createSubscriptionManager(
+              vertx, transactionPool, webSocketConfiguration.getRefreshDelay());
 
       createLogsSubscriptionService(
           context.getBlockchain(), context.getWorldStateArchive(), subscriptionManager);
@@ -324,8 +325,8 @@ private Map<String, JsonRpcMethod> jsonRpcMethods(
   }
 
   private SubscriptionManager createSubscriptionManager(
-      final Vertx vertx, final TransactionPool transactionPool) {
-    final SubscriptionManager subscriptionManager = new SubscriptionManager();
+      final Vertx vertx, final TransactionPool transactionPool, final long refreshDelay) {
+    final SubscriptionManager subscriptionManager = new SubscriptionManager(refreshDelay);
     final PendingTransactionSubscriptionService pendingTransactions =
         new PendingTransactionSubscriptionService(subscriptionManager);
     transactionPool.addTransactionListener(pendingTransactions);

pantheon/src/main/java/tech/pegasys/pantheon/cli/PantheonCommand.java

@@ -312,6 +312,26 @@ public static class RpcApisConversionException extends Exception {
   )
   private final Collection<RpcApi> wsApis = null;
 
+  private Long refreshDelay;
+
+  @Option(
+    names = {"--ws-refresh-delay"},
+    paramLabel = "<refresh delay>",
+    arity = "1",
+    description =
+        "Refresh delay of websocket subscription sync in milliseconds. "
+            + "default: ${DEFAULT-VALUE}",
+    defaultValue = "5000"
+  )
+  private void setRefreshDelay(final Long refreshDelay) {
+    if (refreshDelay < 1 || refreshDelay > 3600000) {
+      throw new ParameterException(
+          new CommandLine(this),
+          "refreshDelay must be a positive integer smaller than 3600000 (1 hour)");
+    }
+    this.refreshDelay = refreshDelay;
+  }
+
   @Option(
     names = {"--host-whitelist"},
     paramLabel = "<hostname>",
@@ -504,6 +524,7 @@ private WebSocketConfiguration webSocketConfiguration() {
     webSocketConfiguration.setHost(wsHostAndPort.getHost());
     webSocketConfiguration.setPort(wsHostAndPort.getPort());
     webSocketConfiguration.setRpcApis(wsApis);
+    webSocketConfiguration.setRefreshDelay(refreshDelay);
     return webSocketConfiguration;
   }
 

pantheon/src/test/java/tech/pegasys/pantheon/cli/PantheonCommandTest.java

@@ -443,6 +443,28 @@ public void bootnodesOptionMustBeUsed() {
     assertThat(commandErrorOutput.toString()).isEmpty();
   }
 
+  @Test
+  public void callingWithRefreshDelayWithValueMustUseValue() {
+    parseCommand("--ws-refresh-delay", "2000");
+
+    verify(mockRunnerBuilder).webSocketConfiguration(wsRpcConfigArgumentCaptor.capture());
+    verify(mockRunnerBuilder).build();
+
+    assertThat(wsRpcConfigArgumentCaptor.getValue().getRefreshDelay()).isEqualTo(2000);
+
+    assertThat(commandOutput.toString()).isEmpty();
+    assertThat(commandErrorOutput.toString()).isEmpty();
+  }
+
+  @Test
+  public void callingWithRefreshDelayWithNegativeValueMustError() {
+    parseCommand("--ws-refresh-delay", "-2000");
+
+    assertThat(commandOutput.toString()).isEmpty();
+    final String expectedErrorMsg = "refreshDelay must be a positive integer";
+    assertThat(commandErrorOutput.toString()).startsWith(expectedErrorMsg);
+  }
+
   @Test
   public void callingWithNodesWhitelistOptionButNoValueMustNotError() {
     parseCommand("--nodes-whitelist");