[SOLVED] PubSub Emulator - ( Support Proto Buffer publish/receive msg)

I am developing a solution to use a common Proto Buffer library to send and receive msg using directly proto buffer serialized (ByteString) and deserialization from a (ByteString) directly into the same Proto Buffer Class. My solution until now it is not working. Just when I use a real PubSub.

Based on The doc: Testing apps locally with the emulator information and more specific in the section knowing limitations:

Emulator doesn't provide Schema support for protocol buffers.

Although, I am not using any schema definition in Topic/Subscription. Just using a common proto buffer library programmatically. I'm afraid there is a Pubsub emulation limitation and for this reason my solution doesn't work with the Emulator.

Bellow my Test Class any clarification we be very welcome.

package com.example.pubsubgcpspringapplications;


import java.io.IOException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import com.alpian.common.pubsub.messages.OnfidoVerificationEvent;
import com.example.pubsubgcpspringapplications.config.PubSubTestConfig;
import com.example.pubsubgcpspringapplications.services.MessageRealGcpService;
import com.example.pubsubgcpspringapplications.util.DataGenerationUtils;
import com.google.api.core.ApiFuture;
import com.google.cloud.pubsub.v1.AckReplyConsumer;
import com.google.cloud.pubsub.v1.MessageReceiver;
import com.google.cloud.pubsub.v1.Publisher;
import com.google.cloud.pubsub.v1.Subscriber;
import com.google.protobuf.ByteString;
import com.google.protobuf.InvalidProtocolBufferException;
import com.google.protobuf.util.JsonFormat;
import com.google.pubsub.v1.ProjectSubscriptionName;
import com.google.pubsub.v1.PubsubMessage;
import lombok.SneakyThrows;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.ActiveProfiles;

//@ActiveProfiles("test")
public class EmulatorPubSubWithSpringTest {

  @BeforeAll
  static void startUpTests() throws IOException {
    PubSubTestConfig.setupPubSubEmulator();
  }

  @SneakyThrows
  @Test
  void successfulTest() throws InterruptedException {

    var status = DataGenerationUtils.STATUS_COMPLETE;
    var result = DataGenerationUtils.RESULT_CLEAR;
    var subResult = DataGenerationUtils.SUB_RESULT_CLEAR;

    var documentReport = DataGenerationUtils.generateOnfidoDocumentReport(status, result, subResult);
    var facialSimilarityReport = DataGenerationUtils
        .generateOnfidoFacialSimiliratyVideoReport(status, result, subResult);

    OnfidoVerificationEvent.Builder builder = OnfidoVerificationEvent.newBuilder();
    builder.setCheckId(DataGenerationUtils.FAKE_CHECK_ID);
    builder.setApplicantId(DataGenerationUtils.FAKE_APPLICANT_ID);
    builder.setDocument(documentReport);
    builder.setFacialSimilarityVideo(facialSimilarityReport);
    OnfidoVerificationEvent onfidoVerificationEvent = builder.build();

    publishProtoMessageTest(onfidoVerificationEvent);

    MessageReceiver receiver =
        (PubsubMessage message, AckReplyConsumer consumer) -> {
          ByteString data = message.getData();

          // Get the schema encoding type.
          String encoding = message.getAttributesMap().get("googclient_schemaencoding");

          block:
          try {
            switch (encoding) {
              case "BINARY":
                // Obtain an object of the generated proto class.
                OnfidoVerificationEvent state = OnfidoVerificationEvent.parseFrom(data);
                System.out.println("Received a BINARY-formatted message: " + state);
                break;

              case "JSON":
                OnfidoVerificationEvent.Builder stateBuilder = OnfidoVerificationEvent.newBuilder();
                JsonFormat.parser().merge(data.toStringUtf8(), stateBuilder);
                System.out.println("Received a JSON-formatted message:" + stateBuilder.build());
                break;

              default:
                break block;
            }
          } catch (InvalidProtocolBufferException e) {
            e.printStackTrace();
          }

          consumer.ack();
          System.out.println("Ack'ed the message");
        };

    ProjectSubscriptionName subscriptionName =
        ProjectSubscriptionName.of(PubSubTestConfig.PROJECT_ID, PubSubTestConfig.SUBSCRIPTION_NAME);

    // Create subscriber client.
    Subscriber subscriber = Subscriber.newBuilder(subscriptionName, receiver).build();

    try {
      subscriber.startAsync().awaitRunning();
      System.out.printf("Listening for messages on %s:\n", subscriptionName);
      subscriber.awaitTerminated(30, TimeUnit.SECONDS);
    } catch (TimeoutException timeoutException) {
      subscriber.stopAsync();
    }

    Thread.sleep(15000);

  }

  public static void publishProtoMessageTest(OnfidoVerificationEvent onfidoVerificationEvent)
      throws IOException, ExecutionException, InterruptedException {

    Publisher publisher = null;

    block:
    try {
      publisher = Publisher.newBuilder("projects/my-project-id/topics/topic-one").build();
      PubsubMessage.Builder message = PubsubMessage.newBuilder();
      // Prepare an appropriately formatted message based on topic encoding.
      message.setData(onfidoVerificationEvent.toByteString());
      System.out.println("Publishing a BINARY-formatted message:\n" + message);

      // Publish the message.
      ApiFuture<String> future = publisher.publish(message.build());
      //System.out.println("Published message ID: " + future.get());

    } finally {
      if (publisher != null) {
        publisher.shutdown();
        publisher.awaitTermination(1, TimeUnit.MINUTES);
      }
    }
  }


}

Note: Please, I just copied some sniped code from google tutorial and modified it. I don't want to use JSON just publish and receive msg using proto files.

Many Thanks in advance!