Skip to content

Commit

Permalink
fix: Make BufferingPullSubscriber use List instead of ImmutableList s…
Browse files Browse the repository at this point in the history
…o it is beam friendly. (#256)
  • Loading branch information
dpcollins-google committed Sep 28, 2020
1 parent fcf5282 commit a23e26f
Show file tree
Hide file tree
Showing 2 changed files with 5 additions and 5 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -25,10 +25,10 @@
import com.google.cloud.pubsublite.proto.FlowControlRequest;
import com.google.cloud.pubsublite.proto.SeekRequest;
import com.google.cloud.pubsublite.proto.SeekRequest.NamedTarget;
import com.google.common.collect.ImmutableList;
import com.google.common.util.concurrent.MoreExecutors;
import io.grpc.StatusException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicReference;
Expand Down Expand Up @@ -75,7 +75,7 @@ public void failed(State state, Throwable throwable) {
}

@Override
public ImmutableList<SequencedMessage> pull() throws StatusException {
public List<SequencedMessage> pull() throws StatusException {
@Nullable StatusException maybeError = error.get();
if (maybeError != null) {
throw maybeError;
Expand All @@ -88,7 +88,7 @@ public ImmutableList<SequencedMessage> pull() throws StatusException {
.setAllowedBytes(bytes)
.setAllowedMessages(collection.size())
.build());
return ImmutableList.copyOf(collection);
return collection;
}

@Override
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -16,11 +16,11 @@

package com.google.cloud.pubsublite.internal;

import com.google.common.collect.ImmutableList;
import io.grpc.StatusException;
import java.util.List;

// A PullSubscriber exposes a "pull" mechanism for retrieving messages.
public interface PullSubscriber<T> extends AutoCloseable {
// Pull currently available messages from this subscriber. Does not block.
ImmutableList<T> pull() throws StatusException;
List<T> pull() throws StatusException;
}

0 comments on commit a23e26f

Please sign in to comment.