fix: Publisher API is missing some request forms

* Added the missing `EndStream` item for publish request
   * Note, this does _not_ implement handling for this new message; it only adds the API definitions.
* Moved publish end stream code and response end stream code into their respective messages
* Removed `Acknowledgement` wrapper around `BlockAcknowledgement`.
* Removed unnecessary enum name prefixes
* Discovered another PBJ bug (unit tests generation) related to message name conflicts
* Fixed all references in code and tests

Fixes #1151

Signed-off-by: Joseph S <[email protected]>

Author: jsync-swirlds

block-node/publisher/src/main/java/org/hiero/block/node/publisher/BlockStreamProducerSession.java

@@ -20,13 +20,12 @@
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Consumer;
 import org.hiero.block.api.PublishStreamResponse;
-import org.hiero.block.api.PublishStreamResponse.Acknowledgement;
 import org.hiero.block.api.PublishStreamResponse.BlockAcknowledgement;
 import org.hiero.block.api.PublishStreamResponse.EndOfStream;
+import org.hiero.block.api.PublishStreamResponse.EndOfStream.Code;
 import org.hiero.block.api.PublishStreamResponse.ResendBlock;
 import org.hiero.block.api.PublishStreamResponse.ResponseOneOfType;
 import org.hiero.block.api.PublishStreamResponse.SkipBlock;
-import org.hiero.block.api.PublishStreamResponseCode;
 import org.hiero.block.internal.BlockItemUnparsed;
 import org.hiero.block.node.publisher.UpdateCallback.UpdateType;
 import org.hiero.block.node.spi.BlockNodePlugin;
@@ -201,9 +200,8 @@ void sendDuplicateAck(final long latestAckBlock) {
         // sending a duplicate ack should also update the latestAck.
         latestAcknowledgedBlock = latestAckBlock;
         final BlockAcknowledgement ack = new BlockAcknowledgement(latestAckBlock, null, true);
-        final Acknowledgement acknowledgement = new Acknowledgement(ack);
         final PublishStreamResponse duplicateResponse =
-                new PublishStreamResponse(new OneOf<>(ResponseOneOfType.ACKNOWLEDGEMENT, acknowledgement));
+                new PublishStreamResponse(new OneOf<>(ResponseOneOfType.ACKNOWLEDGEMENT, ack));
         sendResponse(duplicateResponse);
     }
 
