feat: Convert internal interfaces to use protos (#1335)

* feat: Convert internal interfaces to use protos

* feat: Convert internal interfaces to use protos

Author: dpcollins-google

google-cloud-pubsublite/src/main/java/com/google/cloud/pubsublite/cloudpubsub/internal/ResettableSubscriberFactory.java

@@ -17,9 +17,9 @@
 package com.google.cloud.pubsublite.cloudpubsub.internal;
 
 import com.google.api.gax.rpc.ApiException;
-import com.google.cloud.pubsublite.SequencedMessage;
 import com.google.cloud.pubsublite.internal.wire.Subscriber;
 import com.google.cloud.pubsublite.internal.wire.SubscriberResetHandler;
+import com.google.cloud.pubsublite.proto.SequencedMessage;
 import java.io.Serializable;
 import java.util.List;
 import java.util.function.Consumer;

google-cloud-pubsublite/src/main/java/com/google/cloud/pubsublite/cloudpubsub/internal/SinglePartitionSubscriber.java

@@ -83,9 +83,10 @@ private boolean terminated() {
   }
 
   @VisibleForTesting
-  void onMessages(List<SequencedMessage> sequencedMessages) {
+  void onMessages(List<com.google.cloud.pubsublite.proto.SequencedMessage> sequencedMessages) {
     try {
-      for (SequencedMessage message : sequencedMessages) {
+      for (com.google.cloud.pubsublite.proto.SequencedMessage proto : sequencedMessages) {
+        SequencedMessage message = SequencedMessage.fromProto(proto);
         PubsubMessage userMessage = transformer.transform(message);
         long bytes = message.byteSize();
         Runnable trackerConsumer = ackSetTracker.track(message);

google-cloud-pubsublite/src/main/java/com/google/cloud/pubsublite/cloudpubsub/internal/WrappingPublisher.java

@@ -73,7 +73,7 @@ public ApiFuture<String> publish(PubsubMessage message) {
       return ApiFutures.immediateFailedFuture(e.underlying);
     }
     return ApiFutures.transform(
-        wirePublisher.publish(wireMessage),
+        wirePublisher.publish(wireMessage.toProto()),
         MessageMetadata::encode,
         MoreExecutors.directExecutor());
   }

google-cloud-pubsublite/src/main/java/com/google/cloud/pubsublite/internal/BlockingPullSubscriber.java

@@ -17,7 +17,7 @@
 package com.google.cloud.pubsublite.internal;
 
 import com.google.api.core.ApiFuture;
-import com.google.cloud.pubsublite.SequencedMessage;
+import com.google.cloud.pubsublite.proto.SequencedMessage;
 import java.util.Optional;
 import javax.annotation.concurrent.ThreadSafe;
 

google-cloud-pubsublite/src/main/java/com/google/cloud/pubsublite/internal/BlockingPullSubscriberImpl.java

@@ -24,12 +24,12 @@
 import com.google.api.core.ApiService.State;
 import com.google.api.core.SettableApiFuture;
 import com.google.api.gax.rpc.StatusCode;
-import com.google.cloud.pubsublite.SequencedMessage;
 import com.google.cloud.pubsublite.cloudpubsub.FlowControlSettings;
 import com.google.cloud.pubsublite.internal.wire.Subscriber;
 import com.google.cloud.pubsublite.internal.wire.SubscriberFactory;
 import com.google.cloud.pubsublite.internal.wire.SystemExecutors;
 import com.google.cloud.pubsublite.proto.FlowControlRequest;
+import com.google.cloud.pubsublite.proto.SequencedMessage;
 import com.google.errorprone.annotations.concurrent.GuardedBy;
 import java.util.ArrayDeque;
 import java.util.Collection;
@@ -110,7 +110,7 @@ public synchronized Optional<SequencedMessage> messageIfAvailable() throws Check
     underlying.allowFlow(
         FlowControlRequest.newBuilder()
             .setAllowedMessages(1)
-            .setAllowedBytes(msg.byteSize())
+            .setAllowedBytes(msg.getSizeBytes())
             .build());
     return Optional.of(msg);
   }

google-cloud-pubsublite/src/main/java/com/google/cloud/pubsublite/internal/BufferingPullSubscriber.java

@@ -21,12 +21,12 @@
 import com.google.api.core.ApiService.Listener;
 import com.google.api.core.ApiService.State;
 import com.google.cloud.pubsublite.Offset;
-import com.google.cloud.pubsublite.SequencedMessage;
 import com.google.cloud.pubsublite.cloudpubsub.FlowControlSettings;
 import com.google.cloud.pubsublite.internal.wire.Subscriber;
 import com.google.cloud.pubsublite.internal.wire.SubscriberFactory;
 import com.google.cloud.pubsublite.internal.wire.SystemExecutors;
 import com.google.cloud.pubsublite.proto.FlowControlRequest;
+import com.google.cloud.pubsublite.proto.SequencedMessage;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterables;
 import com.google.errorprone.annotations.concurrent.GuardedBy;
@@ -85,13 +85,13 @@ public synchronized List<SequencedMessage> pull() throws CheckedApiException {
     }
     Deque<SequencedMessage> collection = messages;
     messages = new ArrayDeque<>();
-    long bytes = collection.stream().mapToLong(SequencedMessage::byteSize).sum();
+    long bytes = collection.stream().mapToLong(SequencedMessage::getSizeBytes).sum();
     underlying.allowFlow(
         FlowControlRequest.newBuilder()
             .setAllowedBytes(bytes)
             .setAllowedMessages(collection.size())
             .build());
-    lastDelivered = Optional.of(Iterables.getLast(collection).offset());
+    lastDelivered = Optional.of(Offset.of(Iterables.getLast(collection).getCursor().getOffset()));
     return ImmutableList.copyOf(collection);
   }
 

google-cloud-pubsublite/src/main/java/com/google/cloud/pubsublite/internal/Publisher.java

@@ -18,7 +18,7 @@
 
 import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiService;
-import com.google.cloud.pubsublite.Message;
+import com.google.cloud.pubsublite.proto.PubSubMessage;
 import java.io.Flushable;
 
 /** A generic PubSub Lite publisher. Errors are handled out of band. Thread safe. */
@@ -28,7 +28,7 @@ public interface Publisher<ResponseT> extends ApiService, Flushable {
   //
   // Guarantees that if a single publish future has an exception set, all publish calls made after
   // that will also have an exception set.
-  ApiFuture<ResponseT> publish(Message message);
+  ApiFuture<ResponseT> publish(PubSubMessage message);
 
   // Attempts to cancel all outstanding publishes.
   void cancelOutstandingPublishes();

google-cloud-pubsublite/src/main/java/com/google/cloud/pubsublite/internal/SequencedPublisher.java

@@ -18,7 +18,7 @@
 
 import com.google.api.core.ApiFuture;
 import com.google.api.core.ApiService;
-import com.google.cloud.pubsublite.Message;
+import com.google.cloud.pubsublite.proto.PubSubMessage;
 import java.io.Flushable;
 
 /**
@@ -35,7 +35,7 @@ public interface SequencedPublisher<ResponseT> extends ApiService, Flushable {
    * <p>Guarantees that if a single publish future has an exception set, all publish calls made
    * after that will also have an exception set.
    */
-  ApiFuture<ResponseT> publish(Message message, PublishSequenceNumber sequenceNumber);
+  ApiFuture<ResponseT> publish(PubSubMessage message, PublishSequenceNumber sequenceNumber);
 
   /** Attempts to cancel all outstanding publishes. */
   void cancelOutstandingPublishes();

google-cloud-pubsublite/src/main/java/com/google/cloud/pubsublite/internal/wire/ApiExceptionPublisher.java

@@ -20,9 +20,9 @@
 
 import com.google.api.core.ApiFuture;
 import com.google.api.gax.rpc.ApiException;
-import com.google.cloud.pubsublite.Message;
 import com.google.cloud.pubsublite.internal.ProxyService;
 import com.google.cloud.pubsublite.internal.Publisher;
+import com.google.cloud.pubsublite.proto.PubSubMessage;
 import java.io.IOException;
 
 public class ApiExceptionPublisher<T> extends ProxyService implements Publisher<T> {
@@ -34,7 +34,7 @@ public class ApiExceptionPublisher<T> extends ProxyService implements Publisher<
   }
 
   @Override
-  public ApiFuture<T> publish(Message message) {
+  public ApiFuture<T> publish(PubSubMessage message) {
     return toClientFuture(publisher.publish(message));
   }
 

google-cloud-pubsublite/src/main/java/com/google/cloud/pubsublite/internal/wire/ConnectedSubscriberFactory.java

@@ -16,7 +16,7 @@
 
 package com.google.cloud.pubsublite.internal.wire;
 
-import com.google.cloud.pubsublite.SequencedMessage;
+import com.google.cloud.pubsublite.proto.SequencedMessage;
 import com.google.cloud.pubsublite.proto.SubscribeRequest;
 import com.google.cloud.pubsublite.proto.SubscribeResponse;
 import java.util.List;