Skip to content

Commit 65ce8b0

Browse files
mattp-swirldslabsmattp-swirldslabs
committed
fix: used spotless
Signed-off-by: Matt Peterson <[email protected]>
1 parent 9fa5a54 commit 65ce8b0

16 files changed

+426
-356
lines changed

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

+30-25
Original file line numberDiff line numberDiff line change
@@ -16,30 +16,30 @@
1616

1717
package com.hedera.block.server;
1818

19+
import static com.hedera.block.protos.BlockStreamService.BlockItem;
20+
import static com.hedera.block.protos.BlockStreamService.PublishStreamRequest;
21+
import static com.hedera.block.protos.BlockStreamService.PublishStreamResponse;
22+
import static com.hedera.block.protos.BlockStreamService.SubscribeStreamRequest;
23+
import static com.hedera.block.server.Constants.*;
24+
1925
import com.google.protobuf.Descriptors;
2026
import com.hedera.block.server.consumer.ConsumerBlockItemObserver;
2127
import com.hedera.block.server.data.ObjectEvent;
2228
import com.hedera.block.server.mediator.StreamMediator;
2329
import com.hedera.block.server.producer.ProducerBlockItemObserver;
2430
import io.grpc.stub.StreamObserver;
2531
import io.helidon.webserver.grpc.GrpcService;
26-
2732
import java.time.Clock;
2833

