Skip to content

fix: Moved GRPC services over to use PBJ generated ones #1155

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 2 commits into
base: main
Choose a base branch
from
Open

fix: Moved GRPC services over to use PBJ generated ones #1155

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b007379
Moved GRPC services over to use PBJ generated ones
jasperpotts May 16, 2025
952a363
Merge branch 'main' into 1088-move-grpc-servcies-to-pbj
jasperpotts May 16, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 2 additions & 3 deletions block-node/block-access/src/main/java/module-info.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,11 +4,10 @@
module org.hiero.block.node.access.service {
uses com.swirlds.config.api.spi.ConfigurationBuilderFactory;

requires transitive com.hedera.pbj.runtime;
requires transitive org.hiero.block.protobuf;
requires transitive org.hiero.block.node.spi;
requires com.swirlds.metrics.api;
requires org.hiero.block.protobuf;
requires com.github.spotbugs.annotations;
requires com.hedera.pbj.runtime;

provides org.hiero.block.node.spi.BlockNodePlugin with
BlockAccessServicePlugin;
Expand Down
86 changes: 13 additions & 73 deletions ...ck-access/src/main/java/org/hiero/block/node/access/service/BlockAccessServicePlugin.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,20 +6,12 @@
import static java.lang.System.Logger.Level.WARNING;

import com.hedera.hapi.block.stream.Block;
import com.hedera.pbj.runtime.grpc.GrpcException;
import com.hedera.pbj.runtime.grpc.Pipeline;
import com.hedera.pbj.runtime.grpc.Pipelines;
import com.hedera.pbj.runtime.grpc.ServiceInterface;
import com.hedera.pbj.runtime.io.buffer.Bytes;
import com.swirlds.metrics.api.Counter;
import edu.umd.cs.findbugs.annotations.NonNull;
import java.util.Arrays;
import java.util.List;
import org.hiero.block.api.BlockAccessServiceInterface;
import org.hiero.block.api.BlockRequest;
import org.hiero.block.api.BlockResponse;
import org.hiero.block.api.BlockResponse.Code;
// PBJ doesn't generate GRPC stubs for some reason, also the proto file is broken when PBJ compiles it...
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Might want to remove this comment as well.

import org.hiero.block.api.protoc.BlockAccessServiceGrpc;
import org.hiero.block.node.spi.BlockNodeContext;
import org.hiero.block.node.spi.BlockNodePlugin;
import org.hiero.block.node.spi.ServiceBuilder;
Expand All @@ -28,7 +20,7 @@
/**
* Plugin that implements the BlockAccessService and provides the 'block' RPC.
*/
public class BlockAccessServicePlugin implements BlockNodePlugin, ServiceInterface {
public class BlockAccessServicePlugin implements BlockNodePlugin, BlockAccessServiceInterface {

/** The logger for this class. */
private final System.Logger LOGGER = System.getLogger(getClass().getName());
Expand All @@ -43,13 +35,15 @@ public class BlockAccessServicePlugin implements BlockNodePlugin, ServiceInterfa
/** Counter for the number of responses not found */
private Counter responseCounterNotFound;

// ==== BlockAccessServiceInterface Methods ========================================================================

/**
* Handle a request for a single block
*
* @param request the request containing the block number or latest flag
* @return the response containing the block or an error status
*/
private BlockResponse handleBlockRequest(BlockRequest request) {
public BlockResponse getBlock(BlockRequest request) {
LOGGER.log(DEBUG, "Received BlockRequest for block number: {0}", request.blockNumber());
requestCounter.increment();

Expand All @@ -71,7 +65,7 @@ private BlockResponse handleBlockRequest(BlockRequest request) {
blockNumberToRetrieve = request.blockNumber();
}

// Check if block is within available range
// Check if block is within the available range
if (!blockProvider.availableBlocks().contains(blockNumberToRetrieve)) {
long lowestBlockNumber = blockProvider.availableBlocks().min();
long highestBlockNumber = blockProvider.availableBlocks().max();
Expand Down Expand Up @@ -99,11 +93,18 @@ private BlockResponse handleBlockRequest(BlockRequest request) {
}

// ==== BlockNodePlugin Methods ====================================================================================

/**
* {@inheritDoc}
*/
@Override
public String name() {
return "BlockAccessServicePlugin";
}

/**
* {@inheritDoc}
*/
@Override
public void init(BlockNodeContext context, ServiceBuilder serviceBuilder) {
// Create the metrics
Expand All @@ -124,65 +125,4 @@ public void init(BlockNodeContext context, ServiceBuilder serviceBuilder) {
// Register this service
serviceBuilder.registerGrpcService(this);
}

// ==== ServiceInterface Methods ===================================================================================
/**
* BlockAccessService methods define the gRPC methods available on the BlockAccessService.
*/
enum BlockAccessServiceMethod implements Method {
/**
* The getBlock method retrieves a single block from the block node.
*/
getBlock
}

/**
* {@inheritDoc}
*/
@NonNull
@Override
public String serviceName() {
String[] parts = fullName().split("\\.");
return parts[parts.length - 1];
}

/**
* {@inheritDoc}
*/
@NonNull
@Override
public String fullName() {
return BlockAccessServiceGrpc.SERVICE_NAME;
}

/**
* {@inheritDoc}
*/
@NonNull
@Override
public List<Method> methods() {
return Arrays.asList(BlockAccessServiceMethod.values());
}

/**
* {@inheritDoc}
*
* This is called each time a new request is received.
*/
@NonNull
@Override
public Pipeline<? super Bytes> open(
@NonNull Method method, @NonNull RequestOptions requestOptions, @NonNull Pipeline<? super Bytes> pipeline)
throws GrpcException {
final BlockAccessServiceMethod blockAccessServiceMethod = (BlockAccessServiceMethod) method;
return switch (blockAccessServiceMethod) {
case getBlock:
yield Pipelines.<BlockRequest, BlockResponse>unary()
.mapRequest(BlockRequest.PROTOBUF::parse)
.method(this::handleBlockRequest)
.mapResponse(BlockResponse.PROTOBUF::toBytes)
.respondTo(pipeline)
.build();
};
}
}
6 changes: 3 additions & 3 deletions ...ccess/src/test/java/org/hiero/block/node/access/service/BlockAccessServicePluginTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
// SPDX-License-Identifier: Apache-2.0
package org.hiero.block.node.access.service;

import static org.hiero.block.node.access.service.BlockAccessServicePlugin.BlockAccessServiceMethod.getBlock;
import static org.hiero.block.node.app.fixtures.TestUtils.enableDebugLogging;
import static org.hiero.block.node.app.fixtures.blocks.BlockItemUtils.toBlockItemsUnparsed;
import static org.hiero.block.node.app.fixtures.blocks.SimpleTestBlockItemBuilder.createNumberOfVerySimpleBlocks;
Expand All @@ -25,10 +24,11 @@
import org.junit.jupiter.api.Test;

public class BlockAccessServicePluginTest extends GrpcPluginTestBase<BlockAccessServicePlugin> {
private final BlockAccessServicePlugin plugin = new BlockAccessServicePlugin();

public BlockAccessServicePluginTest() {
super();
start(new BlockAccessServicePlugin(), getBlock, new SimpleInMemoryHistoricalBlockFacility());
start(plugin, plugin.methods().getFirst(), new SimpleInMemoryHistoricalBlockFacility());
}

/**
Expand All @@ -54,7 +54,7 @@ void testServiceInterfaceBasics() {
List<ServiceInterface.Method> methods = serviceInterface.methods();
assertNotNull(methods);
assertEquals(1, methods.size());
assertEquals(getBlock, methods.getFirst());
assertEquals(plugin.methods().getFirst(), methods.getFirst());
}

@Test
Expand Down
59 changes: 16 additions & 43 deletions ...k-node/publisher/src/main/java/org/hiero/block/node/publisher/PublisherServicePlugin.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,12 +8,10 @@
import com.hedera.pbj.runtime.grpc.GrpcException;
import com.hedera.pbj.runtime.grpc.Pipeline;
import com.hedera.pbj.runtime.grpc.Pipelines;
import com.hedera.pbj.runtime.grpc.ServiceInterface;
import com.hedera.pbj.runtime.io.buffer.Bytes;
import com.swirlds.metrics.api.Counter;
import com.swirlds.metrics.api.LongGauge;
import edu.umd.cs.findbugs.annotations.NonNull;
import java.util.Arrays;
import java.util.Comparator;
import java.util.HashSet;
import java.util.List;
Expand All @@ -23,8 +21,9 @@
import java.util.OptionalLong;
import java.util.Set;
import java.util.concurrent.locks.ReentrantLock;
import org.hiero.block.api.BlockStreamPublishServiceInterface;
import org.hiero.block.api.PublishStreamRequest;
import org.hiero.block.api.PublishStreamResponse;
import org.hiero.block.api.protoc.BlockStreamPublishServiceGrpc;
import org.hiero.block.internal.BlockItemUnparsed;
import org.hiero.block.internal.PublishStreamRequestUnparsed;
import org.hiero.block.node.publisher.PublisherConfig.PublisherType;
Expand Down Expand Up @@ -59,7 +58,8 @@
* <p>
* TODO Still lots to work out on tracking the various stages of blocks, latest in flight, etc.
*/
public final class PublisherServicePlugin implements BlockNodePlugin, ServiceInterface, BlockNotificationHandler {
public final class PublisherServicePlugin
implements BlockNodePlugin, BlockStreamPublishServiceInterface, BlockNotificationHandler {
/** The logger for this class. */
private final System.Logger LOGGER = System.getLogger(getClass().getName());

Expand Down Expand Up @@ -465,43 +465,8 @@ public void handlePersisted(PersistedNotification notification) {
}
}

// ==== ServiceInterface Methods ===================================================================================

/**
* BlockStreamPublisherService types define the gRPC methods available on the BlockStreamPublisherService.
*/
enum BlockStreamPublisherServiceMethod implements Method {
/**
* The publishBlockStream method represents the bidirectional gRPC streaming method
* Consensus Nodes should use to publish the BlockStream to the Block Node.
*/
publishBlockStream
}

/**
* {@inheritDoc}
*/
@NonNull
public String serviceName() {
String[] parts = fullName().split("\\.");
return parts[parts.length - 1];
}

/**
* {@inheritDoc}
*/
@NonNull
public String fullName() {
return BlockStreamPublishServiceGrpc.SERVICE_NAME;
}

/**
* {@inheritDoc}
*/
@NonNull
public List<Method> methods() {
return Arrays.asList(BlockStreamPublisherServiceMethod.values());
}
// ==== BlockStreamPublishServiceInterface Methods
// ===================================================================================

/**
* {@inheritDoc}
Expand All @@ -516,8 +481,8 @@ public Pipeline<? super Bytes> open(
throws GrpcException {
stateLock.lock();
try {
final BlockStreamPublisherServiceMethod blockStreamPublisherServiceMethod =
(BlockStreamPublisherServiceMethod) method;
final BlockStreamPublishServiceMethod blockStreamPublisherServiceMethod =
(BlockStreamPublishServiceMethod) method;
return switch (blockStreamPublisherServiceMethod) {
case publishBlockStream:
final var pipe = Pipelines.<List<BlockItemUnparsed>, PublishStreamResponse>bidiStreaming()
Expand Down Expand Up @@ -555,4 +520,12 @@ public Pipeline<? super Bytes> open(
stateLock.unlock();
}
}

/**
* Never used, but required by the interface. We override the open method to handle directly.
*/
@Override
public Pipeline<? super PublishStreamRequest> publishBlockStream(Pipeline<? super PublishStreamResponse> replies) {
return null;
}
}
9 changes: 6 additions & 3 deletions block-node/publisher/src/test/java/org/hiero/block/node/publisher/PublisherTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,6 @@
import static org.hiero.block.node.app.fixtures.blocks.SimpleTestBlockItemBuilder.sampleBlockHeader;
import static org.hiero.block.node.app.fixtures.blocks.SimpleTestBlockItemBuilder.sampleBlockProof;
import static org.hiero.block.node.app.fixtures.blocks.SimpleTestBlockItemBuilder.sampleRoundHeader;
import static org.hiero.block.node.publisher.PublisherServicePlugin.BlockStreamPublisherServiceMethod.publishBlockStream;
import static org.hiero.block.node.spi.BlockNodePlugin.UNKNOWN_BLOCK_NUMBER;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
Expand All @@ -21,6 +20,7 @@
import com.hedera.pbj.runtime.io.buffer.Bytes;
import java.util.List;
import org.hiero.block.api.BlockItemSet;
import org.hiero.block.api.BlockStreamPublishServiceInterface.BlockStreamPublishServiceMethod;
import org.hiero.block.api.PublishStreamRequest;
import org.hiero.block.api.PublishStreamRequest.RequestOneOfType;
import org.hiero.block.api.PublishStreamResponse;
Expand All @@ -42,7 +42,10 @@
public class PublisherTest extends GrpcPluginTestBase<PublisherServicePlugin> {

public PublisherTest() {
start(new PublisherServicePlugin(), publishBlockStream, new NoBlocksHistoricalBlockFacility());
start(
new PublisherServicePlugin(),
BlockStreamPublishServiceMethod.publishBlockStream,
new NoBlocksHistoricalBlockFacility());
}

@Test
Expand All @@ -53,7 +56,7 @@ void testServiceInterfaceBasics() {
List<Method> methods = serviceInterface.methods();
assertNotNull(methods);
assertEquals(1, methods.size());
assertEquals(publishBlockStream, methods.getFirst());
assertEquals(BlockStreamPublishServiceMethod.publishBlockStream, methods.getFirst());
}

@Test
Expand Down
Loading
Loading