Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Minor cleanup in pubsub #4945

Merged
merged 2 commits into from
Apr 15, 2019
Merged

Minor cleanup in pubsub #4945

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f40edff
Minor cleanup in pubsub
sduskis Apr 15, 2019
d7fdee8
Fixing format
sduskis Apr 15, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 7 additions & 11 deletions ...ients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -104,10 +104,6 @@ private PendingModifyAckDeadline(int deadlineExtensionSeconds, Collection<String
this.deadlineExtensionSeconds = deadlineExtensionSeconds;
}

public void addAckId(String ackId) {
ackIds.add(ackId);
}

@Override
public String toString() {
return String.format(
Expand All @@ -129,7 +125,7 @@ private class AckHandler implements ApiFutureCallback<AckReply> {
private final long receivedTimeMillis;
private final Instant totalExpiration;

AckHandler(String ackId, int outstandingBytes, Instant totalExpiration) {
private AckHandler(String ackId, int outstandingBytes, Instant totalExpiration) {
this.ackId = ackId;
this.outstandingBytes = outstandingBytes;
this.receivedTimeMillis = clock.millisTime();
Expand Down Expand Up @@ -182,7 +178,7 @@ public void onSuccess(AckReply reply) {
}
}

public interface AckProcessor {
interface AckProcessor {
void sendAckOperations(
List<String> acksToSend, List<PendingModifyAckDeadline> ackDeadlineExtensions);
}
Expand Down Expand Up @@ -211,7 +207,7 @@ void sendAckOperations(
this.clock = clock;
}

public void start() {
void start() {
final Runnable setExtendDeadline =
new Runnable() {
@Override
Expand Down Expand Up @@ -264,7 +260,7 @@ public void run() {
}
}

public void stop() {
void stop() {
messagesWaiter.waitNoMessages();
jobLock.lock();
try {
Expand All @@ -288,17 +284,17 @@ int getMessageDeadlineSeconds() {
return messageDeadlineSeconds.get();
}

static class OutstandingMessage {
private static class OutstandingMessage {
private final ReceivedMessage receivedMessage;
private final AckHandler ackHandler;

public OutstandingMessage(ReceivedMessage receivedMessage, AckHandler ackHandler) {
private OutstandingMessage(ReceivedMessage receivedMessage, AckHandler ackHandler) {
this.receivedMessage = receivedMessage;
this.ackHandler = ackHandler;
}
}

public void processReceivedMessages(List<ReceivedMessage> messages) {
void processReceivedMessages(List<ReceivedMessage> messages) {
Instant totalExpiration = now().plus(maxAckExtensionPeriod);
List<OutstandingMessage> outstandingBatch = new ArrayList<>(messages.size());
for (ReceivedMessage message : messages) {
Expand Down
7 changes: 0 additions & 7 deletions ...d-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageWaiter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,6 @@
package com.google.cloud.pubsub.v1;

import com.google.api.core.InternalApi;
import java.util.concurrent.atomic.AtomicBoolean;

/** A barrier kind of object that helps to keep track and synchronously wait on pending messages. */
class MessageWaiter {
Expand All @@ -35,16 +34,10 @@ public synchronized void incrementPendingMessages(int messages) {
}

public synchronized void waitNoMessages() {
waitNoMessages(new AtomicBoolean());
}

@InternalApi
synchronized void waitNoMessages(AtomicBoolean waitReached) {
boolean interrupted = false;
try {
while (pendingMessages > 0) {
try {
waitReached.set(true);
wait();
} catch (InterruptedException e) {
// Ignored, uninterruptibly.
Expand Down
8 changes: 3 additions & 5 deletions ...ients/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/MessageWaiterTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,6 @@

import static org.junit.Assert.assertEquals;

import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
Expand All @@ -32,22 +31,21 @@ public void test() throws Exception {
final MessageWaiter waiter = new MessageWaiter();
waiter.incrementPendingMessages(1);

final AtomicBoolean waitReached = new AtomicBoolean();

final Thread mainThread = Thread.currentThread();
Thread t =
new Thread(
new Runnable() {
@Override
public void run() {
while (!waitReached.get()) {
while (mainThread.getState() != Thread.State.WAITING) {
Thread.yield();
}
waiter.incrementPendingMessages(-1);
}
});
t.start();

waiter.waitNoMessages(waitReached);
waiter.waitNoMessages();
t.join();

assertEquals(0, waiter.pendingMessages());
Expand Down