29-
import static com.hedera.block.server.Constants.*;
30-
import static com.hedera.block.protos.BlockStreamService.PublishStreamRequest;
31-
import static com.hedera.block.protos.BlockStreamService.PublishStreamResponse;
32-
import static com.hedera.block.protos.BlockStreamService.SubscribeStreamRequest;
33-
import static com.hedera.block.protos.BlockStreamService.SubscribeStreamResponse;
34-
import static com.hedera.block.protos.BlockStreamService.BlockItem;
35-
3634
/**
37-
* This class implements the GrpcService interface and provides the functionality for the BlockStreamService.
38-
* It sets up the bidirectional streaming methods for the service and handles the routing for these methods.
39-
* It also initializes the StreamMediator, BlockStorage, and BlockCache upon creation.
35+
* This class implements the GrpcService interface and provides the functionality for the
36+
* BlockStreamService. It sets up the bidirectional streaming methods for the service and handles
37+
* the routing for these methods. It also initializes the StreamMediator, BlockStorage, and
38+
* BlockCache upon creation.
4039
*
41-
* <p>The class provides two main methods, streamSink and streamSource, which handle the client and server streaming
42-
* respectively. These methods return custom StreamObservers which are used to observe and respond to the streams.
40+
* <p>The class provides two main methods, streamSink and streamSource, which handle the client and
41+
* server streaming respectively. These methods return custom StreamObservers which are used to
42+
* observe and respond to the streams.
4343
*/
4444
public class BlockStreamService implements GrpcService {
4545

@@ -53,8 +53,9 @@ public class BlockStreamService implements GrpcService {
5353
*
5454
* @param timeoutThresholdMillis the timeout threshold in milliseconds
5555
*/
56-
public BlockStreamService(final long timeoutThresholdMillis,
57-
final StreamMediator<ObjectEvent<BlockItem>, SubscribeStreamRequest> streamMediator) {
56+
public BlockStreamService(
57+
final long timeoutThresholdMillis,
58+
final StreamMediator<ObjectEvent<BlockItem>, SubscribeStreamRequest> streamMediator) {
5859
this.timeoutThresholdMillis = timeoutThresholdMillis;
5960
this.streamMediator = streamMediator;
6061
}
@@ -70,8 +71,8 @@ public Descriptors.FileDescriptor proto() {
7071
}
7172

7273
/**
73-
* Returns the service name for the BlockStreamService. This service name corresponds to the service name in
74-
* the proto file.
74+
* Returns the service name for the BlockStreamService. This service name corresponds to the
75+
* service name in the proto file.
7576
*
7677
* @return the service name corresponding to the service name in the proto file
7778
*/
@@ -81,7 +82,8 @@ public String serviceName() {
8182
}
8283

8384
/**
84-
* Updates the routing for the BlockStreamService. It sets up the bidirectional streaming methods for the service.
85+
* Updates the routing for the BlockStreamService. It sets up the bidirectional streaming
86+
* methods for the service.
8587
*
8688
* @param routing the routing for the BlockStreamService
8789
*/
@@ -91,21 +93,24 @@ public void update(final Routing routing) {
9193
routing.bidi(SERVER_STREAMING_METHOD_NAME, this::subscribeBlockStream);
9294
}
9395

94-
private StreamObserver<PublishStreamRequest> publishBlockStream(final StreamObserver<PublishStreamResponse> publishStreamResponseObserver) {
96+
private StreamObserver<PublishStreamRequest> publishBlockStream(
97+
final StreamObserver<PublishStreamResponse> publishStreamResponseObserver) {
9598
LOGGER.log(System.Logger.Level.DEBUG, "Executing bidirectional streamSink method");
9699

97100
return new ProducerBlockItemObserver(streamMediator, publishStreamResponseObserver);
98101
}
99102

100-
private StreamObserver<SubscribeStreamRequest> subscribeBlockStream(final StreamObserver<BlockItem> subscribeStreamRequestObserver) {
103+
private StreamObserver<SubscribeStreamRequest> subscribeBlockStream(
104+
final StreamObserver<BlockItem> subscribeStreamRequestObserver) {
101105
LOGGER.log(System.Logger.Level.DEBUG, "Executing bidirectional streamSource method");
102106

103107
// Return a custom StreamObserver to handle streaming blocks from the producer.
104-
final var streamObserver = new ConsumerBlockItemObserver(
105-
timeoutThresholdMillis,
106-
Clock.systemDefaultZone(),
107-
streamMediator,
108-
subscribeStreamRequestObserver);
108+
final var streamObserver =
109+
new ConsumerBlockItemObserver(
110+
timeoutThresholdMillis,
111+
Clock.systemDefaultZone(),
112+
streamMediator,
113+
subscribeStreamRequestObserver);
109114

110115
streamMediator.subscribe(streamObserver);
111116

‎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 = "BlockStreamGrpcService";

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

+45-31
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,13 @@
1616

1717
package com.hedera.block.server;
1818

19+
import static com.hedera.block.protos.BlockStreamService.BlockItem;
20+
import static com.hedera.block.protos.BlockStreamService.PublishStreamRequest;
21+
import static com.hedera.block.protos.BlockStreamService.PublishStreamResponse;
22+
import static com.hedera.block.protos.BlockStreamService.SubscribeStreamRequest;
23+
import static com.hedera.block.protos.BlockStreamService.SubscribeStreamResponse;
24+
import static com.hedera.block.server.Constants.*;
25+
1926
import com.hedera.block.server.mediator.LiveStreamMediatorImpl;
2027
import com.hedera.block.server.persistence.WriteThroughCacheHandler;
2128
import com.hedera.block.server.persistence.storage.BlockStorage;
@@ -25,28 +32,24 @@
2532
import io.helidon.config.Config;
2633
import io.helidon.webserver.WebServer;
2734
import io.helidon.webserver.grpc.GrpcRouting;
28-
2935
import java.io.IOException;
3036

31-
import static com.hedera.block.server.Constants.*;
32-
import static com.hedera.block.protos.BlockStreamService.BlockItem;
33-
import static com.hedera.block.protos.BlockStreamService.PublishStreamRequest;
34-
import static com.hedera.block.protos.BlockStreamService.PublishStreamResponse;
35-
import static com.hedera.block.protos.BlockStreamService.SubscribeStreamRequest;
36-
import static com.hedera.block.protos.BlockStreamService.SubscribeStreamResponse;
37-
38-
/**
39-
* Main class for the block node server
40-
*/
37+
/** Main class for the block node server */
4138
public class Server {
4239

43-
// public interface BidiStreamingMethod<ReqT, RespT> extends ServerCalls.StreamingRequestMethod<ReqT, RespT> {
44-
// @Override StreamObserver<ReqT> invoke(StreamObserver<RespT> responseObserver);
45-
// }
40+
// public interface BidiStreamingMethod<ReqT, RespT> extends
41+
// ServerCalls.StreamingRequestMethod<ReqT, RespT> {
42+
// @Override StreamObserver<ReqT> invoke(StreamObserver<RespT> responseObserver);
43+
// }
4644

47-
// Function stubs to satisfy the bidi routing param signatures. The implementations are in the service class.
48-
private static ServerCalls.BidiStreamingMethod<StreamObserver<PublishStreamRequest>, StreamObserver<PublishStreamResponse>> clientBidiStreamingMethod;
49-
private static ServerCalls.BidiStreamingMethod<StreamObserver<SubscribeStreamRequest>, StreamObserver<SubscribeStreamResponse>> serverBidiStreamingMethod;
45+
// Function stubs to satisfy the bidi routing param signatures. The implementations are in the
46+
// service class.
47+
private static ServerCalls.BidiStreamingMethod<
48+
StreamObserver<PublishStreamRequest>, StreamObserver<PublishStreamResponse>>
49+
clientBidiStreamingMethod;
50+
private static ServerCalls.BidiStreamingMethod<
51+
StreamObserver<SubscribeStreamRequest>, StreamObserver<SubscribeStreamResponse>>
52+
serverBidiStreamingMethod;
5053

5154
private static final System.Logger LOGGER = System.getLogger(Server.class.getName());
5255

@@ -66,26 +69,37 @@ public static void main(final String[] args) {
6669
Config.global(config);
6770

6871
// Get Timeout threshold from configuration
69-
final long consumerTimeoutThreshold = config.get(BLOCKNODE_SERVER_CONSUMER_TIMEOUT_THRESHOLD_KEY).asLong().orElse(1500L);
72+
final long consumerTimeoutThreshold =
73+
config.get(BLOCKNODE_SERVER_CONSUMER_TIMEOUT_THRESHOLD_KEY)
74+
.asLong()
75+
.orElse(1500L);
7076

7177
// Initialize the block storage, cache, and service
72-
final BlockStorage<BlockItem> blockStorage = new FileSystemBlockStorage(BLOCKNODE_STORAGE_ROOT_PATH_KEY, config);
73-
final BlockStreamService blockStreamService = new BlockStreamService(consumerTimeoutThreshold,
74-
new LiveStreamMediatorImpl(new WriteThroughCacheHandler(blockStorage)));
78+
final BlockStorage<BlockItem> blockStorage =
79+
new FileSystemBlockStorage(BLOCKNODE_STORAGE_ROOT_PATH_KEY, config);
80+
final BlockStreamService blockStreamService =
81+
new BlockStreamService(
82+
consumerTimeoutThreshold,
83+
new LiveStreamMediatorImpl(new WriteThroughCacheHandler(blockStorage)));
7584

7685
// Start the web server
7786
WebServer.builder()
7887
.port(8080)
79-
.addRouting(GrpcRouting.builder()
80-
.service(blockStreamService)
81-
.bidi(com.hedera.block.protos.BlockStreamService.getDescriptor(),
82-
SERVICE_NAME,
83-
CLIENT_STREAMING_METHOD_NAME,
84-
clientBidiStreamingMethod)
85-
.bidi(com.hedera.block.protos.BlockStreamService.getDescriptor(),
86-
SERVICE_NAME,
87-
SERVER_STREAMING_METHOD_NAME,
88-
serverBidiStreamingMethod))
88+
.addRouting(
89+
GrpcRouting.builder()
90+
.service(blockStreamService)
91+
.bidi(
92+
com.hedera.block.protos.BlockStreamService
93+
.getDescriptor(),
94+
SERVICE_NAME,
95+
CLIENT_STREAMING_METHOD_NAME,
96+
clientBidiStreamingMethod)
97+
.bidi(
98+
com.hedera.block.protos.BlockStreamService
99+
.getDescriptor(),
100+
SERVICE_NAME,
101+
SERVER_STREAMING_METHOD_NAME,
102+
serverBidiStreamingMethod))
89103
.build()
90104
.start();
91105

‎server/src/main/java/com/hedera/block/server/consumer/BlockItemEventHandler.java

+1
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@
1313
* See the License for the specific language governing permissions and
1414
* limitations under the License.
1515
*/
16+
1617
package com.hedera.block.server.consumer;
1718

1819
import com.lmax.disruptor.EventHandler;

‎server/src/main/java/com/hedera/block/server/consumer/ConsumerBlockItemObserver.java

+41-27
Original file line numberDiff line numberDiff line change
@@ -16,23 +16,24 @@
1616

1717
package com.hedera.block.server.consumer;
1818

19+
import static com.hedera.block.protos.BlockStreamService.BlockItem;
20+
import static com.hedera.block.protos.BlockStreamService.SubscribeStreamRequest;
21+
1922
import com.hedera.block.server.data.ObjectEvent;
2023
import com.hedera.block.server.mediator.StreamMediator;
2124
import io.grpc.stub.ServerCallStreamObserver;
2225
import io.grpc.stub.StreamObserver;
23-
2426
import java.time.Clock;
2527
import java.time.InstantSource;
2628
import java.util.concurrent.CountDownLatch;
2729

28-
import static com.hedera.block.protos.BlockStreamService.BlockItem;
29-
import static com.hedera.block.protos.BlockStreamService.SubscribeStreamRequest;
30-
3130
/**
32-
* The LiveStreamObserverImpl class implements the LiveStreamObserver interface to pass blocks to the downstream consumer
33-
* via the notify method and manage the bidirectional stream to the consumer via the onNext, onError, and onCompleted methods.
31+
* The LiveStreamObserverImpl class implements the LiveStreamObserver interface to pass blocks to
32+
* the downstream consumer via the notify method and manage the bidirectional stream to the consumer
33+
* via the onNext, onError, and onCompleted methods.
3434
*/
35-
public class ConsumerBlockItemObserver implements BlockItemEventHandler<ObjectEvent<BlockItem>, SubscribeStreamRequest> {
35+
public class ConsumerBlockItemObserver
36+
implements BlockItemEventHandler<ObjectEvent<BlockItem>, SubscribeStreamRequest> {
3637

3738
private final System.Logger LOGGER = System.getLogger(getClass().getName());
3839

@@ -68,13 +69,17 @@ public ConsumerBlockItemObserver(
6869
// this observer to avoid orphaning subscribed resources.
6970
if (subscribeStreamResponseObserver instanceof ServerCallStreamObserver) {
7071

71-
// Unfortunately we have to cast the responseStreamObserver to a ServerCallStreamObserver
72+
// Unfortunately we have to cast the responseStreamObserver to a
73+
// ServerCallStreamObserver
7274
// to register the onCancelHandler.
73-
((ServerCallStreamObserver<BlockItem>)subscribeStreamResponseObserver)
74-
.setOnCancelHandler(() -> {
75-
LOGGER.log(System.Logger.Level.DEBUG, "Consumer cancelled stream. Unsubscribing observer.");
76-
streamMediator.unsubscribe(this);
77-
});
75+
((ServerCallStreamObserver<BlockItem>) subscribeStreamResponseObserver)
76+
.setOnCancelHandler(
77+
() -> {
78+
LOGGER.log(
79+
System.Logger.Level.DEBUG,
80+
"Consumer cancelled stream. Unsubscribing observer.");
81+
streamMediator.unsubscribe(this);
82+
});
7883
}
7984

8085
this.subscribeStreamResponseObserver = subscribeStreamResponseObserver;
@@ -83,12 +88,10 @@ public ConsumerBlockItemObserver(
8388
this.streamMediator = streamMediator;
8489
}
8590

86-
/**
87-
* Pass the block to the observer provided by Helidon
88-
*
89-
*/
91+
/** Pass the block to the observer provided by Helidon */
9092
@Override
91-
public void onEvent(final ObjectEvent<BlockItem> event, final long l, final boolean b) throws Exception {
93+
public void onEvent(final ObjectEvent<BlockItem> event, final long l, final boolean b)
94+
throws Exception {
9295

9396
// Refresh the producer liveness and pass the block to the observer.
9497
producerLivenessMillis = producerLivenessClock.millis();
@@ -104,33 +107,39 @@ public void onEvent(final ObjectEvent<BlockItem> event, final long l, final bool
104107
}
105108

106109
/**
107-
* The onNext() method is triggered by Helidon when a consumer sends a blockResponse via the bidirectional stream.
108-
*
110+
* The onNext() method is triggered by Helidon when a consumer sends a blockResponse via the
111+
* bidirectional stream.
109112
*/
110113
@Override
111114
public void onNext(final SubscribeStreamRequest subscribeStreamRequest) {
112115

113116
// Check if the producer has timed out. If so, unsubscribe the observer from the mediator.
114117
if (isThresholdExceeded(producerLivenessMillis)) {
115-
LOGGER.log(System.Logger.Level.DEBUG, "Producer timeout threshold exceeded. Unsubscribing observer.");
118+
LOGGER.log(
119+
System.Logger.Level.DEBUG,
120+
"Producer timeout threshold exceeded. Unsubscribing observer.");
116121
streamMediator.unsubscribe(this);
117122
}
118123
}
119124

120125
/**
121-
* The onError() method is triggered by Helidon when an error occurs on the bidirectional stream to the downstream consumer.
122-
* Unsubscribe the observer from the mediator.
126+
* The onError() method is triggered by Helidon when an error occurs on the bidirectional stream
127+
* to the downstream consumer. Unsubscribe the observer from the mediator.
123128
*
124129
* @param t the error occurred on the stream
125130
*/
126131
@Override
127132
public void onError(final Throwable t) {
128-
LOGGER.log(System.Logger.Level.ERROR, "Unexpected consumer stream communication failure: %s".formatted(t), t);
133+
LOGGER.log(
134+
System.Logger.Level.ERROR,
135+
"Unexpected consumer stream communication failure: %s".formatted(t),
136+
t);
129137
}
130138

131139
/**
132-
* The onCompleted() method is triggered by Helidon when the bidirectional stream to the downstream consumer is completed.
133-
* This implementation will then unsubscribe the observer from the mediator.
140+
* The onCompleted() method is triggered by Helidon when the bidirectional stream to the
141+
* downstream consumer is completed. This implementation will then unsubscribe the observer from
142+
* the mediator.
134143
*/
135144
@Override
136145
public void onCompleted() {
@@ -142,7 +151,12 @@ private boolean isThresholdExceeded(long livenessMillis) {
142151
final long currentTimeMillis = Clock.systemDefaultZone().millis();
143152
final long elapsedMillis = currentTimeMillis - livenessMillis;
144153
if (elapsedMillis > timeoutThresholdMillis) {
145-
LOGGER.log(System.Logger.Level.INFO, "Elapsed milliseconds: " + elapsedMillis + ", timeout threshold: " + timeoutThresholdMillis);
154+
LOGGER.log(
155+
System.Logger.Level.INFO,
156+
"Elapsed milliseconds: "
157+
+ elapsedMillis
158+
+ ", timeout threshold: "
159+
+ timeoutThresholdMillis);
146160
return true;
147161
}
148162

‎server/src/main/java/com/hedera/block/server/data/ObjectEvent.java

+2-2
Original file line numberDiff line numberDiff line change
@@ -13,10 +13,10 @@
1313
* See the License for the specific language governing permissions and
1414
* limitations under the License.
1515
*/
16+
1617
package com.hedera.block.server.data;
1718

18-
public class ObjectEvent<T>
19-
{
19+
public class ObjectEvent<T> {
2020
T val;
2121

2222
public void clear() {

0 commit comments

Comments
 (0)