allow upgrade whether websockets enabled or not (#4019)

* allow upgrade whether websockets enabled or not

Signed-off-by: Justin Florentine <justin+github@florentine.us>

Co-authored-by: Sally MacFarlane <sally.macfarlane@consensys.net>

besu/src/main/java/org/hyperledger/besu/RunnerBuilder.java

@@ -627,6 +627,9 @@ public class RunnerBuilder {
                   new HealthService(new ReadinessCheck(peerNetwork, synchronizer))));
     }
 
+    final SubscriptionManager subscriptionManager =
+        createSubscriptionManager(vertx, transactionPool, blockchainQueries);
+
     Optional<JsonRpcService> engineJsonRpcService = Optional.empty();
     if (engineJsonRpcConfiguration.isPresent() && engineJsonRpcConfiguration.get().isEnabled()) {
       final Map<String, JsonRpcMethod> engineMethods =
@@ -669,6 +672,9 @@ public class RunnerBuilder {
               ? webSocketConfiguration
               : WebSocketConfiguration.createEngineDefault();
 
+      final WebSocketMethodsFactory websocketMethodsFactory =
+          new WebSocketMethodsFactory(subscriptionManager, engineMethods);
+
       engineJsonRpcService =
           Optional.of(
               new JsonRpcService(
@@ -677,7 +683,7 @@ public class RunnerBuilder {
                   engineJsonRpcConfiguration.orElse(JsonRpcConfiguration.createEngineDefault()),
                   metricsSystem,
                   natService,
-                  engineMethods,
+                  websocketMethodsFactory.methods(),
                   Optional.ofNullable(engineSocketConfig),
                   besuController.getProtocolManager().ethContext().getScheduler(),
                   authToUse,
@@ -743,9 +749,6 @@ public class RunnerBuilder {
               dataDir,
               rpcEndpointServiceImpl);
 
-      final SubscriptionManager subscriptionManager =
-          createSubscriptionManager(vertx, transactionPool, blockchainQueries);
-
       createLogsSubscriptionService(
           context.getBlockchain(),
           context.getWorldStateArchive(),

besu/src/test/java/org/hyperledger/besu/RunnerBuilderTest.java

@@ -304,6 +304,49 @@ public final class RunnerBuilderTest {
     assertThat(runner.getEngineJsonRpcPort()).isPresent();
   }
 
+  @Test
+  public void whenEngineApiAddedEthSubscribeAvailable() {
+    WebSocketConfiguration wsRpc = WebSocketConfiguration.createDefault();
+    wsRpc.setEnabled(true);
+    EthNetworkConfig mockMainnet = mock(EthNetworkConfig.class);
+    when(mockMainnet.getNetworkId()).thenReturn(BigInteger.ONE);
+    MergeConfigOptions.setMergeEnabled(true);
+    when(besuController.getMiningCoordinator()).thenReturn(mock(MergeMiningCoordinator.class));
+    JsonRpcConfiguration engineConf = JsonRpcConfiguration.createEngineDefault();
+    engineConf.setEnabled(true);
+
+    final Runner runner =
+        new RunnerBuilder()
+            .discovery(true)
+            .p2pListenInterface("0.0.0.0")
+            .p2pListenPort(30303)
+            .p2pAdvertisedHost("127.0.0.1")
+            .p2pEnabled(true)
+            .natMethod(NatMethod.NONE)
+            .besuController(besuController)
+            .ethNetworkConfig(mockMainnet)
+            .metricsSystem(mock(ObservableMetricsSystem.class))
+            .permissioningService(mock(PermissioningServiceImpl.class))
+            .jsonRpcConfiguration(JsonRpcConfiguration.createDefault())
+            .engineJsonRpcConfiguration(engineConf)
+            .webSocketConfiguration(wsRpc)
+            .jsonRpcIpcConfiguration(mock(JsonRpcIpcConfiguration.class))
+            .graphQLConfiguration(mock(GraphQLConfiguration.class))
+            .metricsConfiguration(mock(MetricsConfiguration.class))
+            .vertx(Vertx.vertx())
+            .dataDir(dataDir.getRoot().toPath())
+            .storageProvider(mock(KeyValueStorageProvider.class))
+            .forkIdSupplier(() -> Collections.singletonList(Bytes.EMPTY))
+            .rpcEndpointService(new RpcEndpointServiceImpl())
+            .besuPluginContext(mock(BesuPluginContextImpl.class))
+            .build();
+
+    assertThat(runner.getEngineJsonRpcPort()).isPresent();
+    runner.startExternalServices();
+    // assert that rpc method collection has eth_subscribe in it.
+    runner.stop();
+  }
+
   @Test
   public void noEngineApiNoServiceForMethods() {
     JsonRpcConfiguration defaultRpcConfig = JsonRpcConfiguration.createDefault();