Skip to content

Commit 0cafd92

Browse files
a-saksenaa-saksena
committed
Addressing Spotless styling issues.
Signed-off-by: a-saksena <[email protected]>
1 parent 59b1f6e commit 0cafd92

File tree

3 files changed

+115
-80
lines changed

3 files changed

+115
-80
lines changed

‎server/src/main/java/com/hedera/block/server/BlockStreamService.java

+67-47
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,9 @@
1616

1717
package com.hedera.block.server;
1818

19+
import static com.hedera.block.server.Constants.*;
20+
import static io.helidon.webserver.grpc.ResponseHelper.complete;
21+
1922
import com.google.protobuf.Descriptors;
2023
import com.hedera.block.protos.BlockStreamServiceGrpcProto;
2124
import com.hedera.block.server.consumer.LiveStreamObserver;
@@ -26,39 +29,44 @@
2629
import io.grpc.Status;
2730
import io.grpc.stub.StreamObserver;
2831
import io.helidon.webserver.grpc.GrpcService;
29-
3032
import java.time.Clock;
3133
import java.util.Optional;
3234

33-
import static io.helidon.webserver.grpc.ResponseHelper.complete;
34-
35-
import static com.hedera.block.server.Constants.*;
36-
3735
/**
38-
* This class implements the GrpcService interface and provides the functionality for the BlockStreamService.
39-
* It sets up the bidirectional streaming methods for the service and handles the routing for these methods.
40-
* It also initializes the StreamMediator, BlockStorage, and BlockCache upon creation.
36+
* This class implements the GrpcService interface and provides the functionality for the
37+
* BlockStreamService. It sets up the bidirectional streaming methods for the service and handles
38+
* the routing for these methods. It also initializes the StreamMediator, BlockStorage, and
39+
* BlockCache upon creation.
4140
*
42-
* <p>The class provides two main methods, streamSink and streamSource, which handle the client and server streaming
43-
* respectively. These methods return custom StreamObservers which are used to observe and respond to the streams.
41+
* <p>The class provides two main methods, streamSink and streamSource, which handle the client and
42+
* server streaming respectively. These methods return custom StreamObservers which are used to
43+
* observe and respond to the streams.
4444
*/
4545
public class BlockStreamService implements GrpcService {
4646

4747
private final System.Logger LOGGER = System.getLogger(getClass().getName());
4848

4949
private final long timeoutThresholdMillis;
50-
private final StreamMediator<BlockStreamServiceGrpcProto.Block, BlockStreamServiceGrpcProto.BlockResponse> streamMediator;
51-
private final BlockPersistenceHandler<BlockStreamServiceGrpcProto.Block> blockPersistenceHandler;
50+
private final StreamMediator<
51+
BlockStreamServiceGrpcProto.Block, BlockStreamServiceGrpcProto.BlockResponse>
52+
streamMediator;
53+
private final BlockPersistenceHandler<BlockStreamServiceGrpcProto.Block>
54+
blockPersistenceHandler;
5255

5356
/**
5457
* Constructor for the BlockStreamService class.
5558
*
5659
* @param timeoutThresholdMillis the timeout threshold in milliseconds
5760
* @param streamMediator the stream mediator
5861
*/
59-
public BlockStreamService(final long timeoutThresholdMillis,
60-
final StreamMediator<BlockStreamServiceGrpcProto.Block, BlockStreamServiceGrpcProto.BlockResponse> streamMediator,
61-
final BlockPersistenceHandler<BlockStreamServiceGrpcProto.Block> blockPersistenceHandler) {
62+
public BlockStreamService(
63+
final long timeoutThresholdMillis,
64+
final StreamMediator<
65+
BlockStreamServiceGrpcProto.Block,
66+
BlockStreamServiceGrpcProto.BlockResponse>
67+
streamMediator,
68+
final BlockPersistenceHandler<BlockStreamServiceGrpcProto.Block>
69+
blockPersistenceHandler) {
6270

6371
this.timeoutThresholdMillis = timeoutThresholdMillis;
6472
this.streamMediator = streamMediator;
@@ -76,8 +84,8 @@ public Descriptors.FileDescriptor proto() {
7684
}
7785

7886
/**
79-
* Returns the service name for the BlockStreamService. This service name corresponds to the service name in
80-
* the proto file.
87+
* Returns the service name for the BlockStreamService. This service name corresponds to the
88+
* service name in the proto file.
8189
*
8290
* @return the service name corresponding to the service name in the proto file
8391
*/
@@ -87,7 +95,8 @@ public String serviceName() {
8795
}
8896

8997
/**
90-
* Updates the routing for the BlockStreamService. It sets up the bidirectional streaming methods for the service.
98+
* Updates the routing for the BlockStreamService. It sets up the bidirectional streaming
99+
* methods for the service.
91100
*
92101
* @param routing the routing for the BlockStreamService
93102
*/
@@ -99,64 +108,75 @@ public void update(final Routing routing) {
99108
}
100109

101110
/**
102-
* The streamSink method is called by Helidon each time a producer initiates a bidirectional stream.
103-
*
104-
* @param responseStreamObserver Helidon provides a StreamObserver to handle responses back to the producer.
111+
* The streamSink method is called by Helidon each time a producer initiates a bidirectional
112+
* stream.
105113
*
106-
* @return a custom StreamObserver to handle streaming blocks from the producer to all subscribed consumer
107-
* via the streamMediator as well as sending responses back to the producer.
114+
* @param responseStreamObserver Helidon provides a StreamObserver to handle responses back to
115+
* the producer.
116+
* @return a custom StreamObserver to handle streaming blocks from the producer to all
117+
* subscribed consumer via the streamMediator as well as sending responses back to the
118+
* producer.
108119
*/
109120
private StreamObserver<BlockStreamServiceGrpcProto.Block> streamSink(
110-
final StreamObserver<BlockStreamServiceGrpcProto.BlockResponse> responseStreamObserver) {
121+
final StreamObserver<BlockStreamServiceGrpcProto.BlockResponse>
122+
responseStreamObserver) {
111123
LOGGER.log(System.Logger.Level.DEBUG, "Executing bidirectional streamSink method");
112124

113125
return new ProducerBlockStreamObserver(streamMediator, responseStreamObserver);
114126
}
115127

116128
/**
117-
* The streamSource method is called by Helidon each time a consumer initiates a bidirectional stream.
129+
* The streamSource method is called by Helidon each time a consumer initiates a bidirectional
130+
* stream.
118131
*
119-
* @param responseStreamObserver Helidon provides a StreamObserver to handle responses from the consumer
120-
* back to the server.
121-
*
122-
* @return a custom StreamObserver to handle streaming blocks from the producer to the consumer as well
123-
* as handling responses from the consumer.
132+
* @param responseStreamObserver Helidon provides a StreamObserver to handle responses from the
133+
* consumer back to the server.
134+
* @return a custom StreamObserver to handle streaming blocks from the producer to the consumer
135+
* as well as handling responses from the consumer.
124136
*/
125-
private StreamObserver<BlockStreamServiceGrpcProto.BlockResponse> streamSource(final StreamObserver<BlockStreamServiceGrpcProto.Block> responseStreamObserver) {
137+
private StreamObserver<BlockStreamServiceGrpcProto.BlockResponse> streamSource(
138+
final StreamObserver<BlockStreamServiceGrpcProto.Block> responseStreamObserver) {
126139
LOGGER.log(System.Logger.Level.DEBUG, "Executing bidirectional streamSource method");
127140

128141
// Return a custom StreamObserver to handle streaming blocks from the producer.
129-
final LiveStreamObserver<BlockStreamServiceGrpcProto.Block, BlockStreamServiceGrpcProto.BlockResponse> streamObserver = new LiveStreamObserverImpl(
130-
timeoutThresholdMillis,
131-
Clock.systemDefaultZone(),
132-
Clock.systemDefaultZone(),
133-
streamMediator,
134-
responseStreamObserver);
142+
final LiveStreamObserver<
143+
BlockStreamServiceGrpcProto.Block,
144+
BlockStreamServiceGrpcProto.BlockResponse>
145+
streamObserver =
146+
new LiveStreamObserverImpl(
147+
timeoutThresholdMillis,
148+
Clock.systemDefaultZone(),
149+
Clock.systemDefaultZone(),
150+
streamMediator,
151+
responseStreamObserver);
135152

136153
// Subscribe the observer to the mediator
137154
streamMediator.subscribe(streamObserver);
138155

139156
return streamObserver;
140157
}
141158

142-
private void getBlock(BlockStreamServiceGrpcProto.Block block, StreamObserver<BlockStreamServiceGrpcProto.Block> responseObserver) {
159+
private void getBlock(
160+
BlockStreamServiceGrpcProto.Block block,
161+
StreamObserver<BlockStreamServiceGrpcProto.Block> responseObserver) {
143162
String message = "GET BLOCK RESPONSE! ";
144163
LOGGER.log(System.Logger.Level.INFO, "GetBlock request received");
145-
Optional<BlockStreamServiceGrpcProto.Block> responseBlock = blockPersistenceHandler.read(block.getId());
146-
if(responseBlock.isPresent()) {
164+
Optional<BlockStreamServiceGrpcProto.Block> responseBlock =
165+
blockPersistenceHandler.read(block.getId());
166+
if (responseBlock.isPresent()) {
147167
LOGGER.log(System.Logger.Level.INFO, "SENDING BLOCK # " + block.getId());
148-
complete(responseObserver, responseBlock.get()); // TODO: Should return int and not quoted string
168+
complete(
169+
responseObserver,
170+
responseBlock.get()); // TODO: Should return int and not quoted string
149171
} else {
150172
LOGGER.log(System.Logger.Level.INFO, "DID NOT FIND YOUR BLOCK");
151173
// TODO: Fix below. It could return gRPC equivalent of 404 NOT FOUND
152-
responseObserver.onError(Status.NOT_FOUND
153-
.withDescription("DID NOT FIND YOUR BLOCK")
154-
.asRuntimeException()
155-
);
174+
responseObserver.onError(
175+
Status.NOT_FOUND
176+
.withDescription("DID NOT FIND YOUR BLOCK")
177+
.asRuntimeException());
156178
// Keeping below comments for the fix needed above.
157179
// complete(responseObserver, BlockStreamServiceGrpcProto.Block.getDefaultInstance());
158180
}
159181
}
160182
}
161-
162-

‎server/src/main/java/com/hedera/block/server/Constants.java

+3-4
Original file line numberDiff line numberDiff line change
@@ -16,15 +16,14 @@
1616

1717
package com.hedera.block.server;
1818

19-
/**
20-
* Constants used in the BlockNode service.
21-
*/
19+
/** Constants used in the BlockNode service. */
2220
public final class Constants {
2321
private Constants() {}
2422

2523
// Config Constants
2624
public static final String BLOCKNODE_STORAGE_ROOT_PATH_KEY = "blocknode.storage.root.path";
27-
public static final String BLOCKNODE_SERVER_CONSUMER_TIMEOUT_THRESHOLD_KEY = "blocknode.server.consumer.timeout.threshold";
25+
public static final String BLOCKNODE_SERVER_CONSUMER_TIMEOUT_THRESHOLD_KEY =
26+
"blocknode.server.consumer.timeout.threshold";
2827

2928
// Constants specified in the service definition of the .proto file
3029
public static final String SERVICE_NAME = "BlockStreamGrpc";

‎server/src/main/java/com/hedera/block/server/Server.java

+45-29
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,8 @@
1616

1717
package com.hedera.block.server;
1818

19+
import static com.hedera.block.server.Constants.*;
20+
1921
import com.hedera.block.protos.BlockStreamServiceGrpcProto;
2022
import com.hedera.block.server.mediator.LiveStreamMediatorImpl;
2123
import com.hedera.block.server.persistence.WriteThroughCacheHandler;
@@ -26,20 +28,22 @@
2628
import io.helidon.config.Config;
2729
import io.helidon.webserver.WebServer;
2830
import io.helidon.webserver.grpc.GrpcRouting;
29-
3031
import java.io.IOException;
3132
import java.util.stream.Stream;
3233

33-
import static com.hedera.block.server.Constants.*;
34-
35-
/**
36-
* Main class for the block node server
37-
*/
34+
/** Main class for the block node server */
3835
public class Server {
3936

40-
// Function stubs to satisfy the bidi routing param signatures. The implementations are in the service class.
41-
private static ServerCalls.BidiStreamingMethod<Stream<BlockStreamServiceGrpcProto.Block>, StreamObserver<BlockStreamServiceGrpcProto.Block>> clientBidiStreamingMethod;
42-
private static ServerCalls.BidiStreamingMethod<Stream<BlockStreamServiceGrpcProto.BlockResponse>, StreamObserver<BlockStreamServiceGrpcProto.Block>> serverBidiStreamingMethod;
37+
// Function stubs to satisfy the bidi routing param signatures. The implementations are in the
38+
// service class.
39+
private static ServerCalls.BidiStreamingMethod<
40+
Stream<BlockStreamServiceGrpcProto.Block>,
41+
StreamObserver<BlockStreamServiceGrpcProto.Block>>
42+
clientBidiStreamingMethod;
43+
private static ServerCalls.BidiStreamingMethod<
44+
Stream<BlockStreamServiceGrpcProto.BlockResponse>,
45+
StreamObserver<BlockStreamServiceGrpcProto.Block>>
46+
serverBidiStreamingMethod;
4347

4448
private static final System.Logger LOGGER = System.getLogger(Server.class.getName());
4549

@@ -59,33 +63,43 @@ public static void main(final String[] args) {
5963
Config.global(config);
6064

6165
// Get Timeout threshold from configuration
62-
final long consumerTimeoutThreshold = config.get(BLOCKNODE_SERVER_CONSUMER_TIMEOUT_THRESHOLD_KEY).asLong().orElse(1500L);
66+
final long consumerTimeoutThreshold =
67+
config.get(BLOCKNODE_SERVER_CONSUMER_TIMEOUT_THRESHOLD_KEY)
68+
.asLong()
69+
.orElse(1500L);
6370

6471
// Initialize the block storage, cache, and service
65-
final BlockStorage<BlockStreamServiceGrpcProto.Block> blockStorage = new FileSystemBlockStorage(BLOCKNODE_STORAGE_ROOT_PATH_KEY, config);
72+
final BlockStorage<BlockStreamServiceGrpcProto.Block> blockStorage =
73+
new FileSystemBlockStorage(BLOCKNODE_STORAGE_ROOT_PATH_KEY, config);
6674

6775
// Initialize blockStreamService with Live Stream and Cache
68-
final BlockStreamService blockStreamService = new BlockStreamService(consumerTimeoutThreshold,
69-
new LiveStreamMediatorImpl(new WriteThroughCacheHandler(blockStorage)),
70-
new WriteThroughCacheHandler(blockStorage));
76+
final BlockStreamService blockStreamService =
77+
new BlockStreamService(
78+
consumerTimeoutThreshold,
79+
new LiveStreamMediatorImpl(new WriteThroughCacheHandler(blockStorage)),
80+
new WriteThroughCacheHandler(blockStorage));
7181

7282
// Start the web server
7383
WebServer.builder()
7484
.port(8080)
75-
.addRouting(GrpcRouting.builder()
76-
.service(blockStreamService)
77-
.bidi(BlockStreamServiceGrpcProto.getDescriptor(),
78-
SERVICE_NAME,
79-
CLIENT_STREAMING_METHOD_NAME,
80-
clientBidiStreamingMethod)
81-
.bidi(BlockStreamServiceGrpcProto.getDescriptor(),
82-
SERVICE_NAME,
83-
SERVER_STREAMING_METHOD_NAME,
84-
serverBidiStreamingMethod)
85-
.unary(BlockStreamServiceGrpcProto.getDescriptor(),
86-
SERVICE_NAME,
87-
GET_BLOCK_METHOD_NAME,
88-
Server::grpcGetBlock))
85+
.addRouting(
86+
GrpcRouting.builder()
87+
.service(blockStreamService)
88+
.bidi(
89+
BlockStreamServiceGrpcProto.getDescriptor(),
90+
SERVICE_NAME,
91+
CLIENT_STREAMING_METHOD_NAME,
92+
clientBidiStreamingMethod)
93+
.bidi(
94+
BlockStreamServiceGrpcProto.getDescriptor(),
95+
SERVICE_NAME,
96+
SERVER_STREAMING_METHOD_NAME,
97+
serverBidiStreamingMethod)
98+
.unary(
99+
BlockStreamServiceGrpcProto.getDescriptor(),
100+
SERVICE_NAME,
101+
GET_BLOCK_METHOD_NAME,
102+
Server::grpcGetBlock))
89103
.build()
90104
.start();
91105
} catch (IOException e) {
@@ -94,5 +108,7 @@ public static void main(final String[] args) {
94108
}
95109
}
96110

97-
static void grpcGetBlock(BlockStreamServiceGrpcProto.BlockRequest request, StreamObserver<BlockStreamServiceGrpcProto.Block> responseObserver) {}
111+
static void grpcGetBlock(
112+
BlockStreamServiceGrpcProto.BlockRequest request,
113+
StreamObserver<BlockStreamServiceGrpcProto.Block> responseObserver) {}
98114
}

0 commit comments

Comments
 (0)