wip

pongad · pongad · commit daa24e501fa7 · 2017-01-20T18:05:23.000+11:00
diff --git a/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/MessageDispatcher.java b/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/spi/v1/MessageDispatcher.java
@@ -268,7 +268,6 @@ public void processReceivedMessages(List<com.google.pubsub.v1.ReceivedMessage> r
     }
     Instant expiration = now.plus(messageDeadlineSeconds * 1000);
     logger.debug("Received {} messages at {}", responseMessages.size(), now);
-    setupNextAckDeadlineExtensionAlarm(expiration);
 
     messagesWaiter.incrementPendingMessages(responseMessages.size());
     Iterator<AckHandler> acksIterator = ackHandlers.iterator();
@@ -288,6 +287,7 @@ public void run() {
       outstandingAckHandlers.add(
           new ExtensionJob(expiration, INITIAL_ACK_DEADLINE_EXTENSION_SECONDS, ackHandlers));
     }
+    setupNextAckDeadlineExtensionAlarm(expiration);
 
     try {
       flowController.reserve(receivedMessagesCount, totalByteCount);
diff --git a/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/spi/v1/FakeScheduledExecutorService.java b/google-cloud-pubsub/src/test/java/com/google/cloud/pubsub/spi/v1/FakeScheduledExecutorService.java
@@ -91,20 +91,29 @@ public void advanceTime(Duration toAdvance) {
     clock.advance(toAdvance.getMillis(), TimeUnit.MILLISECONDS);
     DateTime cmpTime = new DateTime(clock.millis());
 
-    synchronized (pendingCallables) {
-      while (!pendingCallables.isEmpty()
-          && pendingCallables.peek().getScheduledTime().compareTo(cmpTime) <= 0) {
-        try {
-          pendingCallables.poll().call();
-          if (shutdown.get() && pendingCallables.isEmpty()) {
-            pendingCallables.notifyAll();
-          }
+    for (;;) {
+      PendingCallable<?> callable = null;
+      synchronized (pendingCallables) {
+        if (pendingCallables.isEmpty() || pendingCallables.peek().getScheduledTime().isAfter(cmpTime)) {
+          break;
+        }
+        callable = pendingCallables.poll();
+      }
+      if (callable != null) {
+        try{
+          callable.call();
         } catch (Exception e) {
           // We ignore any callable exception, which should be set to the future but not relevant to
           // advanceTime.
         }
       }
     }
+
+    synchronized (pendingCallables) {
+      if (shutdown.get() && pendingCallables.isEmpty()) {
+        pendingCallables.notifyAll();
+      }
+    }
   }
 
   @Override
@@ -172,6 +181,7 @@ <V> ScheduledFuture<V> schedulePendingCallable(PendingCallable<V> callable) {
     synchronized (pendingCallables) {
       pendingCallables.add(callable);
     }
+    tick(0, TimeUnit.MILLISECONDS);
     return callable.getScheduledFuture();
   }
 

Original file line number	Diff line number	Diff line change
`@@ -268,7 +268,6 @@ public void processReceivedMessages(List<com.google.pubsub.v1.ReceivedMessage> r`
`268`	`268`	`}`
`269`	`269`	`Instant expiration = now.plus(messageDeadlineSeconds * 1000);`
`270`	`270`	`logger.debug("Received {} messages at {}", responseMessages.size(), now);`
`271`		`- setupNextAckDeadlineExtensionAlarm(expiration);`
`272`	`271`
`273`	`272`	`messagesWaiter.incrementPendingMessages(responseMessages.size());`
`274`	`273`	`Iterator<AckHandler> acksIterator = ackHandlers.iterator();`
`@@ -288,6 +287,7 @@ public void run() {`
`288`	`287`	`outstandingAckHandlers.add(`
`289`	`288`	`new ExtensionJob(expiration, INITIAL_ACK_DEADLINE_EXTENSION_SECONDS, ackHandlers));`
`290`	`289`	`}`
	`290`	`+ setupNextAckDeadlineExtensionAlarm(expiration);`
`291`	`291`
`292`	`292`	`try {`
`293`	`293`	`flowController.reserve(receivedMessagesCount, totalByteCount);`