feat: Create ReadChangeStreamQuery and ChangeStreamRecode::Heartbeat/CloseStream

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamContinuationToken.java

@@ -0,0 +1,99 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.google.cloud.bigtable.data.v2.models;
+
+import com.google.bigtable.v2.RowRange;
+import com.google.bigtable.v2.StreamContinuationToken;
+import com.google.common.base.MoreObjects;
+import com.google.common.base.Objects;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.io.Serializable;
+import javax.annotation.Nonnull;
+
+/** A simple wrapper for {@link StreamContinuationToken}. */
+public final class ChangeStreamContinuationToken implements Serializable {
+  private static final long serialVersionUID = 524679926247095L;
+
+  private transient StreamContinuationToken.Builder builder;
+
+  private ChangeStreamContinuationToken(@Nonnull StreamContinuationToken.Builder builder) {
+    this.builder = builder;
+  }
+
+  private void readObject(ObjectInputStream input) throws IOException, ClassNotFoundException {
+    input.defaultReadObject();
+    builder = StreamContinuationToken.newBuilder().mergeFrom(input);
+  }
+
+  private void writeObject(ObjectOutputStream output) throws IOException {
+    output.defaultWriteObject();
+    builder.build().writeTo(output);
+  }
+
+  public RowRange getRowRange() {
+    return this.builder.getPartition().getRowRange();
+  }
+
+  public String getToken() {
+    return this.builder.getToken();
+  }
+
+  /**
+   * Creates the protobuf. This method is considered an internal implementation detail and not meant
+   * to be used by applications.
+   */
+  StreamContinuationToken toProto() {
+    return builder.build();
+  }
+
+  /** Wraps the protobuf {@link StreamContinuationToken}. */
+  static ChangeStreamContinuationToken fromProto(
+      @Nonnull StreamContinuationToken streamContinuationToken) {
+    return new ChangeStreamContinuationToken(streamContinuationToken.toBuilder());
+  }
+
+  public ChangeStreamContinuationToken clone() {
+    return new ChangeStreamContinuationToken(this.builder.clone());
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    ChangeStreamContinuationToken otherToken = (ChangeStreamContinuationToken) o;
+    return Objects.equal(getRowRange(), otherToken.getRowRange())
+        && Objects.equal(getToken(), otherToken.getToken());
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hashCode(getRowRange(), getToken());
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("rowRange", getRowRange())
+        .add("token", getToken())
+        .toString();
+  }
+}

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/ChangeStreamRecord.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.google.cloud.bigtable.data.v2.models;
+
+import com.google.api.core.InternalExtensionOnly;
+
+/**
+ * Default representation of a change stream record, which can be a Heartbeat, a CloseStream, or a
+ * logical mutation.
+ */
+@InternalExtensionOnly
+public interface ChangeStreamRecord {}

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/CloseStream.java