@@ -215,7 +213,7 @@ void sendStreamItemsBehind(final long latestAckBlock) {
         currentBlockState = BlockState.WAITING_FOR_RESEND;
         newBlockItems.clear();
 
-        final EndOfStream endOfStream = new EndOfStream(PublishStreamResponseCode.STREAM_ITEMS_BEHIND, latestAckBlock);
+        final EndOfStream endOfStream = new EndOfStream(Code.BEHIND, latestAckBlock);
         final PublishStreamResponse response =
                 new PublishStreamResponse(new OneOf<>(ResponseOneOfType.END_STREAM, endOfStream));
         sendResponse(response);
@@ -247,9 +245,8 @@ void close() {
             currentBlockState = BlockState.DISCONNECTED;
             // try to send a close response to the client
 
-            final PublishStreamResponse closeResponse = new PublishStreamResponse(new OneOf<>(
-                    ResponseOneOfType.END_STREAM,
-                    new EndOfStream(PublishStreamResponseCode.STREAM_ITEMS_SUCCESS, currentBlockNumber)));
+            final PublishStreamResponse closeResponse = new PublishStreamResponse(
+                    new OneOf<>(ResponseOneOfType.END_STREAM, new EndOfStream(Code.SUCCESS, currentBlockNumber)));
             sendResponse(closeResponse);
 
             if (subscription != null) {
@@ -286,8 +283,7 @@ void handlePersisted(PersistedNotification notification) {
                 latestAcknowledgedBlock = blockToSend;
                 // TODO BlockAcknowledgement block hash should be removed from spec as not needed
                 final PublishStreamResponse goodBlockResponse = new PublishStreamResponse(new OneOf<>(
-                        ResponseOneOfType.ACKNOWLEDGEMENT,
-                        new Acknowledgement(new BlockAcknowledgement(blockToSend, null, false))));
+                        ResponseOneOfType.ACKNOWLEDGEMENT, new BlockAcknowledgement(blockToSend, null, false)));
                 // send the acknowledgment to the client
                 sendResponse(goodBlockResponse);
                 blockToSend++;

block-node/publisher/src/main/java/org/hiero/block/node/publisher/PublisherServicePlugin.java

@@ -96,7 +96,7 @@ public final class PublisherServicePlugin implements BlockNodePlugin, ServiceInt
     /** The current ACKED block number */
     private long latestAckedBlockNumber = UNKNOWN_BLOCK_NUMBER;
     /** The current chosen primary consensus node session, or null if there is no primary */
-    private BlockStreamProducerSession currentPrimarySession = null;
+    private BlockStreamProducerSession currentPrimarySession;
     /** The next session id to use when a new session is created */
     private long nextSessionId = 0;
 

block-node/publisher/src/test/java/org/hiero/block/node/publisher/PublisherTest.java

@@ -24,9 +24,8 @@
 import org.hiero.block.api.PublishStreamRequest;
 import org.hiero.block.api.PublishStreamRequest.RequestOneOfType;
 import org.hiero.block.api.PublishStreamResponse;
-import org.hiero.block.api.PublishStreamResponse.Acknowledgement;
 import org.hiero.block.api.PublishStreamResponse.BlockAcknowledgement;
-import org.hiero.block.api.PublishStreamResponseCode;
+import org.hiero.block.api.PublishStreamResponse.EndOfStream.Code;
 import org.hiero.block.internal.BlockUnparsed;
 import org.hiero.block.node.app.fixtures.plugintest.GrpcPluginTestBase;
 import org.hiero.block.node.app.fixtures.plugintest.NoBlocksHistoricalBlockFacility;
@@ -108,14 +107,14 @@ void testPublisherSendsOnBlockPersistedNotification() throws ParseException {
         assertEquals(1, fromPluginBytes.size());
         PublishStreamResponse response = PublishStreamResponse.PROTOBUF.parse(fromPluginBytes.getFirst());
         assertEquals(ACKNOWLEDGEMENT, response.response().kind());
-        final Acknowledgement ack = response.response().as();
-        assertEquals(new BlockAcknowledgement(0, null, false), ack.blockAck());
+        final BlockAcknowledgement ack = response.response().as();
+        assertEquals(new BlockAcknowledgement(0, null, false), ack);
         // check second block
         blockMessaging.sendBlockPersisted(new PersistedNotification(1, 1, 1));
         PublishStreamResponse response1 = PublishStreamResponse.PROTOBUF.parse(fromPluginBytes.getLast());
         assertEquals(ACKNOWLEDGEMENT, response1.response().kind());
-        final Acknowledgement ack1 = response1.response().as();
-        assertEquals(new BlockAcknowledgement(1, null, false), ack1.blockAck());
+        final BlockAcknowledgement ack1 = response1.response().as();
+        assertEquals(new BlockAcknowledgement(1, null, false), ack1);
     }
 
     @Test
@@ -142,7 +141,7 @@ void testPublisherDuplicateBlock() throws ParseException {
         // last fromPluginBytes should be Duplicate
         PublishStreamResponse response = PublishStreamResponse.PROTOBUF.parse(fromPluginBytes.getLast());
         assertEquals(ACKNOWLEDGEMENT, response.response().kind());
-        BlockAcknowledgement ack = response.acknowledgement().blockAck();
+        BlockAcknowledgement ack = response.acknowledgement();
         BlockAcknowledgement expectedAck = new BlockAcknowledgement(0, null, true);
         assertEquals(expectedAck, ack, "Expected an ACK with flag of duplicate (blockAlreadyExists=true)");
     }
@@ -171,10 +170,8 @@ void testFutureBlock() throws ParseException {
         assertEquals(2, fromPluginBytes.size());
         PublishStreamResponse aheadResponse = PublishStreamResponse.PROTOBUF.parse(fromPluginBytes.getLast());
 
-        // verify we get an endOfStream with status code STREAM_ITEMS_BEHIND
-        assertEquals(
-                PublishStreamResponseCode.STREAM_ITEMS_BEHIND,
-                aheadResponse.endStream().status());
+        // verify we get an endOfStream with status code BEHIND
+        assertEquals(Code.BEHIND, aheadResponse.endStream().status());
         assertEquals(0, aheadResponse.endStream().blockNumber());
     }