api: Add java.time.Duration overloads to CallOptions, AbstractStub

api/src/main/java/io/grpc/CallOptions.java

@@ -17,9 +17,11 @@
 package io.grpc;
 
 import static com.google.common.base.Preconditions.checkArgument;
+import static io.grpc.InternalTimeUtils.convert;
 
 import com.google.common.base.MoreObjects;
 import com.google.common.base.Preconditions;
+import java.time.Duration;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -176,6 +178,10 @@ public CallOptions withDeadlineAfter(long duration, TimeUnit unit) {
     return withDeadline(Deadline.after(duration, unit));
   }
 
+  public CallOptions withDeadlineAfter(Duration duration) {
+    return withDeadlineAfter(convert(duration), TimeUnit.NANOSECONDS);
+  }
+
   /**
    * Returns the deadline or {@code null} if the deadline is not set.
    */

api/src/main/java/io/grpc/InternalTimeUtils.java

@@ -0,0 +1,30 @@
+/*
+ * Copyright 2024 The gRPC Authors
+ *
+ * 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
+ *
+ *     http://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 io.grpc;
+
+import java.time.Duration;
+
+@Internal
+public final class InternalTimeUtils {
+  public static long convert(Duration duration) {
+    try {
+      return duration.toNanos();
+    } catch (ArithmeticException tooBig) {
+      return duration.isNegative() ? Long.MIN_VALUE : Long.MAX_VALUE;
+    }
+  }
+}

api/src/main/java/io/grpc/SynchronizationContext.java

@@ -18,8 +18,10 @@
 
 import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.base.Preconditions.checkState;
+import static io.grpc.InternalTimeUtils.convert;
 
 import java.lang.Thread.UncaughtExceptionHandler;
+import java.time.Duration;
 import java.util.Queue;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.Executor;
@@ -193,6 +195,13 @@ public String toString() {
     return new ScheduledHandle(runnable, future);
   }
 
+  public final ScheduledHandle scheduleWithFixedDelay(
+      final Runnable task, Duration initialDelay, Duration delay,
+      ScheduledExecutorService timerService) {
+    return scheduleWithFixedDelay(task, convert(initialDelay), convert(delay),
+        TimeUnit.NANOSECONDS, timerService);
+  }
+
 
   private static class ManagedRunnable implements Runnable {
     final Runnable task;
@@ -246,4 +255,4 @@ public boolean isPending() {
       return !(runnable.hasStarted || runnable.isCancelled);
     }
   }
-}
+}

api/src/test/java/io/grpc/CallOptionsTest.java

@@ -32,6 +32,7 @@
 import com.google.common.base.Objects;
 import io.grpc.ClientStreamTracer.StreamInfo;
 import io.grpc.internal.SerializingExecutor;
+import java.time.Duration;
 import java.util.concurrent.Executor;
 import org.junit.Test;
 import org.junit.runner.RunWith;
@@ -150,6 +151,14 @@ public void withDeadlineAfter() {
     assertAbout(deadline()).that(actual).isWithin(10, MILLISECONDS).of(expected);
   }
 
+  @Test
+  public void withDeadlineAfterDuration() {
+    Deadline actual = CallOptions.DEFAULT.withDeadlineAfter(Duration.ofMinutes(1L)).getDeadline();
+    Deadline expected = Deadline.after(1, MINUTES);
+
+    assertAbout(deadline()).that(actual).isWithin(10, MILLISECONDS).of(expected);
+  }
+
   @Test
   public void toStringMatches_noDeadline_default() {
     String actual = allSet

api/src/test/java/io/grpc/InternalTimeUtilsTest.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2024 The gRPC Authors
+ *
+ * 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
+ *
+ *     http://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 io.grpc;
+
+import static io.grpc.InternalTimeUtils.convert;
+import static org.junit.Assert.assertEquals;
+
+import java.time.Duration;
+import org.junit.Test;
+
+public class InternalTimeUtilsTest {
+
+  @Test
+  public void testConvertNormalDuration() {
+    Duration duration = Duration.ofSeconds(10);
+    long expected = 10 * 1_000_000_000L;
+
+    assertEquals(expected, convert(duration));
+  }
+
+  @Test
+  public void testConvertNegativeDuration() {
+    Duration duration = Duration.ofSeconds(-3);
+    long expected = -3 * 1_000_000_000L;
+
+    assertEquals(expected, convert(duration));
+  }
+
+  @Test
+  public void testConvertTooLargeDuration() {
+    Duration duration = Duration.ofSeconds(Long.MAX_VALUE / 1_000_000_000L + 1);
+
+    assertEquals(Long.MAX_VALUE, convert(duration));
+  }
+
+  @Test
+  public void testConvertTooLargeNegativeDuration() {
+    Duration duration = Duration.ofSeconds(Long.MIN_VALUE / 1_000_000_000L - 1);
+
+    assertEquals(Long.MIN_VALUE, convert(duration));
+  }
+}

api/src/test/java/io/grpc/SynchronizationContextTest.java

@@ -27,6 +27,7 @@
 
 import com.google.common.util.concurrent.testing.TestingExecutors;
 import io.grpc.SynchronizationContext.ScheduledHandle;
+import java.time.Duration;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.LinkedBlockingQueue;
@@ -72,7 +73,7 @@ public void uncaughtException(Thread t, Throwable e) {
 
   @Mock
   private Runnable task3;
-  
+
   @After public void tearDown() {
     assertThat(uncaughtErrors).isEmpty();
   }
@@ -105,36 +106,36 @@ public void multiThread() throws Exception {
     final AtomicReference<Thread> task2Thread = new AtomicReference<>();
 
     doAnswer(new Answer<Void>() {
-        @Override
-        public Void answer(InvocationOnMock invocation) {
-          task1Thread.set(Thread.currentThread());
-          task1Running.countDown();
-          try {
-            assertTrue(task1Proceed.await(5, TimeUnit.SECONDS));
-          } catch (InterruptedException e) {
-            throw new RuntimeException(e);
-          }
-          return null;
+      @Override
+      public Void answer(InvocationOnMock invocation) {
+        task1Thread.set(Thread.currentThread());
+        task1Running.countDown();
+        try {
+          assertTrue(task1Proceed.await(5, TimeUnit.SECONDS));
+        } catch (InterruptedException e) {
+          throw new RuntimeException(e);
         }
-      }).when(task1).run();
+        return null;
+      }
+    }).when(task1).run();
 
     doAnswer(new Answer<Void>() {
-        @Override
-        public Void answer(InvocationOnMock invocation) {
-          task2Thread.set(Thread.currentThread());
-          return null;
-        }
-      }).when(task2).run();
+      @Override
+      public Void answer(InvocationOnMock invocation) {
+        task2Thread.set(Thread.currentThread());
+        return null;
+      }
+    }).when(task2).run();
 
     Thread sideThread = new Thread() {
-        @Override
-        public void run() {
-          syncContext.executeLater(task1);
-          task1Added.countDown();
-          syncContext.drain();
-          sideThreadDone.countDown();
-        }
-      };
+      @Override
+      public void run() {
+        syncContext.executeLater(task1);
+        task1Added.countDown();
+        syncContext.drain();
+        sideThreadDone.countDown();
+      }
+    };
     sideThread.start();
 
     assertTrue(task1Added.await(5, TimeUnit.SECONDS));
@@ -162,26 +163,26 @@ public void throwIfNotInThisSynchronizationContext() throws Exception {
     final CountDownLatch task1Proceed = new CountDownLatch(1);
 
     doAnswer(new Answer<Void>() {
-        @Override
-        public Void answer(InvocationOnMock invocation) {
-          task1Running.countDown();
-          syncContext.throwIfNotInThisSynchronizationContext();
-          try {
-            assertTrue(task1Proceed.await(5, TimeUnit.SECONDS));
-          } catch (InterruptedException e) {
-            throw new RuntimeException(e);
-          }
-          taskSuccess.set(true);
-          return null;
+      @Override
+      public Void answer(InvocationOnMock invocation) {
+        task1Running.countDown();
+        syncContext.throwIfNotInThisSynchronizationContext();
+        try {
+          assertTrue(task1Proceed.await(5, TimeUnit.SECONDS));
+        } catch (InterruptedException e) {
+          throw new RuntimeException(e);
         }
-      }).when(task1).run();
+        taskSuccess.set(true);
+        return null;
+      }
+    }).when(task1).run();
 
     Thread sideThread = new Thread() {
-        @Override
-        public void run() {
-          syncContext.execute(task1);
-        }
-      };
+      @Override
+      public void run() {
+        syncContext.execute(task1);
+      }
+    };
     sideThread.start();
 
     assertThat(task1Running.await(5, TimeUnit.SECONDS)).isTrue();
@@ -215,11 +216,11 @@ public void taskThrows() {
     InOrder inOrder = inOrder(task1, task2, task3);
     final RuntimeException e = new RuntimeException("Simulated");
     doAnswer(new Answer<Void>() {
-        @Override
-        public Void answer(InvocationOnMock invocation) {
-          throw e;
-        }
-      }).when(task2).run();
+      @Override
+      public Void answer(InvocationOnMock invocation) {
+        throw e;
+      }
+    }).when(task2).run();
     syncContext.executeLater(task1);
     syncContext.executeLater(task2);
     syncContext.executeLater(task3);
@@ -246,6 +247,24 @@ public void schedule() {
     verify(task1).run();
   }
 
+  @Test
+  public void scheduleWithFixedDelayDuration() {
+    MockScheduledExecutorService executorService = new MockScheduledExecutorService();
+    ScheduledHandle handle =
+        syncContext.scheduleWithFixedDelay(task1, Duration.ofSeconds(10),
+            Duration.ofSeconds(10), executorService);
+
+    assertThat(executorService.delay)
+        .isEqualTo(executorService.unit.convert(10, TimeUnit.SECONDS));
+    assertThat(handle.isPending()).isTrue();
+    verify(task1, never()).run();
+
+    executorService.command.run();
+
+    assertThat(handle.isPending()).isFalse();
+    verify(task1).run();
+  }
+
   @Test
   public void scheduleDueImmediately() {
     MockScheduledExecutorService executorService = new MockScheduledExecutorService();
@@ -288,28 +307,28 @@ public void scheduledHandle_cancelRacesWithTimerExpiration() throws Exception {
     final CountDownLatch sideThreadDone = new CountDownLatch(1);
 
     doAnswer(new Answer<Void>() {
-        @Override
-        public Void answer(InvocationOnMock invocation) {
-          task1Running.countDown();
-          try {
-            ScheduledHandle task2Handle;
-            assertThat(task2Handle = task2HandleQueue.poll(5, TimeUnit.SECONDS)).isNotNull();
-            task2Handle.cancel();
-          } catch (InterruptedException e) {
-            throw new RuntimeException(e);
-          }
-          task1Done.set(true);
-          return null;
+      @Override
+      public Void answer(InvocationOnMock invocation) {
+        task1Running.countDown();
+        try {
+          ScheduledHandle task2Handle;
+          assertThat(task2Handle = task2HandleQueue.poll(5, TimeUnit.SECONDS)).isNotNull();
+          task2Handle.cancel();
+        } catch (InterruptedException e) {
+          throw new RuntimeException(e);
         }
-      }).when(task1).run();
+        task1Done.set(true);
+        return null;
+      }
+    }).when(task1).run();
 
     Thread sideThread = new Thread() {
-        @Override
-        public void run() {
-          syncContext.execute(task1);
-          sideThreadDone.countDown();
-        }
-      };
+      @Override
+      public void run() {
+        syncContext.execute(task1);
+        sideThreadDone.countDown();
+      }
+    };
 
     ScheduledHandle handle = syncContext.schedule(task2, 10, TimeUnit.NANOSECONDS, executorService);
     // This will execute and block in task1
@@ -357,5 +376,13 @@ static class MockScheduledExecutorService extends ForwardingScheduledExecutorSer
       this.unit = unit;
       return future = super.schedule(command, delay, unit);
     }
+
+    @Override public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long intialDelay,
+        long delay, TimeUnit unit) {
+      this.command = command;
+      this.delay = delay;
+      this.unit = unit;
+      return future = super.scheduleWithFixedDelay(command, intialDelay, delay, unit);
+    }
   }
-}
+}