@@ -0,0 +1,102 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.google.cloud.bigtable.data.v2.models;
+
+import com.google.api.core.InternalApi;
+import com.google.bigtable.v2.ReadChangeStreamResponse;
+import com.google.bigtable.v2.StreamContinuationToken;
+import com.google.common.base.MoreObjects;
+import com.google.common.base.Objects;
+import com.google.common.collect.ImmutableList;
+import com.google.rpc.Status;
+import java.io.IOException;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutputStream;
+import java.io.Serializable;
+import java.util.List;
+import javax.annotation.Nonnull;
+
+public final class CloseStream implements ChangeStreamRecord, Serializable {
+  private static final long serialVersionUID = 7316215828353608505L;
+  private final Status status;
+  private transient ImmutableList.Builder<ChangeStreamContinuationToken>
+      changeStreamContinuationTokens = ImmutableList.builder();
+
+  private CloseStream(Status status, List<StreamContinuationToken> continuationTokens) {
+    this.status = status;
+    for (StreamContinuationToken streamContinuationToken : continuationTokens) {
+      changeStreamContinuationTokens.add(
+          ChangeStreamContinuationToken.fromProto(streamContinuationToken));
+    }
+  }
+
+  @InternalApi("Used in Changestream beam pipeline.")
+  public Status getStatus() {
+    return this.status;
+  }
+
+  @InternalApi("Used in Changestream beam pipeline.")
+  public List<ChangeStreamContinuationToken> getChangeStreamContinuationTokens() {
+    return changeStreamContinuationTokens.build();
+  }
+
+  private void readObject(ObjectInputStream input) throws IOException, ClassNotFoundException {
+    input.defaultReadObject();
+
+    @SuppressWarnings("unchecked")
+    ImmutableList<ChangeStreamContinuationToken> deserialized =
+        (ImmutableList<ChangeStreamContinuationToken>) input.readObject();
+    this.changeStreamContinuationTokens =
+        ImmutableList.<ChangeStreamContinuationToken>builder().addAll(deserialized);
+  }
+
+  private void writeObject(ObjectOutputStream output) throws IOException {
+    output.defaultWriteObject();
+    output.writeObject(changeStreamContinuationTokens.build());
+  }
+
+  /** Wraps the protobuf {@link ReadChangeStreamResponse.CloseStream}. */
+  static CloseStream fromProto(@Nonnull ReadChangeStreamResponse.CloseStream closeStream) {
+    return new CloseStream(closeStream.getStatus(), closeStream.getContinuationTokensList());
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    CloseStream record = (CloseStream) o;
+    return Objects.equal(status, record.getStatus())
+        && Objects.equal(
+            changeStreamContinuationTokens.build(), record.getChangeStreamContinuationTokens());
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hashCode(status, changeStreamContinuationTokens);
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("status", status)
+        .add("changeStreamContinuationTokens", changeStreamContinuationTokens)
+        .toString();
+  }
+}

google-cloud-bigtable/src/main/java/com/google/cloud/bigtable/data/v2/models/Heartbeat.java

@@ -0,0 +1,79 @@
+/*
+ * Copyright 2022 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package com.google.cloud.bigtable.data.v2.models;
+
+import com.google.api.core.InternalApi;
+import com.google.bigtable.v2.ReadChangeStreamResponse;
+import com.google.common.base.MoreObjects;
+import com.google.common.base.Objects;
+import com.google.protobuf.Timestamp;
+import java.io.Serializable;
+import javax.annotation.Nonnull;
+
+public final class Heartbeat implements ChangeStreamRecord, Serializable {
+  private static final long serialVersionUID = 7316215828353608504L;
+  private final Timestamp lowWatermark;
+  private final ChangeStreamContinuationToken changeStreamContinuationToken;
+
+  private Heartbeat(
+      Timestamp lowWatermark, ChangeStreamContinuationToken changeStreamContinuationToken) {
+    this.lowWatermark = lowWatermark;
+    this.changeStreamContinuationToken = changeStreamContinuationToken;
+  }
+
+  @InternalApi("Used in Changestream beam pipeline.")
+  public ChangeStreamContinuationToken getChangeStreamContinuationToken() {
+    return changeStreamContinuationToken;
+  }
+
+  @InternalApi("Used in Changestream beam pipeline.")
+  public Timestamp getLowWatermark() {
+    return lowWatermark;
+  }
+
+  /** Wraps the protobuf {@link ReadChangeStreamResponse.Heartbeat}. */
+  static Heartbeat fromProto(@Nonnull ReadChangeStreamResponse.Heartbeat heartbeat) {
+    return new Heartbeat(
+        heartbeat.getLowWatermark(),
+        ChangeStreamContinuationToken.fromProto(heartbeat.getContinuationToken()));
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (o == null || getClass() != o.getClass()) {
+      return false;
+    }
+    Heartbeat record = (Heartbeat) o;
+    return Objects.equal(lowWatermark, record.getLowWatermark())
+        && Objects.equal(changeStreamContinuationToken, record.getChangeStreamContinuationToken());
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hashCode(lowWatermark, changeStreamContinuationToken);
+  }
+
+  @Override
+  public String toString() {
+    return MoreObjects.toStringHelper(this)
+        .add("lowWatermark", lowWatermark)
+        .add("changeStreamContinuationToken", changeStreamContinuationToken)
+        .toString();
+  }
+}