diff --git a/examples/src/main/java/io/dapr/examples/state/README.md b/examples/src/main/java/io/dapr/examples/state/README.md index d92dd23a88..f0a7f6b119 100644 --- a/examples/src/main/java/io/dapr/examples/state/README.md +++ b/examples/src/main/java/io/dapr/examples/state/README.md @@ -43,6 +43,10 @@ public class StateClient { ///... public static void main(String[] args) throws Exception { try (DaprClient client = new DaprClientBuilder().build()) { + System.out.println("Waiting for Dapr sidecar ..."); + client.waitForSidecar(10000).block(); + System.out.println("Dapr sidecar is ready."); + String message = args.length == 0 ? " " : args[0]; MyClass myClass = new MyClass(); @@ -102,7 +106,8 @@ public class StateClient { ``` The code uses the `DaprClient` created by the `DaprClientBuilder`. Notice that this builder uses default settings. Internally, it is using `DefaultObjectSerializer` for two properties: `objectSerializer` is for Dapr's sent and received objects, and `stateSerializer` is for objects to be persisted. -This example performs multiple operations: +This example performs multiple operations: +* `client.waitForSidecar(...)` for waiting until Dapr sidecar is ready. * `client.saveState(...)` for persisting an instance of `MyClass`. * `client.getState(...)` operation in order to retrieve back the persisted state using the same key. * `client.executeStateTransaction(...)` operation in order to update existing state and add new state. diff --git a/examples/src/main/java/io/dapr/examples/state/StateClient.java b/examples/src/main/java/io/dapr/examples/state/StateClient.java index dccacad0f1..c75f8e9ff9 100644 --- a/examples/src/main/java/io/dapr/examples/state/StateClient.java +++ b/examples/src/main/java/io/dapr/examples/state/StateClient.java @@ -46,6 +46,10 @@ public String toString() { */ public static void main(String[] args) throws Exception { try (DaprClient client = new DaprClientBuilder().build()) { + System.out.println("Waiting for Dapr sidecar ..."); + client.waitForSidecar(10000).block(); + System.out.println("Dapr sidecar is ready."); + String message = args.length == 0 ? " " : args[0]; MyClass myClass = new MyClass(); diff --git a/sdk/src/main/java/io/dapr/client/DaprClient.java b/sdk/src/main/java/io/dapr/client/DaprClient.java index f7c111f0a6..5b99dad1e2 100644 --- a/sdk/src/main/java/io/dapr/client/DaprClient.java +++ b/sdk/src/main/java/io/dapr/client/DaprClient.java @@ -32,6 +32,13 @@ */ public interface DaprClient extends AutoCloseable { + /** + * Waits for the sidecar, giving up after timeout. + * @param timeoutInMilliseconds Timeout in milliseconds to wait for sidecar. + * @return a Mono plan of type Void. + */ + Mono waitForSidecar(int timeoutInMilliseconds); + /** * Publish an event. * diff --git a/sdk/src/main/java/io/dapr/client/DaprClientGrpc.java b/sdk/src/main/java/io/dapr/client/DaprClientGrpc.java index 1d73b264db..408d587e7e 100644 --- a/sdk/src/main/java/io/dapr/client/DaprClientGrpc.java +++ b/sdk/src/main/java/io/dapr/client/DaprClientGrpc.java @@ -26,6 +26,7 @@ import io.dapr.config.Properties; import io.dapr.exceptions.DaprException; import io.dapr.serializer.DaprObjectSerializer; +import io.dapr.utils.NetworkUtils; import io.dapr.utils.TypeRef; import io.dapr.v1.CommonProtos; import io.dapr.v1.DaprGrpc; @@ -131,6 +132,20 @@ private CommonProtos.StateOptions.StateConcurrency getGrpcStateConcurrency(State } } + /** + * {@inheritDoc} + */ + @Override + public Mono waitForSidecar(int timeoutInMilliseconds) { + return Mono.fromRunnable(() -> { + try { + NetworkUtils.waitForSocket(Properties.SIDECAR_IP.get(), Properties.GRPC_PORT.get(), timeoutInMilliseconds); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + }); + } + /** * {@inheritDoc} */ diff --git a/sdk/src/main/java/io/dapr/client/DaprClientHttp.java b/sdk/src/main/java/io/dapr/client/DaprClientHttp.java index 732e84445a..2516c45d1c 100644 --- a/sdk/src/main/java/io/dapr/client/DaprClientHttp.java +++ b/sdk/src/main/java/io/dapr/client/DaprClientHttp.java @@ -26,6 +26,7 @@ import io.dapr.exceptions.DaprException; import io.dapr.serializer.DaprObjectSerializer; import io.dapr.serializer.DefaultObjectSerializer; +import io.dapr.utils.NetworkUtils; import io.dapr.utils.TypeRef; import io.opentelemetry.context.Context; import reactor.core.publisher.Mono; @@ -101,6 +102,20 @@ public class DaprClientHttp extends AbstractDaprClient { this(client, new DefaultObjectSerializer(), new DefaultObjectSerializer()); } + /** + * {@inheritDoc} + */ + @Override + public Mono waitForSidecar(int timeoutInMilliseconds) { + return Mono.fromRunnable(() -> { + try { + NetworkUtils.waitForSocket(Properties.SIDECAR_IP.get(), Properties.HTTP_PORT.get(), timeoutInMilliseconds); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + }); + } + /** * {@inheritDoc} */ diff --git a/sdk/src/main/java/io/dapr/utils/NetworkUtils.java b/sdk/src/main/java/io/dapr/utils/NetworkUtils.java new file mode 100644 index 0000000000..b9fff405f5 --- /dev/null +++ b/sdk/src/main/java/io/dapr/utils/NetworkUtils.java @@ -0,0 +1,42 @@ +/* + * Copyright (c) Microsoft Corporation. + * Licensed under the MIT License. + */ + +package io.dapr.utils; + +import java.io.IOException; +import java.net.InetSocketAddress; +import java.net.Socket; + +/** + * Utility methods for network, internal to Dapr SDK. + */ +public final class NetworkUtils { + + private NetworkUtils() { + } + + /** + * Tries to connect to a socket, retrying every 1 second. + * @param host Host to connect to. + * @param port Port to connect to. + * @param timeoutInMilliseconds Timeout in milliseconds to give up trying. + * @throws InterruptedException If retry is interrupted. + */ + public static void waitForSocket(String host, int port, int timeoutInMilliseconds) throws InterruptedException { + long started = System.currentTimeMillis(); + Retry.callWithRetry(() -> { + try { + try (Socket socket = new Socket()) { + // timeout cannot be negative. + // zero timeout means infinite, so 1 is the practical minimum. + int remainingTimeout = (int) Math.max(1, timeoutInMilliseconds - (System.currentTimeMillis() - started)); + socket.connect(new InetSocketAddress(host, port), remainingTimeout); + } + } catch (IOException e) { + throw new RuntimeException(e); + } + }, timeoutInMilliseconds); + } +} diff --git a/sdk/src/main/java/io/dapr/utils/Retry.java b/sdk/src/main/java/io/dapr/utils/Retry.java new file mode 100644 index 0000000000..09cc01c890 --- /dev/null +++ b/sdk/src/main/java/io/dapr/utils/Retry.java @@ -0,0 +1,41 @@ +/* + * Copyright (c) Microsoft Corporation. + * Licensed under the MIT License. + */ + +package io.dapr.utils; + +class Retry { + + private static final long RETRY_WAIT_MILLISECONDS = 1000; + + private Retry() { + } + + static void callWithRetry(Runnable function, long retryTimeoutMilliseconds) throws InterruptedException { + long started = System.currentTimeMillis(); + while (true) { + Throwable exception; + try { + function.run(); + return; + } catch (Exception e) { + exception = e; + } catch (AssertionError e) { + exception = e; + } + + long elapsed = System.currentTimeMillis() - started; + if (elapsed >= retryTimeoutMilliseconds) { + if (exception instanceof RuntimeException) { + throw (RuntimeException)exception; + } + + throw new RuntimeException(exception); + } + + long remaining = retryTimeoutMilliseconds - elapsed; + Thread.sleep(Math.min(remaining, RETRY_WAIT_MILLISECONDS)); + } + } +} diff --git a/sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java b/sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java index 0a341480db..770f5f5d2c 100644 --- a/sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java +++ b/sdk/src/test/java/io/dapr/client/DaprClientGrpcTest.java @@ -23,6 +23,7 @@ import io.dapr.client.domain.State; import io.dapr.client.domain.StateOptions; import io.dapr.client.domain.TransactionalStateOperation; +import io.dapr.config.Properties; import io.dapr.serializer.DaprObjectSerializer; import io.dapr.serializer.DefaultObjectSerializer; import io.dapr.utils.TypeRef; @@ -40,6 +41,8 @@ import java.io.Closeable; import java.io.IOException; +import java.net.ServerSocket; +import java.net.Socket; import java.nio.charset.StandardCharsets; import java.util.Arrays; import java.util.Collections; @@ -51,7 +54,9 @@ import static com.google.common.util.concurrent.Futures.addCallback; import static com.google.common.util.concurrent.MoreExecutors.directExecutor; import static io.dapr.utils.TestUtils.assertThrowsDaprException; +import static io.dapr.utils.TestUtils.findFreePort; import static org.junit.Assert.*; +import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.argThat; import static org.mockito.Mockito.doNothing; @@ -89,6 +94,30 @@ public void tearDown() throws Exception { verifyNoMoreInteractions(closeable); } + @Test + public void waitForSidecarTimeout() throws Exception { + int port = findFreePort(); + System.setProperty(Properties.GRPC_PORT.getName(), Integer.toString(port)); + assertThrows(RuntimeException.class, () -> adapter.waitForSidecar(1).block()); + } + + @Test + public void waitForSidecarTimeoutOK() throws Exception { + try (ServerSocket serverSocket = new ServerSocket(0)) { + final int port = serverSocket.getLocalPort(); + System.setProperty(Properties.GRPC_PORT.getName(), Integer.toString(port)); + Thread t = new Thread(() -> { + try { + try (Socket socket = serverSocket.accept()) { + } + } catch (IOException e) { + } + }); + t.start(); + adapter.waitForSidecar(10000).block(); + } + } + @Test public void publishEventExceptionThrownTest() { when(client.publishEvent(any(DaprProtos.PublishEventRequest.class))) diff --git a/sdk/src/test/java/io/dapr/client/DaprClientHttpTest.java b/sdk/src/test/java/io/dapr/client/DaprClientHttpTest.java index 467642aa6e..4622a70643 100644 --- a/sdk/src/test/java/io/dapr/client/DaprClientHttpTest.java +++ b/sdk/src/test/java/io/dapr/client/DaprClientHttpTest.java @@ -6,8 +6,8 @@ import com.fasterxml.jackson.core.JsonParseException; import io.dapr.client.domain.DeleteStateRequestBuilder; -import io.dapr.client.domain.GetStateRequestBuilder; import io.dapr.client.domain.GetBulkStateRequestBuilder; +import io.dapr.client.domain.GetStateRequestBuilder; import io.dapr.client.domain.HttpExtension; import io.dapr.client.domain.Response; import io.dapr.client.domain.State; @@ -26,6 +26,8 @@ import reactor.core.publisher.Mono; import java.io.IOException; +import java.net.ServerSocket; +import java.net.Socket; import java.util.ArrayList; import java.util.Arrays; import java.util.Base64; @@ -35,10 +37,8 @@ import java.util.Map; import static io.dapr.utils.TestUtils.assertThrowsDaprException; -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertNotNull; -import static org.junit.Assert.assertNull; -import static org.junit.Assert.assertTrue; +import static io.dapr.utils.TestUtils.findFreePort; +import static org.junit.Assert.*; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.Mockito.mock; @@ -65,6 +65,34 @@ public void setUp() { okHttpClient = new OkHttpClient.Builder().addInterceptor(mockInterceptor).build(); } + @Test + public void waitForSidecarTimeout() throws Exception { + int port = findFreePort(); + System.setProperty(Properties.HTTP_PORT.getName(), Integer.toString(port)); + daprHttp = new DaprHttp(Properties.SIDECAR_IP.get(), port, okHttpClient); + DaprClientHttp daprClientHttp = new DaprClientHttp(daprHttp); + assertThrows(RuntimeException.class, () -> daprClientHttp.waitForSidecar(1).block()); + } + + @Test + public void waitForSidecarTimeoutOK() throws Exception { + try (ServerSocket serverSocket = new ServerSocket(0)) { + final int port = serverSocket.getLocalPort(); + System.setProperty(Properties.HTTP_PORT.getName(), Integer.toString(port)); + Thread t = new Thread(() -> { + try { + try (Socket socket = serverSocket.accept()) { + } + } catch (IOException e) { + } + }); + t.start(); + daprHttp = new DaprHttp(Properties.SIDECAR_IP.get(), port, okHttpClient); + DaprClientHttp daprClientHttp = new DaprClientHttp(daprHttp); + daprClientHttp.waitForSidecar(10000).block(); + } + } + @Test public void publishEventInvokation() { mockInterceptor.addRule() diff --git a/sdk/src/test/java/io/dapr/utils/TestUtils.java b/sdk/src/test/java/io/dapr/utils/TestUtils.java index d29a77be61..60489970fa 100644 --- a/sdk/src/test/java/io/dapr/utils/TestUtils.java +++ b/sdk/src/test/java/io/dapr/utils/TestUtils.java @@ -9,6 +9,11 @@ import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.function.Executable; +import java.io.IOException; +import java.net.ServerSocket; +import java.util.HashSet; +import java.util.Set; + public final class TestUtils { private TestUtils() {} @@ -46,4 +51,11 @@ public static void assertThrowsDaprException( Assertions.assertEquals(expectedErrorCode, daprException.getErrorCode()); Assertions.assertEquals(expectedErrorMessage, daprException.getMessage()); } + + public static int findFreePort() throws IOException { + try (ServerSocket socket = new ServerSocket(0)) { + socket.setReuseAddress(true); + return socket.getLocalPort(); + } + } }