---

sduskis · web-flow · commit cb1013d8381b · 2019-04-15T14:26:14.000-04:00
yaml --- r: 32703 b: refs/heads/autosynth-iot c: 2fc4e55 h: refs/heads/master i: 32701: 63fdd7b 32699: 34b7560 32695: 5f5112e 32687: c981b4a 32671: 7b025c6 32639: d9d66f6
diff --git a/[refs] b/[refs]
@@ -131,7 +131,7 @@ refs/heads/autosynth-datastore: f1efc3dc465174f41041acd56cf29badcec3e5bd
 refs/heads/autosynth-dialogflow: 73974cc32e5212aec0126472e0bc1442886fedaf
 refs/heads/autosynth-errorreporting: effe8001d110ad584187b30aafc473db0dd4a15f
 refs/heads/autosynth-firestore: e79eeb26930dfae4439424ad2fda5874eeca54c8
-refs/heads/autosynth-iot: 78360de040eb4d86d1b25a7792269d2952174d66
+refs/heads/autosynth-iot: 2fc4e551b5263fe1e902efdbc1fd01acf8e3b5e9
 refs/heads/autosynth-kms: 6b65b0f34c12d141031c7288cdc01e550212d0f6
 refs/heads/autosynth-language: e73905aa7672afa47240e65b25c087207f4594f9
 refs/heads/autosynth-os-login: 123ba209c5769d0ee067e0ce5848bec13b42a4f4
diff --git a/branches/autosynth-iot/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java b/branches/autosynth-iot/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageDispatcher.java
@@ -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(
@@ -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();
@@ -182,7 +178,7 @@ public void onSuccess(AckReply reply) {
     }
   }
 
-  public interface AckProcessor {
+  interface AckProcessor {
     void sendAckOperations(
         List<String> acksToSend, List<PendingModifyAckDeadline> ackDeadlineExtensions);
   }
@@ -211,7 +207,7 @@ void sendAckOperations(
     this.clock = clock;
   }
 
-  public void start() {
+  void start() {
     final Runnable setExtendDeadline =
         new Runnable() {
           @Override
@@ -264,7 +260,7 @@ public void run() {
     }
   }
 
-  public void stop() {
+  void stop() {
     messagesWaiter.waitNoMessages();
     jobLock.lock();
     try {
@@ -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) {
diff --git a/branches/autosynth-iot/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageWaiter.java b/branches/autosynth-iot/google-cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/MessageWaiter.java
@@ -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 {
@@ -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.
diff --git a/branches/autosynth-iot/google-cloud-clients/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/MessageWaiterTest.java b/branches/autosynth-iot/google-cloud-clients/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/v1/MessageWaiterTest.java
@@ -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;
@@ -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());
