feat: Move out concrete working implementation and add working modes

simulator/src/main/java/com/hedera/block/simulator/BlockStreamSimulator.java

@@ -46,7 +46,7 @@ private BlockStreamSimulator() {}
     public static void main(String[] args)
             throws IOException, InterruptedException, BlockSimulatorParsingException {
 
-        LOGGER.log(INFO, "Starting Block Stream Simulator");
+        LOGGER.log(INFO, "Starting Block Stream Simulator!");
 
         ConfigurationBuilder configurationBuilder =
                 ConfigurationBuilder.create()

simulator/src/main/java/com/hedera/block/simulator/BlockStreamSimulatorApp.java

@@ -17,11 +17,14 @@
 package com.hedera.block.simulator;
 
 import com.hedera.block.simulator.config.data.BlockStreamConfig;
-import com.hedera.block.simulator.config.types.StreamingMode;
+import com.hedera.block.simulator.config.types.SimulatorMode;
 import com.hedera.block.simulator.exception.BlockSimulatorParsingException;
 import com.hedera.block.simulator.generator.BlockStreamManager;
 import com.hedera.block.simulator.grpc.PublishStreamGrpcClient;
-import com.hedera.hapi.block.stream.Block;
+import com.hedera.block.simulator.mode.CombinedModeHandler;
+import com.hedera.block.simulator.mode.ConsumerModeHandler;
+import com.hedera.block.simulator.mode.PublisherModeHandler;
+import com.hedera.block.simulator.mode.SimulatorModeHandler;
 import com.swirlds.config.api.Configuration;
 import edu.umd.cs.findbugs.annotations.NonNull;
 import java.io.IOException;
@@ -37,10 +40,7 @@ public class BlockStreamSimulatorApp {
     private final BlockStreamManager blockStreamManager;
     private final PublishStreamGrpcClient publishStreamGrpcClient;
     private final BlockStreamConfig blockStreamConfig;
-    private final StreamingMode streamingMode;
-
-    private final int delayBetweenBlockItems;
-    private final int millisecondsPerBlock;
+    private final SimulatorModeHandler simulatorModeHandler;
     private final AtomicBoolean isRunning = new AtomicBoolean(false);
 
     /**
@@ -60,9 +60,15 @@ public BlockStreamSimulatorApp(
 
         blockStreamConfig = configuration.getConfigData(BlockStreamConfig.class);
 
-        streamingMode = blockStreamConfig.streamingMode();
-        millisecondsPerBlock = blockStreamConfig.millisecondsPerBlock();
-        delayBetweenBlockItems = blockStreamConfig.delayBetweenBlockItems();
+        SimulatorMode simulatorMode = blockStreamConfig.simulatorMode();
+        if (simulatorMode == SimulatorMode.PUBLISHER) {
+            simulatorModeHandler =
+                    new PublisherModeHandler(blockStreamConfig, publishStreamGrpcClient);
+        } else if (simulatorMode == SimulatorMode.CONSUMER) {
+            simulatorModeHandler = new ConsumerModeHandler(blockStreamConfig);
+        } else {
+            simulatorModeHandler = new CombinedModeHandler(blockStreamConfig);
+        }
     }
 
     /**
@@ -73,77 +79,13 @@ public BlockStreamSimulatorApp(
      * @throws IOException if an I/O error occurs
      */
     public void start() throws InterruptedException, BlockSimulatorParsingException, IOException {
-
+        LOGGER.log(
+                System.Logger.Level.INFO,
+                "Block Stream Simulator started initializing components...");
+        publishStreamGrpcClient.init();
         isRunning.set(true);
-        LOGGER.log(System.Logger.Level.INFO, "Block Stream Simulator has started");
-
-        if (streamingMode == StreamingMode.MILLIS_PER_BLOCK) {
-            millisPerBlockStreaming();
-        } else {
-            constantRateStreaming();
-        }
-
-        LOGGER.log(System.Logger.Level.INFO, "Block Stream Simulator has stopped");
-    }
-
-    private void millisPerBlockStreaming()
-            throws IOException, InterruptedException, BlockSimulatorParsingException {
 
-        final long secondsPerBlockNanos = millisecondsPerBlock * 1_000_000L;
-
-        Block nextBlock = blockStreamManager.getNextBlock();
-        while (nextBlock != null) {
-            long startTime = System.nanoTime();
-            publishStreamGrpcClient.streamBlock(nextBlock);
-            long elapsedTime = System.nanoTime() - startTime;
-            long timeToDelay = secondsPerBlockNanos - elapsedTime;
-            if (timeToDelay > 0) {
-                Thread.sleep(timeToDelay / 1_000_000, (int) (timeToDelay % 1_000_000));
-            } else {
-                LOGGER.log(
-                        System.Logger.Level.WARNING,
-                        "Block Server is running behind. Streaming took: "
-                                + (elapsedTime / 1_000_000)
-                                + "ms - Longer than max expected of: "
-                                + millisecondsPerBlock
-                                + " milliseconds");
-            }
-            nextBlock = blockStreamManager.getNextBlock();
-        }
-        LOGGER.log(System.Logger.Level.INFO, "Block Stream Simulator has stopped");
-    }
-
-    private void constantRateStreaming()
-            throws InterruptedException, IOException, BlockSimulatorParsingException {
-        int delayMSBetweenBlockItems = delayBetweenBlockItems / 1_000_000;
-        int delayNSBetweenBlockItems = delayBetweenBlockItems % 1_000_000;
-        boolean streamBlockItem = true;
-        int blockItemsStreamed = 0;
-
-        while (streamBlockItem) {
-            // get block
-            Block block = blockStreamManager.getNextBlock();
-
-            if (block == null) {
-                LOGGER.log(
-                        System.Logger.Level.INFO,
-                        "Block Stream Simulator has reached the end of the block items");
-                break;
-            }
-
-            publishStreamGrpcClient.streamBlock(block);
-            blockItemsStreamed += block.items().size();
-
-            Thread.sleep(delayMSBetweenBlockItems, delayNSBetweenBlockItems);
-
-            if (blockItemsStreamed >= blockStreamConfig.maxBlockItemsToStream()) {
-                LOGGER.log(
-                        System.Logger.Level.INFO,
-                        "Block Stream Simulator has reached the maximum number of block items to"
-                                + " stream");
-                streamBlockItem = false;
-            }
-        }
+        simulatorModeHandler.start(this.blockStreamManager);
     }
 
     /**
@@ -155,8 +97,9 @@ public boolean isRunning() {
         return isRunning.get();
     }
 
-    /** Stops the block stream simulator. */
+    /** Stops the Block Stream Simulator and closes off all grpc channels. */
     public void stop() {
+        publishStreamGrpcClient.shutdown();
         isRunning.set(false);
         LOGGER.log(System.Logger.Level.INFO, "Block Stream Simulator has stopped");
     }

simulator/src/main/java/com/hedera/block/simulator/config/data/BlockStreamConfig.java

@@ -16,13 +16,15 @@
 
 package com.hedera.block.simulator.config.data;
 
+import com.hedera.block.simulator.config.types.SimulatorMode;
 import com.hedera.block.simulator.config.types.StreamingMode;
 import com.swirlds.config.api.ConfigData;
 import com.swirlds.config.api.ConfigProperty;
 
 /**
  * Defines the configuration data for the block stream in the Hedera Block Simulator.
  *
+ * @param simulatorMode the mode of the simulator, in terms of publishing, consuming or both
  * @param delayBetweenBlockItems the delay in microseconds between streaming each block item
  * @param maxBlockItemsToStream the maximum number of block items to stream before stopping
  * @param streamingMode the mode of streaming for the block stream (e.g., time-based, count-based)
@@ -31,6 +33,7 @@
  */
 @ConfigData("blockStream")
 public record BlockStreamConfig(
+        @ConfigProperty(defaultValue = "PUBLISHER") SimulatorMode simulatorMode,
         @ConfigProperty(defaultValue = "1_500_000") int delayBetweenBlockItems,
         @ConfigProperty(defaultValue = "100_000") int maxBlockItemsToStream,
         @ConfigProperty(defaultValue = "MILLIS_PER_BLOCK") StreamingMode streamingMode,
@@ -50,6 +53,7 @@ public static Builder builder() {
      * A builder for creating instances of {@link BlockStreamConfig}.
      */
     public static class Builder {
+        private SimulatorMode simulatorMode = SimulatorMode.PUBLISHER;
         private int delayBetweenBlockItems = 1_500_000;
         private int maxBlockItemsToStream = 10_000;
         private StreamingMode streamingMode = StreamingMode.MILLIS_PER_BLOCK;
@@ -63,6 +67,17 @@ public Builder() {
             // Default constructor
         }
 
+        /**
+         * Sets the simulator mode for the block stream.
+         *
+         * @param simulatorMode the {@link SimulatorMode} to use
+         * @return this {@code Builder} instance
+         */
+        public Builder simulatorMode(SimulatorMode simulatorMode) {
+            this.simulatorMode = simulatorMode;
+            return this;
+        }
+
         /**
          * Sets the delay between streaming each block item.
          *
@@ -125,6 +140,7 @@ public Builder blockItemsBatchSize(int blockItemsBatchSize) {
          */
         public BlockStreamConfig build() {
             return new BlockStreamConfig(
+                    simulatorMode,
                     delayBetweenBlockItems,
                     maxBlockItemsToStream,
                     streamingMode,

simulator/src/main/java/com/hedera/block/simulator/config/types/SimulatorMode.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright (C) 2024 Hedera Hashgraph, LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.hedera.block.simulator.config.types;
+
+/** The SimulatorMode enum defines the work modes of the block stream simulator. */
+public enum SimulatorMode {
+    /**
+     * Indicates a work mode in which the simulator is working as both consumer and publisher.
+     */
+    BOTH,
+    /**
+     * Indicates a work mode in which the simulator is working in consumer mode.
+     */
+    CONSUMER,
+    /**
+     * Indicates a work mode in which the simulator is working in publisher mode.
+     */
+    PUBLISHER
+}

simulator/src/main/java/com/hedera/block/simulator/grpc/PublishStreamGrpcClient.java

@@ -24,6 +24,11 @@
  * The PublishStreamGrpcClient interface provides the methods to stream the block and block item.
  */
 public interface PublishStreamGrpcClient {
+    /**
+     * Initialize, opens a gRPC channel and creates the needed stubs with the passed configuration.
+     */
+    void init();
+
     /**
      * Streams the block item.
      *
@@ -39,4 +44,9 @@ public interface PublishStreamGrpcClient {
      * @return true if the block is streamed successfully, false otherwise
      */
     boolean streamBlock(Block block);
+
+    /**
+     * Shutdowns the channel.
+     */
+    void shutdown();
 }

simulator/src/main/java/com/hedera/block/simulator/grpc/PublishStreamGrpcClientImpl.java

@@ -37,9 +37,11 @@
  */
 public class PublishStreamGrpcClientImpl implements PublishStreamGrpcClient {
 
-    private final BlockStreamServiceGrpc.BlockStreamServiceStub stub;
-    private final StreamObserver<PublishStreamRequest> requestStreamObserver;
+    private BlockStreamServiceGrpc.BlockStreamServiceStub stub;
+    private StreamObserver<PublishStreamRequest> requestStreamObserver;
     private final BlockStreamConfig blockStreamConfig;
+    private final GrpcConfig grpcConfig;
+    private ManagedChannel channel;
 
     /**
      * Creates a new PublishStreamGrpcClientImpl instance.
@@ -50,14 +52,22 @@ public class PublishStreamGrpcClientImpl implements PublishStreamGrpcClient {
     @Inject
     public PublishStreamGrpcClientImpl(
             @NonNull GrpcConfig grpcConfig, @NonNull BlockStreamConfig blockStreamConfig) {
-        ManagedChannel channel =
+        this.grpcConfig = grpcConfig;
+        this.blockStreamConfig = blockStreamConfig;
+    }
+
+    /**
+     * Initialize the channel and stub for publishBlockStream with the desired configuration.
+     */
+    @Override
+    public void init() {
+        channel =
                 ManagedChannelBuilder.forAddress(grpcConfig.serverAddress(), grpcConfig.port())
                         .usePlaintext()
                         .build();
         stub = BlockStreamServiceGrpc.newStub(channel);
         PublishStreamObserver publishStreamObserver = new PublishStreamObserver();
         requestStreamObserver = stub.publishBlockStream(publishStreamObserver);
-        this.blockStreamConfig = blockStreamConfig;
     }
 
     /**
@@ -99,4 +109,9 @@ public boolean streamBlock(Block block) {
 
         return true;
     }
+
+    @Override
+    public void shutdown() {
+        channel.shutdown();
+    }
 }

simulator/src/main/java/com/hedera/block/simulator/mode/CombinedModeHandler.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright (C) 2024 Hedera Hashgraph, LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.hedera.block.simulator.mode;
+
+import static java.util.Objects.requireNonNull;
+
+import com.hedera.block.simulator.config.data.BlockStreamConfig;
+import com.hedera.block.simulator.generator.BlockStreamManager;
+import edu.umd.cs.findbugs.annotations.NonNull;
+
+/**
+ * The {@code CombinedModeHandler} class implements the {@link SimulatorModeHandler} interface
+ * and provides the behavior for a mode where both consuming and publishing of block data
+ * occur simultaneously.
+ *
+ * <p>This mode handles dual operations in the block streaming process, utilizing the
+ * {@link BlockStreamConfig} for configuration settings. It is designed for scenarios where
+ * the simulator needs to handle both the consumption and publication of blocks in parallel.
+ *
+ * <p>For now, the actual start behavior is not implemented, as indicated by the
+ * {@link UnsupportedOperationException}.
+ */
+public class CombinedModeHandler implements SimulatorModeHandler {
+    private final BlockStreamConfig blockStreamConfig;
+
+    /**
+     * Constructs a new {@code CombinedModeHandler} with the specified block stream configuration.
+     *
+     * @param blockStreamConfig the configuration data for managing block streams
+     */
+    public CombinedModeHandler(@NonNull final BlockStreamConfig blockStreamConfig) {
+        requireNonNull(blockStreamConfig);
+        this.blockStreamConfig = blockStreamConfig;
+    }
+
+    /**
+     * Starts the simulator in combined mode, handling both consumption and publication
+     * of block stream. However, this method is currently not implemented, and will throw
+     * an {@link UnsupportedOperationException}.
+     *
+     * @param blockStreamManager the {@link BlockStreamManager} responsible for managing block streams
+     * @throws UnsupportedOperationException as the method is not yet implemented
+     */
+    @Override
+    public void start(@NonNull BlockStreamManager blockStreamManager) {
+        throw new UnsupportedOperationException();
+    }
+}