diff --git a/.github/workflows/build-validation.yml b/.github/workflows/build-validation.yml index 20635b43..df14fc6a 100644 --- a/.github/workflows/build-validation.yml +++ b/.github/workflows/build-validation.yml @@ -42,7 +42,25 @@ jobs: uses: gradle/gradle-build-action@v2 - name: Build with Gradle - run: ./gradlew build + run: ./gradlew build -x test + + - name: Run Unit Tests with Gradle + run: | + export JAVA_HOME=$JDK_11 + ./gradlew clean test || echo "UNIT_TEST_FAILED=true" >> $GITHUB_ENV + continue-on-error: true + + - name: Upload test reports if tests failed + if: env.UNIT_TEST_FAILED == 'true' + uses: actions/upload-artifact@v4 + with: + name: Unit Test Reports + path: '**/build/reports/tests/test' + if-no-files-found: ignore # Prevents errors if no reports exist + + - name: Fail the job if unit tests failed + if: env.UNIT_TEST_FAILED == 'true' + run: exit 1 # TODO: Move the sidecar into a central image repository - name: Initialize Durable Task Sidecar diff --git a/.gitignore b/.gitignore index 3ff2e24d..fb0be4ee 100644 --- a/.gitignore +++ b/.gitignore @@ -6,4 +6,7 @@ build/ .project .settings .classpath -repo/ \ No newline at end of file +repo/ + +# Ignore sample application properties or any other properties files used for sample +samples/src/main/resources/*.properties \ No newline at end of file diff --git a/CHANGELOG.md b/CHANGELOG.md index 5ab2a421..b4ce7e4e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,4 +1,5 @@ ## placeholder +* DTS Support ([#201](https://github.com/microsoft/durabletask-java/pull/201)) * Add automatic proto file download and commit hash tracking during build ([#207](https://github.com/microsoft/durabletask-java/pull/207)) * Fix infinite loop when use continueasnew after wait external event ([#183](https://github.com/microsoft/durabletask-java/pull/183)) * Fix the issue "Deserialize Exception got swallowed when use anyOf with external event." ([#185](https://github.com/microsoft/durabletask-java/pull/185)) diff --git a/azuremanaged/build.gradle b/azuremanaged/build.gradle new file mode 100644 index 00000000..c5d645b6 --- /dev/null +++ b/azuremanaged/build.gradle @@ -0,0 +1,116 @@ +/* + * This build file was generated by the Gradle 'init' task. + * + * This generated file contains a sample Java project to get you started. + * For more details take a look at the Java Quickstart chapter in the Gradle + * user guide available at https://docs.gradle.org/4.4.1/userguide/tutorial_java_projects.html + */ + +plugins { + id 'java' + id 'com.google.protobuf' version '0.8.16' + id 'idea' + id 'maven-publish' + id 'signing' +} + +archivesBaseName = 'durabletask-azuremanaged' +group 'com.microsoft' +version = '1.5.0-preview.1' + +def grpcVersion = '1.59.0' +def azureCoreVersion = '1.45.0' +def azureIdentityVersion = '1.11.1' +// When build on local, you need to set this value to your local jdk11 directory. +// Java11 is used to compile and run all the tests. +def PATH_TO_TEST_JAVA_RUNTIME = System.env.JDK_11 ?: System.getProperty("java.home") + +repositories { + mavenCentral() +} + +dependencies { + implementation project(':client') + implementation "io.grpc:grpc-protobuf:${grpcVersion}" + implementation "io.grpc:grpc-stub:${grpcVersion}" + runtimeOnly "io.grpc:grpc-netty-shaded:${grpcVersion}" + + implementation "com.azure:azure-core:${azureCoreVersion}" + implementation "com.azure:azure-identity:${azureIdentityVersion}" + + // Test dependencies + testImplementation 'org.junit.jupiter:junit-jupiter-api:5.10.0' + testImplementation 'org.junit.jupiter:junit-jupiter-params:5.10.0' + testImplementation 'org.mockito:mockito-core:5.3.1' + testImplementation 'org.mockito:mockito-junit-jupiter:5.3.1' + testRuntimeOnly 'org.junit.jupiter:junit-jupiter-engine:5.10.0' +} + +compileJava { + sourceCompatibility = JavaVersion.VERSION_1_8 + targetCompatibility = JavaVersion.VERSION_1_8 +} + +compileTestJava { + sourceCompatibility = JavaVersion.VERSION_11 + targetCompatibility = JavaVersion.VERSION_11 + options.fork = true + options.forkOptions.executable = "${PATH_TO_TEST_JAVA_RUNTIME}/bin/javac" +} + +test { + useJUnitPlatform() +} + +publishing { + repositories { + maven { + url "file://$project.rootDir/repo" + } + } + publications { + mavenJava(MavenPublication) { + from components.java + artifactId = archivesBaseName + pom { + name = 'Durable Task Azure Managed SDK for Java' + description = 'This package contains classes and interfaces for building Durable Task orchestrations in Java using Azure Managed mode.' + url = "https://github.com/microsoft/durabletask-java/tree/main/azuremanaged" + licenses { + license { + name = "MIT License" + url = "https://opensource.org/licenses/MIT" + distribution = "repo" + } + } + developers { + developer { + id = "Microsoft" + name = "Microsoft Corporation" + } + } + scm { + connection = "scm:git:https://github.com/microsoft/durabletask-java" + developerConnection = "scm:git:git@github.com:microsoft/durabletask-java" + url = "https://github.com/microsoft/durabletask-java/tree/main/azuremanaged" + } + withXml { + project.configurations.compileOnly.allDependencies.each { dependency -> + asNode().dependencies[0].appendNode("dependency").with { + it.appendNode("groupId", dependency.group) + it.appendNode("artifactId", dependency.name) + it.appendNode("version", dependency.version) + it.appendNode("scope", "provided") + } + } + } + } + } + } +} + +java { + withSourcesJar() + withJavadocJar() +} + diff --git a/azuremanaged/src/main/java/com/microsoft/durabletask/azuremanaged/AccessTokenCache.java b/azuremanaged/src/main/java/com/microsoft/durabletask/azuremanaged/AccessTokenCache.java new file mode 100644 index 00000000..888d20b3 --- /dev/null +++ b/azuremanaged/src/main/java/com/microsoft/durabletask/azuremanaged/AccessTokenCache.java @@ -0,0 +1,51 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.microsoft.durabletask.azuremanaged; + +import com.azure.core.credential.TokenCredential; +import com.azure.core.credential.AccessToken; +import com.azure.core.credential.TokenRequestContext; + +import java.time.Duration; +import java.time.OffsetDateTime; + +/** + * Caches access tokens for Azure authentication. + * This class is used by both client and worker components to authenticate with Azure-managed Durable Task Scheduler. + */ +public final class AccessTokenCache { + private final TokenCredential credential; + private final TokenRequestContext context; + private final Duration margin; + private AccessToken cachedToken; + + /** + * Creates a new instance of the AccessTokenCache. + * + * @param credential The token credential to use for obtaining tokens. + * @param context The token request context specifying the scopes. + * @param margin The time margin before token expiration to refresh the token. + */ + public AccessTokenCache(TokenCredential credential, TokenRequestContext context, Duration margin) { + this.credential = credential; + this.context = context; + this.margin = margin; + } + + /** + * Gets a valid access token, refreshing it if necessary. + * + * @return A valid access token. + */ + public AccessToken getToken() { + OffsetDateTime nowWithMargin = OffsetDateTime.now().plus(margin); + + if (cachedToken == null + || cachedToken.getExpiresAt().isBefore(nowWithMargin)) { + this.cachedToken = credential.getToken(context).block(); + } + + return cachedToken; + } +} \ No newline at end of file diff --git a/azuremanaged/src/main/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerClientExtensions.java b/azuremanaged/src/main/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerClientExtensions.java new file mode 100644 index 00000000..ffc7d446 --- /dev/null +++ b/azuremanaged/src/main/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerClientExtensions.java @@ -0,0 +1,107 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.microsoft.durabletask.azuremanaged; + +import com.microsoft.durabletask.DurableTaskGrpcClientBuilder; +import com.azure.core.credential.TokenCredential; +import io.grpc.Channel; +import java.util.Objects; +import javax.annotation.Nullable; + +/** + * Extension methods for creating DurableTaskClient instances that connect to Azure-managed Durable Task Scheduler. + * This class provides various methods to create and configure clients using either connection strings or explicit parameters. + */ +public final class DurableTaskSchedulerClientExtensions { + + private DurableTaskSchedulerClientExtensions() {} + + /** + * Configures a DurableTaskGrpcClientBuilder to use Azure-managed Durable Task Scheduler with a connection string. + * + * @param builder The builder to configure. + * @param connectionString The connection string for Azure-managed Durable Task Scheduler. + * @throws NullPointerException if builder or connectionString is null + */ + public static void useDurableTaskScheduler( + DurableTaskGrpcClientBuilder builder, + String connectionString) { + Objects.requireNonNull(builder, "builder must not be null"); + Objects.requireNonNull(connectionString, "connectionString must not be null"); + + configureBuilder(builder, + DurableTaskSchedulerClientOptions.fromConnectionString(connectionString)); + } + + /** + * Configures a DurableTaskGrpcClientBuilder to use Azure-managed Durable Task Scheduler with explicit parameters. + * + * @param builder The builder to configure. + * @param endpoint The endpoint address for Azure-managed Durable Task Scheduler. + * @param taskHubName The name of the task hub to connect to. + * @param tokenCredential The token credential for authentication, or null for anonymous access. + * @throws NullPointerException if builder, endpoint, or taskHubName is null + */ + public static void useDurableTaskScheduler( + DurableTaskGrpcClientBuilder builder, + String endpoint, + String taskHubName, + @Nullable TokenCredential tokenCredential) { + Objects.requireNonNull(builder, "builder must not be null"); + Objects.requireNonNull(endpoint, "endpoint must not be null"); + Objects.requireNonNull(taskHubName, "taskHubName must not be null"); + + configureBuilder(builder, new DurableTaskSchedulerClientOptions() + .setEndpointAddress(endpoint) + .setTaskHubName(taskHubName) + .setCredential(tokenCredential)); + } + + /** + * Creates a DurableTaskGrpcClientBuilder configured for Azure-managed Durable Task Scheduler using a connection string. + * + * @param connectionString The connection string for Azure-managed Durable Task Scheduler. + * @return A new configured DurableTaskGrpcClientBuilder instance. + * @throws NullPointerException if connectionString is null + */ + public static DurableTaskGrpcClientBuilder createClientBuilder( + String connectionString) { + Objects.requireNonNull(connectionString, "connectionString must not be null"); + return createBuilderFromOptions( + DurableTaskSchedulerClientOptions.fromConnectionString(connectionString)); + } + + /** + * Creates a DurableTaskGrpcClientBuilder configured for Azure-managed Durable Task Scheduler using explicit parameters. + * + * @param endpoint The endpoint address for Azure-managed Durable Task Scheduler. + * @param taskHubName The name of the task hub to connect to. + * @param tokenCredential The token credential for authentication, or null for anonymous access. + * @return A new configured DurableTaskGrpcClientBuilder instance. + * @throws NullPointerException if endpoint or taskHubName is null + */ + public static DurableTaskGrpcClientBuilder createClientBuilder( + String endpoint, + String taskHubName, + @Nullable TokenCredential tokenCredential) { + Objects.requireNonNull(endpoint, "endpoint must not be null"); + Objects.requireNonNull(taskHubName, "taskHubName must not be null"); + + return createBuilderFromOptions(new DurableTaskSchedulerClientOptions() + .setEndpointAddress(endpoint) + .setTaskHubName(taskHubName) + .setCredential(tokenCredential)); + } + + // Private helper methods to reduce code duplication + private static DurableTaskGrpcClientBuilder createBuilderFromOptions(DurableTaskSchedulerClientOptions options) { + Channel grpcChannel = options.createGrpcChannel(); + return new DurableTaskGrpcClientBuilder().grpcChannel(grpcChannel); + } + + private static void configureBuilder(DurableTaskGrpcClientBuilder builder, DurableTaskSchedulerClientOptions options) { + Channel grpcChannel = options.createGrpcChannel(); + builder.grpcChannel(grpcChannel); + } +} \ No newline at end of file diff --git a/azuremanaged/src/main/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerClientOptions.java b/azuremanaged/src/main/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerClientOptions.java new file mode 100644 index 00000000..e2bca74c --- /dev/null +++ b/azuremanaged/src/main/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerClientOptions.java @@ -0,0 +1,256 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.microsoft.durabletask.azuremanaged; + +import com.azure.core.credential.TokenCredential; +import com.azure.core.credential.TokenRequestContext; +import io.grpc.*; +import java.net.MalformedURLException; +import java.time.Duration; +import java.util.Objects; +import java.net.URL; + +/** + * Options for configuring the Durable Task Scheduler. + */ +public class DurableTaskSchedulerClientOptions { + private String endpointAddress = ""; + private String taskHubName = ""; + + private TokenCredential credential; + private String resourceId = "https://durabletask.io"; + private boolean allowInsecureCredentials = false; + private Duration tokenRefreshMargin = Duration.ofMinutes(5); + + /** + * Creates a new instance of DurableTaskSchedulerClientOptions. + */ + public DurableTaskSchedulerClientOptions() { + } + + /** + * Creates a new instance of DurableTaskSchedulerClientOptions from a connection string. + * + * @param connectionString The connection string to parse. + * @return A new DurableTaskSchedulerClientOptions object. + */ + public static DurableTaskSchedulerClientOptions fromConnectionString(String connectionString) { + DurableTaskSchedulerConnectionString parsedConnectionString = new DurableTaskSchedulerConnectionString(connectionString); + return fromConnectionString(parsedConnectionString); + } + + /** + * Creates a new instance of DurableTaskSchedulerClientOptions from a parsed connection string. + * + * @param connectionString The parsed connection string. + * @return A new DurableTaskSchedulerClientOptions object. + */ + static DurableTaskSchedulerClientOptions fromConnectionString(DurableTaskSchedulerConnectionString connectionString) { + // TODO: Parse different credential types from connection string + DurableTaskSchedulerClientOptions options = new DurableTaskSchedulerClientOptions(); + options.setEndpointAddress(connectionString.getEndpoint()); + options.setTaskHubName(connectionString.getTaskHubName()); + options.setCredential(connectionString.getCredential()); + options.setAllowInsecureCredentials(options.getCredential() == null); + return options; + } + + /** + * Gets the endpoint address. + * + * @return The endpoint address. + */ + public String getEndpointAddress() { + return endpointAddress; + } + + /** + * Sets the endpoint address. + * + * @param endpointAddress The endpoint address. + * @return This options object. + */ + public DurableTaskSchedulerClientOptions setEndpointAddress(String endpointAddress) { + this.endpointAddress = endpointAddress; + return this; + } + + /** + * Gets the task hub name. + * + * @return The task hub name. + */ + public String getTaskHubName() { + return taskHubName; + } + + /** + * Sets the task hub name. + * + * @param taskHubName The task hub name. + * @return This options object. + */ + public DurableTaskSchedulerClientOptions setTaskHubName(String taskHubName) { + this.taskHubName = taskHubName; + return this; + } + + /** + * Gets the credential used for authentication. + * + * @return The credential. + */ + public TokenCredential getCredential() { + return credential; + } + + /** + * Sets the credential used for authentication. + * + * @param credential The credential. + * @return This options object. + */ + public DurableTaskSchedulerClientOptions setCredential(TokenCredential credential) { + this.credential = credential; + return this; + } + + /** + * Gets the resource ID. + * + * @return The resource ID. + */ + public String getResourceId() { + return resourceId; + } + + /** + * Sets the resource ID. + * + * @param resourceId The resource ID. + * @return This options object. + */ + public DurableTaskSchedulerClientOptions setResourceId(String resourceId) { + this.resourceId = resourceId; + return this; + } + + /** + * Gets whether insecure credentials are allowed. + * + * @return True if insecure credentials are allowed. + */ + public boolean isAllowInsecureCredentials() { + return allowInsecureCredentials; + } + + /** + * Sets whether insecure credentials are allowed. + * + * @param allowInsecureCredentials True to allow insecure credentials. + * @return This options object. + */ + public DurableTaskSchedulerClientOptions setAllowInsecureCredentials(boolean allowInsecureCredentials) { + this.allowInsecureCredentials = allowInsecureCredentials; + return this; + } + + /** + * Gets the token refresh margin. + * + * @return The token refresh margin. + */ + public Duration getTokenRefreshMargin() { + return tokenRefreshMargin; + } + + /** + * Sets the token refresh margin. + * + * @param tokenRefreshMargin The token refresh margin. + * @return This options object. + */ + public DurableTaskSchedulerClientOptions setTokenRefreshMargin(Duration tokenRefreshMargin) { + this.tokenRefreshMargin = tokenRefreshMargin; + return this; + } + + /** + * Creates a gRPC channel using the configured options. + * + * @return A configured gRPC channel for communication with the Durable Task service. + */ + public Channel createGrpcChannel() { + // Create token cache only if credential is not null + AccessTokenCache tokenCache = null; + if (credential != null) { + TokenRequestContext context = new TokenRequestContext(); + context.addScopes(new String[] { this.resourceId + "/.default" }); + tokenCache = new AccessTokenCache(this.credential, context, this.tokenRefreshMargin); + } + + // Parse and normalize the endpoint URL + String endpoint = endpointAddress; + // Add https:// prefix if no protocol is specified + if (!endpoint.startsWith("http://") && !endpoint.startsWith("https://")) { + endpoint = "https://" + endpoint; + } + + URL url; + try { + url = new URL(endpoint); + } catch (MalformedURLException e) { + throw new IllegalArgumentException("Invalid endpoint URL: " + endpoint); + } + + String authority = url.getHost(); + if (url.getPort() != -1) { + authority += ":" + url.getPort(); + } + + // Create metadata interceptor to add task hub name and auth token + AccessTokenCache finalTokenCache = tokenCache; + ClientInterceptor metadataInterceptor = new ClientInterceptor() { + @Override + public ClientCall interceptCall( + MethodDescriptor method, + CallOptions callOptions, + Channel next) { + return new ForwardingClientCall.SimpleForwardingClientCall( + next.newCall(method, callOptions)) { + @Override + public void start(ClientCall.Listener responseListener, Metadata headers) { + headers.put( + Metadata.Key.of("taskhub", Metadata.ASCII_STRING_MARSHALLER), + taskHubName + ); + + // Add authorization token if credentials are configured + if (finalTokenCache != null) { + String token = finalTokenCache.getToken().getToken(); + headers.put( + Metadata.Key.of("Authorization", Metadata.ASCII_STRING_MARSHALLER), + "Bearer " + token + ); + } + + super.start(responseListener, headers); + } + }; + } + }; + + ChannelCredentials credentials; + if (!this.allowInsecureCredentials) { + credentials = io.grpc.TlsChannelCredentials.create(); + } else { + credentials = InsecureChannelCredentials.create(); + } + + // Create channel with credentials + return Grpc.newChannelBuilder(authority, credentials) + .intercept(metadataInterceptor) + .build(); + } +} \ No newline at end of file diff --git a/azuremanaged/src/main/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerConnectionString.java b/azuremanaged/src/main/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerConnectionString.java new file mode 100644 index 00000000..eb2e622c --- /dev/null +++ b/azuremanaged/src/main/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerConnectionString.java @@ -0,0 +1,199 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.microsoft.durabletask.azuremanaged; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import javax.annotation.Nullable; + +import com.azure.core.credential.TokenCredential; +import com.azure.identity.AzureCliCredentialBuilder; +import com.azure.identity.AzurePowerShellCredentialBuilder; +import com.azure.identity.DefaultAzureCredentialBuilder; +import com.azure.identity.EnvironmentCredentialBuilder; +import com.azure.identity.IntelliJCredentialBuilder; +import com.azure.identity.InteractiveBrowserCredentialBuilder; +import com.azure.identity.ManagedIdentityCredentialBuilder; +import com.azure.identity.VisualStudioCodeCredentialBuilder; +import com.azure.identity.WorkloadIdentityCredentialBuilder; + +/** + * Represents the constituent parts of a connection string for a Durable Task Scheduler service. + */ +public class DurableTaskSchedulerConnectionString { + private final Map properties; + + /** + * Initializes a new instance of the DurableTaskSchedulerConnectionString class. + * + * @param connectionString A connection string for a Durable Task Scheduler service. + * @throws IllegalArgumentException If the connection string is invalid or missing required properties. + */ + public DurableTaskSchedulerConnectionString(String connectionString) { + if (connectionString == null || connectionString.trim().isEmpty()) { + throw new IllegalArgumentException("connectionString must not be null or empty"); + } + this.properties = parseConnectionString(connectionString); + + // Validate required properties + this.getAuthentication(); + this.getTaskHubName(); + this.getEndpoint(); + } + + /** + * Gets the authentication method specified in the connection string. + * + * @return The authentication method. + */ + public String getAuthentication() { + return getRequiredValue("Authentication"); + } + + /** + * Gets the managed identity or workload identity client ID specified in the connection string. + * + * @return The client ID, or null if not specified. + */ + public String getClientId() { + return getValue("ClientID"); + } + + /** + * Gets the "AdditionallyAllowedTenants" property, optionally used by Workload Identity. + * Multiple values can be separated by a comma. + * + * @return List of allowed tenants, or null if not specified. + */ + public List getAdditionallyAllowedTenants() { + String value = getValue("AdditionallyAllowedTenants"); + if (value == null || value.isEmpty()) { + return null; + } + return Arrays.asList(value.split(",")); + } + + /** + * Gets the "TenantId" property, optionally used by Workload Identity. + * + * @return The tenant ID, or null if not specified. + */ + public String getTenantId() { + return getValue("TenantId"); + } + + /** + * Gets the "TokenFilePath" property, optionally used by Workload Identity. + * + * @return The token file path, or null if not specified. + */ + public String getTokenFilePath() { + return getValue("TokenFilePath"); + } + + /** + * Gets the endpoint specified in the connection string. + * + * @return The endpoint URL. + */ + public String getEndpoint() { + return getRequiredValue("Endpoint"); + } + + /** + * Gets the task hub name specified in the connection string. + * + * @return The task hub name. + */ + public String getTaskHubName() { + return getRequiredValue("TaskHub"); + } + + private String getValue(String name) { + return properties.get(name); + } + + private String getRequiredValue(String name) { + String value = getValue(name); + if (value == null || value.isEmpty()) { + throw new IllegalArgumentException("The connection string must contain a " + name + " property"); + } + return value; + } + + private static Map parseConnectionString(String connectionString) { + Map properties = new HashMap<>(); + + String[] pairs = connectionString.split(";"); + for (String pair : pairs) { + int equalsIndex = pair.indexOf('='); + if (equalsIndex > 0) { + String key = pair.substring(0, equalsIndex).trim(); + String value = pair.substring(equalsIndex + 1).trim(); + properties.put(key, value); + } + } + + return properties; + } + + /** + * Gets a TokenCredential based on the authentication type specified in the connection string. + * + * @return A TokenCredential instance based on the specified authentication type, or null if authentication type is "none". + * @throws IllegalArgumentException If the connection string contains an unsupported authentication type. + */ + public @Nullable TokenCredential getCredential() { + String authType = getAuthentication(); + + // Parse the supported auth types in a case-insensitive way + switch (authType.toLowerCase().trim()) { + case "defaultazure": + return new DefaultAzureCredentialBuilder().build(); + case "managedidentity": + return new ManagedIdentityCredentialBuilder().clientId(getClientId()).build(); + case "workloadidentity": + WorkloadIdentityCredentialBuilder builder = new WorkloadIdentityCredentialBuilder(); + if (getClientId() != null && !getClientId().isEmpty()) { + builder.clientId(getClientId()); + } + + if (getTenantId() != null && !getTenantId().isEmpty()) { + builder.tenantId(getTenantId()); + } + + if (getTokenFilePath() != null && !getTokenFilePath().isEmpty()) { + builder.tokenFilePath(getTokenFilePath()); + } + + if (getAdditionallyAllowedTenants() != null) { + for (String tenant : getAdditionallyAllowedTenants()) { + builder.additionallyAllowedTenants(tenant); + } + } + + return builder.build(); + case "environment": + return new EnvironmentCredentialBuilder().build(); + case "azurecli": + return new AzureCliCredentialBuilder().build(); + case "azurepowershell": + return new AzurePowerShellCredentialBuilder().build(); + case "visualstudiocode": + return new VisualStudioCodeCredentialBuilder().build(); + case "intellij": + return new IntelliJCredentialBuilder().build(); + case "interactivebrowser": + return new InteractiveBrowserCredentialBuilder().build(); + case "none": + return null; + default: + throw new IllegalArgumentException( + String.format("The connection string contains an unsupported authentication type '%s'.", authType)); + } + } +} \ No newline at end of file diff --git a/azuremanaged/src/main/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerWorkerExtensions.java b/azuremanaged/src/main/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerWorkerExtensions.java new file mode 100644 index 00000000..0ea9b5be --- /dev/null +++ b/azuremanaged/src/main/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerWorkerExtensions.java @@ -0,0 +1,107 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.microsoft.durabletask.azuremanaged; + +import com.azure.core.credential.TokenCredential; +import com.microsoft.durabletask.DurableTaskGrpcWorkerBuilder; + +import io.grpc.Channel; +import java.util.Objects; +import javax.annotation.Nullable; + +/** + * Extension methods for creating DurableTaskWorker instances that connect to Azure-managed Durable Task Scheduler. + * This class provides various methods to create and configure workers using either connection strings or explicit parameters. + */ +public final class DurableTaskSchedulerWorkerExtensions { + private DurableTaskSchedulerWorkerExtensions() {} + + /** + * Configures a DurableTaskGrpcWorkerBuilder to use Azure-managed Durable Task Scheduler with a connection string. + * + * @param builder The builder to configure. + * @param connectionString The connection string for Azure-managed Durable Task Scheduler. + * @throws NullPointerException if builder or connectionString is null + */ + public static void useDurableTaskScheduler( + DurableTaskGrpcWorkerBuilder builder, + String connectionString) { + Objects.requireNonNull(builder, "builder must not be null"); + Objects.requireNonNull(connectionString, "connectionString must not be null"); + + configureBuilder(builder, + DurableTaskSchedulerWorkerOptions.fromConnectionString(connectionString)); + } + + /** + * Configures a DurableTaskGrpcWorkerBuilder to use Azure-managed Durable Task Scheduler with explicit parameters. + * + * @param builder The builder to configure. + * @param endpoint The endpoint address for Azure-managed Durable Task Scheduler. + * @param taskHubName The name of the task hub to connect to. + * @param tokenCredential The token credential for authentication, or null for anonymous access. + * @throws NullPointerException if builder, endpoint, or taskHubName is null + */ + public static void useDurableTaskScheduler( + DurableTaskGrpcWorkerBuilder builder, + String endpoint, + String taskHubName, + @Nullable TokenCredential tokenCredential) { + Objects.requireNonNull(builder, "builder must not be null"); + Objects.requireNonNull(endpoint, "endpoint must not be null"); + Objects.requireNonNull(taskHubName, "taskHubName must not be null"); + + configureBuilder(builder, new DurableTaskSchedulerWorkerOptions() + .setEndpointAddress(endpoint) + .setTaskHubName(taskHubName) + .setCredential(tokenCredential)); + } + + /** + * Creates a DurableTaskGrpcWorkerBuilder configured for Azure-managed Durable Task Scheduler using a connection string. + * + * @param connectionString The connection string for Azure-managed Durable Task Scheduler. + * @return A new configured DurableTaskGrpcWorkerBuilder instance. + * @throws NullPointerException if connectionString is null + */ + public static DurableTaskGrpcWorkerBuilder createWorkerBuilder( + String connectionString) { + Objects.requireNonNull(connectionString, "connectionString must not be null"); + return createBuilderFromOptions( + DurableTaskSchedulerWorkerOptions.fromConnectionString(connectionString)); + } + + /** + * Creates a DurableTaskGrpcWorkerBuilder configured for Azure-managed Durable Task Scheduler using explicit parameters. + * + * @param endpoint The endpoint address for Azure-managed Durable Task Scheduler. + * @param taskHubName The name of the task hub to connect to. + * @param tokenCredential The token credential for authentication, or null for anonymous access. + * @return A new configured DurableTaskGrpcWorkerBuilder instance. + * @throws NullPointerException if endpoint or taskHubName is null + */ + public static DurableTaskGrpcWorkerBuilder createWorkerBuilder( + String endpoint, + String taskHubName, + @Nullable TokenCredential tokenCredential) { + Objects.requireNonNull(endpoint, "endpoint must not be null"); + Objects.requireNonNull(taskHubName, "taskHubName must not be null"); + + return createBuilderFromOptions(new DurableTaskSchedulerWorkerOptions() + .setEndpointAddress(endpoint) + .setTaskHubName(taskHubName) + .setCredential(tokenCredential)); + } + + // Private helper methods to reduce code duplication + private static DurableTaskGrpcWorkerBuilder createBuilderFromOptions(DurableTaskSchedulerWorkerOptions options) { + Channel grpcChannel = options.createGrpcChannel(); + return new DurableTaskGrpcWorkerBuilder().grpcChannel(grpcChannel); + } + + private static void configureBuilder(DurableTaskGrpcWorkerBuilder builder, DurableTaskSchedulerWorkerOptions options) { + Channel grpcChannel = options.createGrpcChannel(); + builder.grpcChannel(grpcChannel); + } +} \ No newline at end of file diff --git a/azuremanaged/src/main/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerWorkerOptions.java b/azuremanaged/src/main/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerWorkerOptions.java new file mode 100644 index 00000000..03f9d60a --- /dev/null +++ b/azuremanaged/src/main/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerWorkerOptions.java @@ -0,0 +1,266 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.microsoft.durabletask.azuremanaged; + +import com.azure.core.credential.TokenCredential; +import com.azure.core.credential.TokenRequestContext; +import io.grpc.Channel; +import io.grpc.ChannelCredentials; +import io.grpc.Grpc; +import io.grpc.InsecureChannelCredentials; +import io.grpc.TlsChannelCredentials; +import io.grpc.ClientInterceptor; +import io.grpc.ClientCall; +import io.grpc.Metadata; +import io.grpc.MethodDescriptor; +import io.grpc.CallOptions; +import io.grpc.ForwardingClientCall; + +import java.time.Duration; +import java.util.Objects; +import java.net.URL; +import java.net.MalformedURLException; + +/** + * Options for configuring the Durable Task Scheduler worker. + */ +public class DurableTaskSchedulerWorkerOptions { + private String endpointAddress = ""; + private String taskHubName = ""; + + private TokenCredential credential; + private String resourceId = "https://durabletask.io"; + private boolean allowInsecureCredentials = false; + private Duration tokenRefreshMargin = Duration.ofMinutes(5); + + /** + * Creates a new instance of DurableTaskSchedulerWorkerOptions. + */ + public DurableTaskSchedulerWorkerOptions() { + } + + /** + * Creates a new instance of DurableTaskSchedulerWorkerOptions from a connection string. + * + * @param connectionString The connection string to parse. + * @return A new DurableTaskSchedulerWorkerOptions object. + */ + public static DurableTaskSchedulerWorkerOptions fromConnectionString(String connectionString) { + DurableTaskSchedulerConnectionString parsedConnectionString = new DurableTaskSchedulerConnectionString(connectionString); + return fromConnectionString(parsedConnectionString); + } + + /** + * Creates a new instance of DurableTaskSchedulerWorkerOptions from a parsed connection string. + * + * @param connectionString The parsed connection string. + * @return A new DurableTaskSchedulerWorkerOptions object. + */ + static DurableTaskSchedulerWorkerOptions fromConnectionString(DurableTaskSchedulerConnectionString connectionString) { + DurableTaskSchedulerWorkerOptions options = new DurableTaskSchedulerWorkerOptions(); + options.setEndpointAddress(connectionString.getEndpoint()); + options.setTaskHubName(connectionString.getTaskHubName()); + options.setCredential(connectionString.getCredential()); + options.setAllowInsecureCredentials(options.getCredential() == null); + return options; + } + + /** + * Gets the endpoint address. + * + * @return The endpoint address. + */ + public String getEndpointAddress() { + return endpointAddress; + } + + /** + * Sets the endpoint address. + * + * @param endpointAddress The endpoint address. + * @return This options object. + */ + public DurableTaskSchedulerWorkerOptions setEndpointAddress(String endpointAddress) { + this.endpointAddress = endpointAddress; + return this; + } + + /** + * Gets the task hub name. + * + * @return The task hub name. + */ + public String getTaskHubName() { + return taskHubName; + } + + /** + * Sets the task hub name. + * + * @param taskHubName The task hub name. + * @return This options object. + */ + public DurableTaskSchedulerWorkerOptions setTaskHubName(String taskHubName) { + this.taskHubName = taskHubName; + return this; + } + + /** + * Gets the credential used for authentication. + * + * @return The credential. + */ + public TokenCredential getCredential() { + return credential; + } + + /** + * Sets the credential used for authentication. + * + * @param credential The credential. + * @return This options object. + */ + public DurableTaskSchedulerWorkerOptions setCredential(TokenCredential credential) { + this.credential = credential; + return this; + } + + /** + * Gets the resource ID. + * + * @return The resource ID. + */ + public String getResourceId() { + return resourceId; + } + + /** + * Sets the resource ID. + * + * @param resourceId The resource ID. + * @return This options object. + */ + public DurableTaskSchedulerWorkerOptions setResourceId(String resourceId) { + this.resourceId = resourceId; + return this; + } + + /** + * Gets whether insecure credentials are allowed. + * + * @return True if insecure credentials are allowed. + */ + public boolean isAllowInsecureCredentials() { + return allowInsecureCredentials; + } + + /** + * Sets whether insecure credentials are allowed. + * + * @param allowInsecureCredentials True to allow insecure credentials. + * @return This options object. + */ + public DurableTaskSchedulerWorkerOptions setAllowInsecureCredentials(boolean allowInsecureCredentials) { + this.allowInsecureCredentials = allowInsecureCredentials; + return this; + } + + /** + * Gets the token refresh margin. + * + * @return The token refresh margin. + */ + public Duration getTokenRefreshMargin() { + return tokenRefreshMargin; + } + + /** + * Sets the token refresh margin. + * + * @param tokenRefreshMargin The token refresh margin. + * @return This options object. + */ + public DurableTaskSchedulerWorkerOptions setTokenRefreshMargin(Duration tokenRefreshMargin) { + this.tokenRefreshMargin = tokenRefreshMargin; + return this; + } + + /** + * Creates a gRPC channel using the configured options. + * + * @return A configured gRPC channel. + */ + public Channel createGrpcChannel() { + // Create token cache only if credential is not null + AccessTokenCache tokenCache = null; + if (credential != null) { + TokenRequestContext context = new TokenRequestContext(); + context.addScopes(new String[] { this.resourceId + "/.default" }); + tokenCache = new AccessTokenCache(this.credential, context, this.tokenRefreshMargin); + } + + // Parse and normalize the endpoint URL + String endpoint = endpointAddress; + // Add https:// prefix if no protocol is specified + if (!endpoint.startsWith("http://") && !endpoint.startsWith("https://")) { + endpoint = "https://" + endpoint; + } + + URL url; + try { + url = new URL(endpoint); + } catch (MalformedURLException e) { + throw new IllegalArgumentException("Invalid endpoint URL: " + endpoint); + } + String authority = url.getHost(); + if (url.getPort() != -1) { + authority += ":" + url.getPort(); + } + + // Create metadata interceptor to add task hub name and auth token + AccessTokenCache finalTokenCache = tokenCache; + ClientInterceptor metadataInterceptor = new ClientInterceptor() { + @Override + public ClientCall interceptCall( + MethodDescriptor method, + CallOptions callOptions, + Channel next) { + return new ForwardingClientCall.SimpleForwardingClientCall( + next.newCall(method, callOptions)) { + @Override + public void start(ClientCall.Listener responseListener, Metadata headers) { + headers.put( + Metadata.Key.of("taskhub", Metadata.ASCII_STRING_MARSHALLER), + taskHubName + ); + + // Add authorization token if credentials are configured + if (finalTokenCache != null) { + String token = finalTokenCache.getToken().getToken(); + headers.put( + Metadata.Key.of("Authorization", Metadata.ASCII_STRING_MARSHALLER), + "Bearer " + token + ); + } + + super.start(responseListener, headers); + } + }; + } + }; + + // Configure channel credentials based on endpoint protocol + ChannelCredentials credentials; + if (endpoint.toLowerCase().startsWith("https://")) { + credentials = TlsChannelCredentials.create(); + } else { + credentials = InsecureChannelCredentials.create(); + } + + // Create channel with credentials + return Grpc.newChannelBuilder(authority, credentials) + .intercept(metadataInterceptor) + .build(); + } +} \ No newline at end of file diff --git a/azuremanaged/src/test/java/com/microsoft/durabletask/azuremanaged/AccessTokenCacheTest.java b/azuremanaged/src/test/java/com/microsoft/durabletask/azuremanaged/AccessTokenCacheTest.java new file mode 100644 index 00000000..f900f2e9 --- /dev/null +++ b/azuremanaged/src/test/java/com/microsoft/durabletask/azuremanaged/AccessTokenCacheTest.java @@ -0,0 +1,116 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. + +package com.microsoft.durabletask.azuremanaged; + +import com.azure.core.credential.AccessToken; +import com.azure.core.credential.TokenCredential; +import com.azure.core.credential.TokenRequestContext; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; +import reactor.core.publisher.Mono; + +import java.time.Duration; +import java.time.OffsetDateTime; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.*; + +/** + * Unit tests for {@link AccessTokenCache}. + */ +@ExtendWith(MockitoExtension.class) +public class AccessTokenCacheTest { + + @Mock + private TokenCredential mockCredential; + + private TokenRequestContext context; + private Duration margin; + private AccessTokenCache tokenCache; + + @BeforeEach + public void setup() { + context = new TokenRequestContext().addScopes("https://durabletask.io/.default"); + margin = Duration.ofMinutes(5); + tokenCache = new AccessTokenCache(mockCredential, context, margin); + } + + @Test + @DisplayName("getToken should fetch a new token when cache is empty") + public void getToken_WhenCacheEmpty_FetchesNewToken() { + // Arrange + AccessToken expectedToken = new AccessToken("token1", OffsetDateTime.now().plusHours(1)); + when(mockCredential.getToken(any(TokenRequestContext.class))).thenReturn(Mono.just(expectedToken)); + + // Act + AccessToken result = tokenCache.getToken(); + + // Assert + assertEquals(expectedToken, result); + verify(mockCredential, times(1)).getToken(context); + } + + @Test + @DisplayName("getToken should reuse cached token when not expired") + public void getToken_WhenTokenNotExpired_ReusesCachedToken() { + // Arrange + AccessToken expectedToken = new AccessToken("token1", OffsetDateTime.now().plusHours(1)); + when(mockCredential.getToken(any(TokenRequestContext.class))).thenReturn(Mono.just(expectedToken)); + + // Act + tokenCache.getToken(); // First call to cache the token + AccessToken result = tokenCache.getToken(); // Second call should use cached token + + // Assert + assertEquals(expectedToken, result); + verify(mockCredential, times(1)).getToken(context); // Should only be called once + } + + @Test + @DisplayName("getToken should fetch a new token when current token is expired") + public void getToken_WhenTokenExpired_FetchesNewToken() { + // Arrange + AccessToken expiredToken = new AccessToken("expired", OffsetDateTime.now().minusMinutes(1)); + AccessToken newToken = new AccessToken("new", OffsetDateTime.now().plusHours(1)); + + when(mockCredential.getToken(any(TokenRequestContext.class))) + .thenReturn(Mono.just(expiredToken)) + .thenReturn(Mono.just(newToken)); + + // Act + AccessToken firstResult = tokenCache.getToken(); + AccessToken secondResult = tokenCache.getToken(); + + // Assert + assertEquals(expiredToken, firstResult); + assertEquals(newToken, secondResult); + verify(mockCredential, times(2)).getToken(context); + } + + @Test + @DisplayName("getToken should fetch a new token when current token is about to expire within margin") + public void getToken_WhenTokenAboutToExpire_FetchesNewToken() { + // Arrange + AccessToken expiringToken = new AccessToken("expiring", OffsetDateTime.now().plus(margin.minusMinutes(1))); + AccessToken newToken = new AccessToken("new", OffsetDateTime.now().plusHours(1)); + + when(mockCredential.getToken(any(TokenRequestContext.class))) + .thenReturn(Mono.just(expiringToken)) + .thenReturn(Mono.just(newToken)); + + // Act + AccessToken firstResult = tokenCache.getToken(); + AccessToken secondResult = tokenCache.getToken(); + + // Assert + assertEquals(expiringToken, firstResult); + assertEquals(newToken, secondResult); + verify(mockCredential, times(2)).getToken(context); + } +} diff --git a/azuremanaged/src/test/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerClientExtensionsTest.java b/azuremanaged/src/test/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerClientExtensionsTest.java new file mode 100644 index 00000000..a486dd4d --- /dev/null +++ b/azuremanaged/src/test/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerClientExtensionsTest.java @@ -0,0 +1,153 @@ +package com.microsoft.durabletask.azuremanaged; + +import com.azure.core.credential.TokenCredential; +import com.microsoft.durabletask.DurableTaskGrpcClientBuilder; +import io.grpc.Channel; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.Mockito.*; + +/** + * Unit tests for {@link DurableTaskSchedulerClientExtensions}. + */ +@ExtendWith(MockitoExtension.class) +public class DurableTaskSchedulerClientExtensionsTest { + + private static final String VALID_CONNECTION_STRING = + "Endpoint=https://example.com;Authentication=ManagedIdentity;TaskHub=myTaskHub"; + private static final String VALID_ENDPOINT = "https://example.com"; + private static final String VALID_TASKHUB = "myTaskHub"; + + @Mock + private DurableTaskGrpcClientBuilder mockBuilder; + + @Mock + private TokenCredential mockCredential; + + @Test + @DisplayName("useDurableTaskScheduler with connection string should configure builder correctly") + public void useDurableTaskScheduler_WithConnectionString_ConfiguresBuilder() { + // Arrange + when(mockBuilder.grpcChannel(any(Channel.class))).thenReturn(mockBuilder); + + // Act + DurableTaskSchedulerClientExtensions.useDurableTaskScheduler( + mockBuilder, VALID_CONNECTION_STRING); + + // Assert + verify(mockBuilder).grpcChannel(any(Channel.class)); + } + + @Test + @DisplayName("useDurableTaskScheduler with connection string should throw for null builder") + public void useDurableTaskScheduler_WithConnectionString_ThrowsForNullBuilder() { + // Act & Assert + assertThrows(NullPointerException.class, + () -> DurableTaskSchedulerClientExtensions.useDurableTaskScheduler( + null, VALID_CONNECTION_STRING)); + } + + @Test + @DisplayName("useDurableTaskScheduler with connection string should throw for null connection string") + public void useDurableTaskScheduler_WithConnectionString_ThrowsForNullConnectionString() { + // Act & Assert + assertThrows(NullPointerException.class, + () -> DurableTaskSchedulerClientExtensions.useDurableTaskScheduler( + mockBuilder, null)); + } + + @Test + @DisplayName("useDurableTaskScheduler with explicit parameters should configure builder correctly") + public void useDurableTaskScheduler_WithExplicitParameters_ConfiguresBuilder() { + // Arrange + when(mockBuilder.grpcChannel(any(Channel.class))).thenReturn(mockBuilder); + + // Act + DurableTaskSchedulerClientExtensions.useDurableTaskScheduler( + mockBuilder, VALID_ENDPOINT, VALID_TASKHUB, mockCredential); + + // Assert + verify(mockBuilder).grpcChannel(any(Channel.class)); + } + + @Test + @DisplayName("useDurableTaskScheduler with explicit parameters should throw for null builder") + public void useDurableTaskScheduler_WithExplicitParameters_ThrowsForNullBuilder() { + // Act & Assert + assertThrows(NullPointerException.class, + () -> DurableTaskSchedulerClientExtensions.useDurableTaskScheduler( + null, VALID_ENDPOINT, VALID_TASKHUB, mockCredential)); + } + + @Test + @DisplayName("useDurableTaskScheduler with explicit parameters should throw for null endpoint") + public void useDurableTaskScheduler_WithExplicitParameters_ThrowsForNullEndpoint() { + // Act & Assert + assertThrows(NullPointerException.class, + () -> DurableTaskSchedulerClientExtensions.useDurableTaskScheduler( + mockBuilder, null, VALID_TASKHUB, mockCredential)); + } + + @Test + @DisplayName("useDurableTaskScheduler with explicit parameters should throw for null task hub name") + public void useDurableTaskScheduler_WithExplicitParameters_ThrowsForNullTaskHubName() { + // Act & Assert + assertThrows(NullPointerException.class, + () -> DurableTaskSchedulerClientExtensions.useDurableTaskScheduler( + mockBuilder, VALID_ENDPOINT, null, mockCredential)); + } + + @Test + @DisplayName("createClientBuilder with connection string should create valid builder") + public void createClientBuilder_WithConnectionString_CreatesValidBuilder() { + // Act + DurableTaskGrpcClientBuilder result = + DurableTaskSchedulerClientExtensions.createClientBuilder(VALID_CONNECTION_STRING); + + // Assert + assertNotNull(result); + } + + @Test + @DisplayName("createClientBuilder with connection string should throw for null connection string") + public void createClientBuilder_WithConnectionString_ThrowsForNullConnectionString() { + // Act & Assert + assertThrows(NullPointerException.class, + () -> DurableTaskSchedulerClientExtensions.createClientBuilder(null)); + } + + @Test + @DisplayName("createClientBuilder with explicit parameters should create valid builder") + public void createClientBuilder_WithExplicitParameters_CreatesValidBuilder() { + // Act + DurableTaskGrpcClientBuilder result = + DurableTaskSchedulerClientExtensions.createClientBuilder( + VALID_ENDPOINT, VALID_TASKHUB, mockCredential); + + // Assert + assertNotNull(result); + } + + @Test + @DisplayName("createClientBuilder with explicit parameters should throw for null endpoint") + public void createClientBuilder_WithExplicitParameters_ThrowsForNullEndpoint() { + // Act & Assert + assertThrows(NullPointerException.class, + () -> DurableTaskSchedulerClientExtensions.createClientBuilder( + null, VALID_TASKHUB, mockCredential)); + } + + @Test + @DisplayName("createClientBuilder with explicit parameters should throw for null task hub name") + public void createClientBuilder_WithExplicitParameters_ThrowsForNullTaskHubName() { + // Act & Assert + assertThrows(NullPointerException.class, + () -> DurableTaskSchedulerClientExtensions.createClientBuilder( + VALID_ENDPOINT, null, mockCredential)); + } +} \ No newline at end of file diff --git a/azuremanaged/src/test/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerClientOptionsTest.java b/azuremanaged/src/test/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerClientOptionsTest.java new file mode 100644 index 00000000..8508b56b --- /dev/null +++ b/azuremanaged/src/test/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerClientOptionsTest.java @@ -0,0 +1,151 @@ +package com.microsoft.durabletask.azuremanaged; + +import com.azure.core.credential.TokenCredential; +import io.grpc.Channel; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import java.time.Duration; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * Unit tests for {@link DurableTaskSchedulerClientOptions}. + */ +@ExtendWith(MockitoExtension.class) +public class DurableTaskSchedulerClientOptionsTest { + + private static final String VALID_CONNECTION_STRING = + "Endpoint=https://example.com;Authentication=ManagedIdentity;TaskHub=myTaskHub"; + private static final String VALID_ENDPOINT = "https://example.com"; + private static final String VALID_TASKHUB = "myTaskHub"; + private static final String CUSTOM_RESOURCE_ID = "https://custom.resource"; + private static final Duration CUSTOM_REFRESH_MARGIN = Duration.ofMinutes(10); + + @Mock + private TokenCredential mockCredential; + + @Test + @DisplayName("fromConnectionString should create valid options") + public void fromConnectionString_CreatesValidOptions() { + // Act + DurableTaskSchedulerClientOptions options = + DurableTaskSchedulerClientOptions.fromConnectionString(VALID_CONNECTION_STRING); + + // Assert + assertNotNull(options); + assertEquals(VALID_ENDPOINT, options.getEndpointAddress()); + assertEquals(VALID_TASKHUB, options.getTaskHubName()); + } + + @Test + @DisplayName("setEndpointAddress should update endpoint") + public void setEndpointAddress_UpdatesEndpoint() { + // Arrange + DurableTaskSchedulerClientOptions options = new DurableTaskSchedulerClientOptions(); + + // Act + options.setEndpointAddress(VALID_ENDPOINT); + + // Assert + assertEquals(VALID_ENDPOINT, options.getEndpointAddress()); + } + + @Test + @DisplayName("setTaskHubName should update task hub name") + public void setTaskHubName_UpdatesTaskHubName() { + // Arrange + DurableTaskSchedulerClientOptions options = new DurableTaskSchedulerClientOptions(); + + // Act + options.setTaskHubName(VALID_TASKHUB); + + // Assert + assertEquals(VALID_TASKHUB, options.getTaskHubName()); + } + + @Test + @DisplayName("setCredential should update credential") + public void setCredential_UpdatesCredential() { + // Arrange + DurableTaskSchedulerClientOptions options = new DurableTaskSchedulerClientOptions(); + + // Act + options.setCredential(mockCredential); + + // Assert + assertEquals(mockCredential, options.getCredential()); + } + + @Test + @DisplayName("setResourceId should update resource ID") + public void setResourceId_UpdatesResourceId() { + // Arrange + DurableTaskSchedulerClientOptions options = new DurableTaskSchedulerClientOptions(); + + // Act + options.setResourceId(CUSTOM_RESOURCE_ID); + + // Assert + assertEquals(CUSTOM_RESOURCE_ID, options.getResourceId()); + } + + @Test + @DisplayName("setAllowInsecureCredentials should update insecure credentials flag") + public void setAllowInsecureCredentials_UpdatesFlag() { + // Arrange + DurableTaskSchedulerClientOptions options = new DurableTaskSchedulerClientOptions(); + + // Act + options.setAllowInsecureCredentials(true); + + // Assert + assertTrue(options.isAllowInsecureCredentials()); + } + + @Test + @DisplayName("setTokenRefreshMargin should update token refresh margin") + public void setTokenRefreshMargin_UpdatesMargin() { + // Arrange + DurableTaskSchedulerClientOptions options = new DurableTaskSchedulerClientOptions(); + + // Act + options.setTokenRefreshMargin(CUSTOM_REFRESH_MARGIN); + + // Assert + assertEquals(CUSTOM_REFRESH_MARGIN, options.getTokenRefreshMargin()); + } + + @Test + @DisplayName("createGrpcChannel should create valid channel") + public void createGrpcChannel_CreatesValidChannel() { + // Arrange + DurableTaskSchedulerClientOptions options = new DurableTaskSchedulerClientOptions() + .setEndpointAddress(VALID_ENDPOINT) + .setTaskHubName(VALID_TASKHUB); + + // Act + Channel channel = options.createGrpcChannel(); + + // Assert + assertNotNull(channel); + } + + @Test + @DisplayName("createGrpcChannel should handle endpoint without protocol") + public void createGrpcChannel_HandlesEndpointWithoutProtocol() { + // Arrange + DurableTaskSchedulerClientOptions options = new DurableTaskSchedulerClientOptions() + .setEndpointAddress("example.com") + .setTaskHubName(VALID_TASKHUB); + + // Act + Channel channel = options.createGrpcChannel(); + + // Assert + assertNotNull(channel); + } +} \ No newline at end of file diff --git a/azuremanaged/src/test/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerConnectionStringTest.java b/azuremanaged/src/test/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerConnectionStringTest.java new file mode 100644 index 00000000..59d974eb --- /dev/null +++ b/azuremanaged/src/test/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerConnectionStringTest.java @@ -0,0 +1,287 @@ +package com.microsoft.durabletask.azuremanaged; + +import com.azure.core.credential.TokenCredential; +import com.azure.identity.*; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.ValueSource; +import org.junit.jupiter.params.provider.NullAndEmptySource; + +import java.util.List; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * Unit tests for {@link DurableTaskSchedulerConnectionString}. + */ +public class DurableTaskSchedulerConnectionStringTest { + + private static final String VALID_CONNECTION_STRING = + "Endpoint=https://example.com;Authentication=ManagedIdentity;TaskHub=myTaskHub"; + + @Test + @DisplayName("Constructor should parse valid connection string") + public void constructor_ParsesValidConnectionString() { + // Arrange & Act + DurableTaskSchedulerConnectionString connectionString = + new DurableTaskSchedulerConnectionString(VALID_CONNECTION_STRING); + + // Assert + assertEquals("https://example.com", connectionString.getEndpoint()); + assertEquals("ManagedIdentity", connectionString.getAuthentication()); + assertEquals("myTaskHub", connectionString.getTaskHubName()); + } + + @Test + @DisplayName("Constructor should handle connection string with whitespace") + public void constructor_HandlesWhitespace() { + // Arrange + String connectionStringWithSpaces = + "Endpoint = https://example.com ; Authentication = ManagedIdentity ; TaskHub = myTaskHub"; + + // Act + DurableTaskSchedulerConnectionString connectionString = + new DurableTaskSchedulerConnectionString(connectionStringWithSpaces); + + // Assert + assertEquals("https://example.com", connectionString.getEndpoint()); + assertEquals("ManagedIdentity", connectionString.getAuthentication()); + assertEquals("myTaskHub", connectionString.getTaskHubName()); + } + + @ParameterizedTest + @NullAndEmptySource + @ValueSource(strings = {" "}) + @DisplayName("Constructor should throw for null or empty connection string") + public void constructor_ThrowsForNullOrEmptyConnectionString(String invalidInput) { + // Act & Assert + assertThrows(IllegalArgumentException.class, + () -> new DurableTaskSchedulerConnectionString(invalidInput)); + } + + @Test + @DisplayName("Constructor should throw when missing required Endpoint property") + public void constructor_ThrowsWhenMissingEndpoint() { + // Arrange + String missingEndpoint = "Authentication=ManagedIdentity;TaskHub=myTaskHub"; + + // Act & Assert + IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, + () -> new DurableTaskSchedulerConnectionString(missingEndpoint)); + + assertTrue(exception.getMessage().contains("Endpoint")); + } + + @Test + @DisplayName("Constructor should throw when missing required Authentication property") + public void constructor_ThrowsWhenMissingAuthentication() { + // Arrange + String missingAuthentication = "Endpoint=https://example.com;TaskHub=myTaskHub"; + + // Act & Assert + IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, + () -> new DurableTaskSchedulerConnectionString(missingAuthentication)); + + assertTrue(exception.getMessage().contains("Authentication")); + } + + @Test + @DisplayName("Constructor should throw when missing required TaskHub property") + public void constructor_ThrowsWhenMissingTaskHub() { + // Arrange + String missingTaskHub = "Endpoint=https://example.com;Authentication=ManagedIdentity"; + + // Act & Assert + IllegalArgumentException exception = assertThrows(IllegalArgumentException.class, + () -> new DurableTaskSchedulerConnectionString(missingTaskHub)); + + assertTrue(exception.getMessage().contains("TaskHub")); + } + + @Test + @DisplayName("getAdditionallyAllowedTenants should return split comma-separated values") + public void getAdditionallyAllowedTenants_ShouldSplitCommaValues() { + // Arrange + String connectionString = VALID_CONNECTION_STRING + + ";AdditionallyAllowedTenants=tenant1,tenant2,tenant3"; + + // Act + DurableTaskSchedulerConnectionString parsedString = + new DurableTaskSchedulerConnectionString(connectionString); + List tenants = parsedString.getAdditionallyAllowedTenants(); + + // Assert + assertNotNull(tenants); + assertEquals(3, tenants.size()); + assertEquals("tenant1", tenants.get(0)); + assertEquals("tenant2", tenants.get(1)); + assertEquals("tenant3", tenants.get(2)); + } + + @Test + @DisplayName("getAdditionallyAllowedTenants should return null when property not present") + public void getAdditionallyAllowedTenants_ReturnsNullWhenNotPresent() { + // Arrange & Act + DurableTaskSchedulerConnectionString connectionString = + new DurableTaskSchedulerConnectionString(VALID_CONNECTION_STRING); + + // Assert + assertNull(connectionString.getAdditionallyAllowedTenants()); + } + + @Test + @DisplayName("getClientId should return correct value when present") + public void getClientId_ReturnsValueWhenPresent() { + // Arrange + String connectionString = VALID_CONNECTION_STRING + ";ClientID=my-client-id"; + + // Act + DurableTaskSchedulerConnectionString parsedString = + new DurableTaskSchedulerConnectionString(connectionString); + + // Assert + assertEquals("my-client-id", parsedString.getClientId()); + } + + @Test + @DisplayName("getTenantId should return correct value when present") + public void getTenantId_ReturnsValueWhenPresent() { + // Arrange + String connectionString = VALID_CONNECTION_STRING + ";TenantId=my-tenant-id"; + + // Act + DurableTaskSchedulerConnectionString parsedString = + new DurableTaskSchedulerConnectionString(connectionString); + + // Assert + assertEquals("my-tenant-id", parsedString.getTenantId()); + } + + @Test + @DisplayName("getTokenFilePath should return correct value when present") + public void getTokenFilePath_ReturnsValueWhenPresent() { + // Arrange + String connectionString = VALID_CONNECTION_STRING + ";TokenFilePath=/path/to/token"; + + // Act + DurableTaskSchedulerConnectionString parsedString = + new DurableTaskSchedulerConnectionString(connectionString); + + // Assert + assertEquals("/path/to/token", parsedString.getTokenFilePath()); + } + + @Test + @DisplayName("getCredential should handle supported authentication types") + public void getCredential_HandlesSupportedAuthTypes() { + // Arrange + String connectionString = String.format( + "Endpoint=%s;Authentication=%s;TaskHub=%s", + "https://example.com", "ManagedIdentity", "myTaskHub"); + + // Act + DurableTaskSchedulerConnectionString result = + new DurableTaskSchedulerConnectionString(connectionString); + + // Assert + TokenCredential credential = result.getCredential(); + assertNotNull(credential); + + // Verify the correct credential type is returned + assertTrue(credential instanceof ManagedIdentityCredential); + } + + @Test + @DisplayName("getCredential should throw for unsupported authentication type") + public void getCredential_ThrowsForUnsupportedAuthType() { + // Arrange + String connectionString = String.format( + "Endpoint=%s;Authentication=%s;TaskHub=%s", + "https://example.com", "UnsupportedType", "myTaskHub"); + DurableTaskSchedulerConnectionString result = + new DurableTaskSchedulerConnectionString(connectionString); + + // Act & Assert + assertThrows(IllegalArgumentException.class, result::getCredential); + } + + @Test + @DisplayName("getCredential should configure WorkloadIdentity with all properties") + public void getCredential_ConfiguresWorkloadIdentityWithAllProperties() { + // Arrange + String connectionString = String.format( + "Endpoint=%s;Authentication=%s;TaskHub=%s;ClientID=%s;TenantId=%s;TokenFilePath=%s;AdditionallyAllowedTenants=%s", + "https://example.com", "WorkloadIdentity", "myTaskHub", "client-id-123", "tenant-id-123", + "/path/to/token", "tenant1,tenant2,tenant3"); + + // Act + DurableTaskSchedulerConnectionString result = + new DurableTaskSchedulerConnectionString(connectionString); + TokenCredential credential = result.getCredential(); + + // Assert + assertNotNull(credential); + assertTrue(credential instanceof WorkloadIdentityCredential); + } + + @Test + @DisplayName("getCredential should return VisualStudioCodeCredential for VisualStudioCode authentication type") + public void getCredential_ReturnsVisualStudioCodeCredential() { + // Arrange + String connectionString = String.format( + "Endpoint=%s;Authentication=%s;TaskHub=%s", + "https://example.com", "VisualStudioCode", "myTaskHub"); + + // Act + DurableTaskSchedulerConnectionString result = + new DurableTaskSchedulerConnectionString(connectionString); + + // Assert + TokenCredential credential = result.getCredential(); + assertNotNull(credential); + + // Verify the correct credential type is returned + assertTrue(credential instanceof VisualStudioCodeCredential); + } + + @Test + @DisplayName("getCredential should return InteractiveBrowserCredential for InteractiveBrowser authentication type") + public void getCredential_ReturnsInteractiveBrowserCredential() { + // Arrange + String connectionString = String.format( + "Endpoint=%s;Authentication=%s;TaskHub=%s", + "https://example.com", "InteractiveBrowser", "myTaskHub"); + + // Act + DurableTaskSchedulerConnectionString result = + new DurableTaskSchedulerConnectionString(connectionString); + + // Assert + TokenCredential credential = result.getCredential(); + assertNotNull(credential); + + // Verify the correct credential type is returned + assertTrue(credential instanceof InteractiveBrowserCredential); + } + + @Test + @DisplayName("getCredential should return IntelliJCredential for IntelliJ authentication type") + public void getCredential_ReturnsIntelliJCredential() { + // Arrange + String connectionString = String.format( + "Endpoint=%s;Authentication=%s;TaskHub=%s", + "https://example.com", "IntelliJ", "myTaskHub"); + + // Act + DurableTaskSchedulerConnectionString result = + new DurableTaskSchedulerConnectionString(connectionString); + + // Assert + TokenCredential credential = result.getCredential(); + assertNotNull(credential); + + // Verify the correct credential type is returned + assertTrue(credential instanceof IntelliJCredential); + } +} \ No newline at end of file diff --git a/azuremanaged/src/test/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerWorkerExtensionsTest.java b/azuremanaged/src/test/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerWorkerExtensionsTest.java new file mode 100644 index 00000000..9ad18281 --- /dev/null +++ b/azuremanaged/src/test/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerWorkerExtensionsTest.java @@ -0,0 +1,153 @@ +package com.microsoft.durabletask.azuremanaged; + +import com.azure.core.credential.TokenCredential; +import com.microsoft.durabletask.DurableTaskGrpcWorkerBuilder; +import io.grpc.Channel; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import static org.junit.jupiter.api.Assertions.*; +import static org.mockito.Mockito.*; + +/** + * Unit tests for {@link DurableTaskSchedulerWorkerExtensions}. + */ +@ExtendWith(MockitoExtension.class) +public class DurableTaskSchedulerWorkerExtensionsTest { + + private static final String VALID_CONNECTION_STRING = + "Endpoint=https://example.com;Authentication=ManagedIdentity;TaskHub=myTaskHub"; + private static final String VALID_ENDPOINT = "https://example.com"; + private static final String VALID_TASKHUB = "myTaskHub"; + + @Mock + private DurableTaskGrpcWorkerBuilder mockBuilder; + + @Mock + private TokenCredential mockCredential; + + @Test + @DisplayName("useDurableTaskScheduler with connection string should configure builder correctly") + public void useDurableTaskScheduler_WithConnectionString_ConfiguresBuilder() { + // Arrange + when(mockBuilder.grpcChannel(any(Channel.class))).thenReturn(mockBuilder); + + // Act + DurableTaskSchedulerWorkerExtensions.useDurableTaskScheduler( + mockBuilder, VALID_CONNECTION_STRING); + + // Assert + verify(mockBuilder).grpcChannel(any(Channel.class)); + } + + @Test + @DisplayName("useDurableTaskScheduler with connection string should throw for null builder") + public void useDurableTaskScheduler_WithConnectionString_ThrowsForNullBuilder() { + // Act & Assert + assertThrows(NullPointerException.class, + () -> DurableTaskSchedulerWorkerExtensions.useDurableTaskScheduler( + null, VALID_CONNECTION_STRING)); + } + + @Test + @DisplayName("useDurableTaskScheduler with connection string should throw for null connection string") + public void useDurableTaskScheduler_WithConnectionString_ThrowsForNullConnectionString() { + // Act & Assert + assertThrows(NullPointerException.class, + () -> DurableTaskSchedulerWorkerExtensions.useDurableTaskScheduler( + mockBuilder, null)); + } + + @Test + @DisplayName("useDurableTaskScheduler with explicit parameters should configure builder correctly") + public void useDurableTaskScheduler_WithExplicitParameters_ConfiguresBuilder() { + // Arrange + when(mockBuilder.grpcChannel(any(Channel.class))).thenReturn(mockBuilder); + + // Act + DurableTaskSchedulerWorkerExtensions.useDurableTaskScheduler( + mockBuilder, VALID_ENDPOINT, VALID_TASKHUB, mockCredential); + + // Assert + verify(mockBuilder).grpcChannel(any(Channel.class)); + } + + @Test + @DisplayName("useDurableTaskScheduler with explicit parameters should throw for null builder") + public void useDurableTaskScheduler_WithExplicitParameters_ThrowsForNullBuilder() { + // Act & Assert + assertThrows(NullPointerException.class, + () -> DurableTaskSchedulerWorkerExtensions.useDurableTaskScheduler( + null, VALID_ENDPOINT, VALID_TASKHUB, mockCredential)); + } + + @Test + @DisplayName("useDurableTaskScheduler with explicit parameters should throw for null endpoint") + public void useDurableTaskScheduler_WithExplicitParameters_ThrowsForNullEndpoint() { + // Act & Assert + assertThrows(NullPointerException.class, + () -> DurableTaskSchedulerWorkerExtensions.useDurableTaskScheduler( + mockBuilder, null, VALID_TASKHUB, mockCredential)); + } + + @Test + @DisplayName("useDurableTaskScheduler with explicit parameters should throw for null task hub name") + public void useDurableTaskScheduler_WithExplicitParameters_ThrowsForNullTaskHubName() { + // Act & Assert + assertThrows(NullPointerException.class, + () -> DurableTaskSchedulerWorkerExtensions.useDurableTaskScheduler( + mockBuilder, VALID_ENDPOINT, null, mockCredential)); + } + + @Test + @DisplayName("createWorkerBuilder with connection string should create valid builder") + public void createWorkerBuilder_WithConnectionString_CreatesValidBuilder() { + // Act + DurableTaskGrpcWorkerBuilder result = + DurableTaskSchedulerWorkerExtensions.createWorkerBuilder(VALID_CONNECTION_STRING); + + // Assert + assertNotNull(result); + } + + @Test + @DisplayName("createWorkerBuilder with connection string should throw for null connection string") + public void createWorkerBuilder_WithConnectionString_ThrowsForNullConnectionString() { + // Act & Assert + assertThrows(NullPointerException.class, + () -> DurableTaskSchedulerWorkerExtensions.createWorkerBuilder(null)); + } + + @Test + @DisplayName("createWorkerBuilder with explicit parameters should create valid builder") + public void createWorkerBuilder_WithExplicitParameters_CreatesValidBuilder() { + // Act + DurableTaskGrpcWorkerBuilder result = + DurableTaskSchedulerWorkerExtensions.createWorkerBuilder( + VALID_ENDPOINT, VALID_TASKHUB, mockCredential); + + // Assert + assertNotNull(result); + } + + @Test + @DisplayName("createWorkerBuilder with explicit parameters should throw for null endpoint") + public void createWorkerBuilder_WithExplicitParameters_ThrowsForNullEndpoint() { + // Act & Assert + assertThrows(NullPointerException.class, + () -> DurableTaskSchedulerWorkerExtensions.createWorkerBuilder( + null, VALID_TASKHUB, mockCredential)); + } + + @Test + @DisplayName("createWorkerBuilder with explicit parameters should throw for null task hub name") + public void createWorkerBuilder_WithExplicitParameters_ThrowsForNullTaskHubName() { + // Act & Assert + assertThrows(NullPointerException.class, + () -> DurableTaskSchedulerWorkerExtensions.createWorkerBuilder( + VALID_ENDPOINT, null, mockCredential)); + } +} \ No newline at end of file diff --git a/azuremanaged/src/test/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerWorkerOptionsTest.java b/azuremanaged/src/test/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerWorkerOptionsTest.java new file mode 100644 index 00000000..5fab6912 --- /dev/null +++ b/azuremanaged/src/test/java/com/microsoft/durabletask/azuremanaged/DurableTaskSchedulerWorkerOptionsTest.java @@ -0,0 +1,198 @@ +package com.microsoft.durabletask.azuremanaged; + +import com.azure.core.credential.TokenCredential; +import io.grpc.Channel; +import org.junit.jupiter.api.DisplayName; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.junit.jupiter.MockitoExtension; + +import java.time.Duration; + +import static org.junit.jupiter.api.Assertions.*; + +/** + * Unit tests for {@link DurableTaskSchedulerWorkerOptions}. + */ +@ExtendWith(MockitoExtension.class) +public class DurableTaskSchedulerWorkerOptionsTest { + + private static final String VALID_ENDPOINT = "https://example.com"; + private static final String VALID_TASKHUB = "myTaskHub"; + private static final String VALID_CONNECTION_STRING = + "Endpoint=https://example.com;Authentication=ManagedIdentity;TaskHub=myTaskHub"; + + @Mock + private TokenCredential mockCredential; + + @Test + @DisplayName("Default constructor should set default values") + public void defaultConstructor_SetsDefaultValues() { + // Act + DurableTaskSchedulerWorkerOptions options = new DurableTaskSchedulerWorkerOptions(); + + // Assert + assertEquals("", options.getEndpointAddress()); + assertEquals("", options.getTaskHubName()); + assertNull(options.getCredential()); + assertEquals("https://durabletask.io", options.getResourceId()); + assertFalse(options.isAllowInsecureCredentials()); + assertEquals(Duration.ofMinutes(5), options.getTokenRefreshMargin()); + } + + @Test + @DisplayName("fromConnectionString should parse connection string correctly") + public void fromConnectionString_ParsesConnectionStringCorrectly() { + // Act + DurableTaskSchedulerWorkerOptions options = + DurableTaskSchedulerWorkerOptions.fromConnectionString(VALID_CONNECTION_STRING); + + // Assert + assertEquals(VALID_ENDPOINT, options.getEndpointAddress()); + assertEquals(VALID_TASKHUB, options.getTaskHubName()); + assertNotNull(options.getCredential()); + assertTrue(options.getCredential() instanceof com.azure.identity.ManagedIdentityCredential); + assertFalse(options.isAllowInsecureCredentials()); + } + + @Test + @DisplayName("fromConnectionString should allow insecure credentials when authentication is None") + public void fromConnectionString_AllowsInsecureCredentialsWhenAuthenticationIsNone() { + // Arrange + String connectionString = "Endpoint=https://example.com;Authentication=None;TaskHub=myTaskHub"; + + // Act + DurableTaskSchedulerWorkerOptions options = + DurableTaskSchedulerWorkerOptions.fromConnectionString(connectionString); + + // Assert + assertTrue(options.isAllowInsecureCredentials()); + assertNull(options.getCredential()); + } + + @Test + @DisplayName("setEndpointAddress should update endpoint address") + public void setEndpointAddress_UpdatesEndpointAddress() { + // Arrange + DurableTaskSchedulerWorkerOptions options = new DurableTaskSchedulerWorkerOptions(); + + // Act + DurableTaskSchedulerWorkerOptions result = options.setEndpointAddress(VALID_ENDPOINT); + + // Assert + assertEquals(VALID_ENDPOINT, options.getEndpointAddress()); + assertSame(options, result); // Builder pattern returns this + } + + @Test + @DisplayName("setTaskHubName should update task hub name") + public void setTaskHubName_UpdatesTaskHubName() { + // Arrange + DurableTaskSchedulerWorkerOptions options = new DurableTaskSchedulerWorkerOptions(); + + // Act + DurableTaskSchedulerWorkerOptions result = options.setTaskHubName(VALID_TASKHUB); + + // Assert + assertEquals(VALID_TASKHUB, options.getTaskHubName()); + assertSame(options, result); // Builder pattern returns this + } + + @Test + @DisplayName("setCredential should update credential") + public void setCredential_UpdatesCredential() { + // Arrange + DurableTaskSchedulerWorkerOptions options = new DurableTaskSchedulerWorkerOptions(); + + // Act + DurableTaskSchedulerWorkerOptions result = options.setCredential(mockCredential); + + // Assert + assertSame(mockCredential, options.getCredential()); + assertSame(options, result); // Builder pattern returns this + } + + @Test + @DisplayName("setResourceId should update resource ID") + public void setResourceId_UpdatesResourceId() { + // Arrange + DurableTaskSchedulerWorkerOptions options = new DurableTaskSchedulerWorkerOptions(); + String customResourceId = "https://custom-resource.example.com"; + + // Act + DurableTaskSchedulerWorkerOptions result = options.setResourceId(customResourceId); + + // Assert + assertEquals(customResourceId, options.getResourceId()); + assertSame(options, result); // Builder pattern returns this + } + + @Test + @DisplayName("setAllowInsecureCredentials should update allowInsecureCredentials") + public void setAllowInsecureCredentials_UpdatesAllowInsecureCredentials() { + // Arrange + DurableTaskSchedulerWorkerOptions options = new DurableTaskSchedulerWorkerOptions(); + + // Act + DurableTaskSchedulerWorkerOptions result = options.setAllowInsecureCredentials(true); + + // Assert + assertTrue(options.isAllowInsecureCredentials()); + assertSame(options, result); // Builder pattern returns this + } + + @Test + @DisplayName("setTokenRefreshMargin should update token refresh margin") + public void setTokenRefreshMargin_UpdatesTokenRefreshMargin() { + // Arrange + DurableTaskSchedulerWorkerOptions options = new DurableTaskSchedulerWorkerOptions(); + Duration customMargin = Duration.ofMinutes(10); + + // Act + DurableTaskSchedulerWorkerOptions result = options.setTokenRefreshMargin(customMargin); + + // Assert + assertEquals(customMargin, options.getTokenRefreshMargin()); + assertSame(options, result); // Builder pattern returns this + } + + @Test + @DisplayName("createGrpcChannel should create a channel") + public void createGrpcChannel_CreatesChannel() { + // Arrange + DurableTaskSchedulerWorkerOptions options = new DurableTaskSchedulerWorkerOptions() + .setEndpointAddress(VALID_ENDPOINT) + .setTaskHubName(VALID_TASKHUB); + + // Act + Channel channel = options.createGrpcChannel(); + + // Assert + assertNotNull(channel); + } + + @Test + @DisplayName("createGrpcChannel should handle endpoints without protocol") + public void createGrpcChannel_HandlesEndpointsWithoutProtocol() { + // Arrange + DurableTaskSchedulerWorkerOptions options = new DurableTaskSchedulerWorkerOptions() + .setEndpointAddress("example.com") + .setTaskHubName(VALID_TASKHUB); + + // Act & Assert + assertDoesNotThrow(() -> options.createGrpcChannel()); + } + + @Test + @DisplayName("createGrpcChannel should throw for invalid URLs") + public void createGrpcChannel_ThrowsForInvalidUrls() { + // Arrange + DurableTaskSchedulerWorkerOptions options = new DurableTaskSchedulerWorkerOptions() + .setEndpointAddress("invalid:url") + .setTaskHubName(VALID_TASKHUB); + + // Act & Assert + assertThrows(IllegalArgumentException.class, () -> options.createGrpcChannel()); + } +} \ No newline at end of file diff --git a/client/build.gradle b/client/build.gradle index 3f5ea75f..98ff6adc 100644 --- a/client/build.gradle +++ b/client/build.gradle @@ -33,7 +33,7 @@ dependencies { implementation "com.fasterxml.jackson.core:jackson-databind:${jacksonVersion}" implementation "com.fasterxml.jackson.core:jackson-annotations:${jacksonVersion}" implementation "com.fasterxml.jackson.datatype:jackson-datatype-jsr310:${jacksonVersion}" - + testImplementation(platform('org.junit:junit-bom:5.7.2')) testImplementation('org.junit.jupiter:junit-jupiter') } diff --git a/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcClient.java b/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcClient.java index 95bb984a..11152ef6 100644 --- a/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcClient.java +++ b/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcClient.java @@ -21,7 +21,7 @@ /** * Durable Task client implementation that uses gRPC to connect to a remote "sidecar" process. */ -final class DurableTaskGrpcClient extends DurableTaskClient { +public final class DurableTaskGrpcClient extends DurableTaskClient { private static final int DEFAULT_PORT = 4001; private static final Logger logger = Logger.getLogger(DurableTaskGrpcClient.class.getPackage().getName()); diff --git a/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorker.java b/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorker.java index 92e2bda4..89f87a9f 100644 --- a/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorker.java +++ b/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorker.java @@ -140,6 +140,7 @@ public void startAndBlock() { .setInstanceId(orchestratorRequest.getInstanceId()) .addAllActions(taskOrchestratorResult.getActions()) .setCustomStatus(StringValue.of(taskOrchestratorResult.getCustomStatus())) + .setCompletionToken(workItem.getCompletionToken()) .build(); this.sidecarClient.completeOrchestratorTask(response); @@ -164,7 +165,8 @@ public void startAndBlock() { ActivityResponse.Builder responseBuilder = ActivityResponse.newBuilder() .setInstanceId(activityRequest.getOrchestrationInstance().getInstanceId()) - .setTaskId(activityRequest.getTaskId()); + .setTaskId(activityRequest.getTaskId()) + .setCompletionToken(workItem.getCompletionToken()); if (output != null) { responseBuilder.setResult(StringValue.of(output)); @@ -175,6 +177,10 @@ public void startAndBlock() { } this.sidecarClient.completeActivityTask(responseBuilder.build()); + } + else if (requestType == RequestCase.HEALTHPING) + { + // No-op } else { logger.log(Level.WARNING, "Received and dropped an unknown '{0}' work-item from the sidecar.", requestType); } diff --git a/client/src/test/java/com/microsoft/durabletask/IntegrationTests.java b/client/src/test/java/com/microsoft/durabletask/IntegrationTests.java index 5e9ac5d9..467151d6 100644 --- a/client/src/test/java/com/microsoft/durabletask/IntegrationTests.java +++ b/client/src/test/java/com/microsoft/durabletask/IntegrationTests.java @@ -556,7 +556,7 @@ void restartOrchestrationThrowsException() { } -// @Test + @Test void suspendResumeOrchestration() throws TimeoutException, InterruptedException { final String orchestratorName = "suspend"; final String eventName = "MyEvent"; diff --git a/internal/durabletask-protobuf/PROTO_SOURCE_COMMIT_HASH b/internal/durabletask-protobuf/PROTO_SOURCE_COMMIT_HASH index e409e9a4..a08790e4 100644 --- a/internal/durabletask-protobuf/PROTO_SOURCE_COMMIT_HASH +++ b/internal/durabletask-protobuf/PROTO_SOURCE_COMMIT_HASH @@ -1 +1 @@ -4792f47019ab2b3e9ea979fb4af72427a4144c51 \ No newline at end of file +b1e9c96572a9fa0b81829fe0e976cd39de91877e \ No newline at end of file diff --git a/internal/durabletask-protobuf/protos/orchestrator_service.proto b/internal/durabletask-protobuf/protos/orchestrator_service.proto index 64e75281..0f7d8220 100644 --- a/internal/durabletask-protobuf/protos/orchestrator_service.proto +++ b/internal/durabletask-protobuf/protos/orchestrator_service.proto @@ -617,6 +617,30 @@ message StartNewOrchestrationAction { google.protobuf.Timestamp scheduledTime = 5; } +message AbandonActivityTaskRequest { + string completionToken = 1; +} + +message AbandonActivityTaskResponse { + // Empty. +} + +message AbandonOrchestrationTaskRequest { + string completionToken = 1; +} + +message AbandonOrchestrationTaskResponse { + // Empty. +} + +message AbandonEntityTaskRequest { + string completionToken = 1; +} + +message AbandonEntityTaskResponse { + // Empty. +} + service TaskHubSidecarService { // Sends a hello request to the sidecar service. rpc Hello(google.protobuf.Empty) returns (google.protobuf.Empty); @@ -678,6 +702,15 @@ service TaskHubSidecarService { // clean entity storage rpc CleanEntityStorage(CleanEntityStorageRequest) returns (CleanEntityStorageResponse); + + // Abandons a single work item + rpc AbandonTaskActivityWorkItem(AbandonActivityTaskRequest) returns (AbandonActivityTaskResponse); + + // Abandon an orchestration work item + rpc AbandonTaskOrchestratorWorkItem(AbandonOrchestrationTaskRequest) returns (AbandonOrchestrationTaskResponse); + + // Abandon an entity work item + rpc AbandonTaskEntityWorkItem(AbandonEntityTaskRequest) returns (AbandonEntityTaskResponse); } message GetWorkItemsRequest { diff --git a/samples/build.gradle b/samples/build.gradle index af90f719..e9f049e8 100644 --- a/samples/build.gradle +++ b/samples/build.gradle @@ -6,12 +6,28 @@ plugins { group 'io.durabletask' version = '0.1.0' +def grpcVersion = '1.59.0' archivesBaseName = 'durabletask-samples' +application { + mainClass = 'io.durabletask.samples.WebAppToDurableTaskSchedulerSample' +} + dependencies { implementation project(':client') - + implementation project(':azuremanaged') + implementation 'org.springframework.boot:spring-boot-starter-web' implementation platform("org.springframework.boot:spring-boot-dependencies:2.5.2") implementation 'org.springframework.boot:spring-boot-starter' + + // https://github.com/grpc/grpc-java#download + implementation "io.grpc:grpc-protobuf:${grpcVersion}" + implementation "io.grpc:grpc-stub:${grpcVersion}" + runtimeOnly "io.grpc:grpc-netty-shaded:${grpcVersion}" + implementation 'com.azure:azure-identity:1.15.0' + + // install lombok + annotationProcessor 'org.projectlombok:lombok:1.18.22' + compileOnly 'org.projectlombok:lombok:1.18.22' } \ No newline at end of file diff --git a/samples/src/main/java/io/durabletask/samples/WebAppToDurableTaskSchedulerSample.java b/samples/src/main/java/io/durabletask/samples/WebAppToDurableTaskSchedulerSample.java new file mode 100644 index 00000000..d09fdd69 --- /dev/null +++ b/samples/src/main/java/io/durabletask/samples/WebAppToDurableTaskSchedulerSample.java @@ -0,0 +1,187 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. +package io.durabletask.samples; + +import com.microsoft.durabletask.*; +import com.microsoft.durabletask.azuremanaged.DurableTaskSchedulerClientExtensions; +import com.microsoft.durabletask.azuremanaged.DurableTaskSchedulerWorkerExtensions; +import org.springframework.boot.SpringApplication; +import org.springframework.boot.autoconfigure.SpringBootApplication; +import org.springframework.web.bind.annotation.*; +import org.springframework.context.ConfigurableApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.boot.context.properties.ConfigurationProperties; +import org.springframework.boot.context.properties.EnableConfigurationProperties; + +@ConfigurationProperties(prefix = "durable.task") +@lombok.Data +class DurableTaskProperties { + private String endpoint; + private String taskHubName; + private String resourceId = "https://durabletask.io"; + private String connectionString; +} + +/** + * Sample Spring Boot application demonstrating Azure-managed Durable Task integration. + * This sample shows how to: + * 1. Configure Azure-managed Durable Task with Spring Boot + * 2. Create orchestrations and activities + * 3. Handle REST API endpoints for order processing + */ +@SpringBootApplication +@EnableConfigurationProperties(DurableTaskProperties.class) +public class WebAppToDurableTaskSchedulerSample { + + public static void main(String[] args) { + ConfigurableApplicationContext context = SpringApplication.run(WebAppToDurableTaskSchedulerSample.class, args); + + // Get the worker bean and start it + DurableTaskGrpcWorker worker = context.getBean(DurableTaskGrpcWorker.class); + worker.start(); + } + + @Configuration + static class DurableTaskConfig { + @Bean + public DurableTaskGrpcWorker durableTaskWorker( + DurableTaskProperties properties) { + + // Create worker using Azure-managed extensions + DurableTaskGrpcWorkerBuilder workerBuilder = DurableTaskSchedulerWorkerExtensions.createWorkerBuilder( + properties.getConnectionString()); + + // Add orchestrations using the factory pattern + workerBuilder.addOrchestration(new TaskOrchestrationFactory() { + @Override + public String getName() { return "ProcessOrderOrchestration"; } + + @Override + public TaskOrchestration create() { + return ctx -> { + // Get the order input as JSON string + String orderJson = ctx.getInput(String.class); + + // Process the order through multiple activities + boolean isValid = ctx.callActivity("ValidateOrder", orderJson, Boolean.class).await(); + if (!isValid) { + ctx.complete("{\"status\": \"FAILED\", \"message\": \"Order validation failed\"}"); + return; + } + + // Process payment + String paymentResult = ctx.callActivity("ProcessPayment", orderJson, String.class).await(); + if (!paymentResult.contains("\"success\":true")) { + ctx.complete("{\"status\": \"FAILED\", \"message\": \"Payment processing failed\"}"); + return; + } + + // Ship order + String shipmentResult = ctx.callActivity("ShipOrder", orderJson, String.class).await(); + + // Return the final result + ctx.complete("{\"status\": \"SUCCESS\", " + + "\"payment\": " + paymentResult + ", " + + "\"shipment\": " + shipmentResult + "}"); + }; + } + }); + + // Add activities using the factory pattern + workerBuilder.addActivity(new TaskActivityFactory() { + @Override + public String getName() { return "ValidateOrder"; } + + @Override + public TaskActivity create() { + return ctx -> { + String orderJson = ctx.getInput(String.class); + // Simple validation - check if order contains amount and it's greater than 0 + return orderJson.contains("\"amount\"") && !orderJson.contains("\"amount\":0"); + }; + } + }); + + workerBuilder.addActivity(new TaskActivityFactory() { + @Override + public String getName() { return "ProcessPayment"; } + + @Override + public TaskActivity create() { + return ctx -> { + // Simulate payment processing + sleep(1000); // Simulate processing time + return "{\"success\":true, \"transactionId\":\"TXN" + System.currentTimeMillis() + "\"}"; + }; + } + }); + + workerBuilder.addActivity(new TaskActivityFactory() { + @Override + public String getName() { return "ShipOrder"; } + + @Override + public TaskActivity create() { + return ctx -> { + // Simulate shipping process + sleep(1000); // Simulate processing time + return "{\"trackingNumber\":\"TRACK" + System.currentTimeMillis() + "\"}"; + }; + } + }); + + return workerBuilder.build(); + } + + @Bean + public DurableTaskClient durableTaskClient( + DurableTaskProperties properties) { + + // Create client using Azure-managed extensions + return DurableTaskSchedulerClientExtensions.createClientBuilder(properties.getConnectionString()).build(); + } + } + + private static void sleep(int millis) { + try { + Thread.sleep(millis); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } +} + +/** + * REST Controller for handling order-related operations. + */ +@RestController +@RequestMapping("/api/orders") +class OrderController { + + private final DurableTaskClient client; + + public OrderController(DurableTaskClient client) { + this.client = client; + } + + @PostMapping + public String createOrder(@RequestBody String orderJson) throws Exception { + String instanceId = client.scheduleNewOrchestrationInstance( + "ProcessOrderOrchestration", + orderJson + ); + + // Return the instance ID immediately without waiting for completion + return "{\"instanceId\": \"" + instanceId + "\"}"; + } + + @GetMapping("/{instanceId}") + public String getOrder(@PathVariable String instanceId) throws Exception { + OrchestrationMetadata metadata = client.getInstanceMetadata(instanceId, true); + if (metadata == null) { + return "{\"error\": \"Order not found\"}"; + } + return metadata.readOutputAs(String.class); + } +} \ No newline at end of file diff --git a/samples/src/main/resources/application.properties b/samples/src/main/resources/application.properties new file mode 100644 index 00000000..171752e1 --- /dev/null +++ b/samples/src/main/resources/application.properties @@ -0,0 +1,8 @@ +durable.task.endpoint=https://dtwbwuks01-gqa9c3ccgbgd.uksouth.durabletask.io +durable.task.taskHubName=dtwbwuks01th01 +durable.task.connection-string=Endpoint=https://dtwbwuks01-gqa9c3ccgbgd.uksouth.durabletask.io;TaskHub=dtwbwuks01th01;Authentication=DefaultAzure; +# Add logging configuration +logging.level.com.microsoft.durabletask=DEBUG +logging.level.root=INFO + +server.port=8082 \ No newline at end of file diff --git a/samples/src/main/resources/web-app-to-durable-task-scheduler-sample.http b/samples/src/main/resources/web-app-to-durable-task-scheduler-sample.http new file mode 100644 index 00000000..9bb87f76 --- /dev/null +++ b/samples/src/main/resources/web-app-to-durable-task-scheduler-sample.http @@ -0,0 +1,60 @@ +### Variables +@instanceId = c32752e9-ae3e-42f7-9dc3-0e91277b1286 + +### Create a new order +POST http://localhost:8082/api/orders +Content-Type: application/json + +{ + "orderId": "ORD123456", + "customerId": "CUST789", + "amount": 125.50, + "items": [ + { + "productId": "PROD001", + "quantity": 2, + "price": 49.99 + }, + { + "productId": "PROD002", + "quantity": 1, + "price": 25.52 + } + ], + "shippingAddress": { + "street": "123 Main St", + "city": "Seattle", + "state": "WA", + "zipCode": "98101", + "country": "USA" + } +} + +### Get order by instance ID +# Replace the instanceId variable with the actual value returned from the create order request +GET http://localhost:8082/api/orders/{{instanceId}} +Content-Type: application/json + +### Example with invalid order (amount is 0) +POST http://localhost:8082/api/orders +Content-Type: application/json + +{ + "orderId": "ORD123457", + "customerId": "CUST789", + "amount": 0, + "items": [ + { + "productId": "PROD003", + "quantity": 1, + "price": 0 + } + ], + "shippingAddress": { + "street": "123 Main St", + "city": "Seattle", + "state": "WA", + "zipCode": "98101", + "country": "USA" + } +} \ No newline at end of file diff --git a/settings.gradle b/settings.gradle index b0f523f0..3813e3ae 100644 --- a/settings.gradle +++ b/settings.gradle @@ -2,7 +2,8 @@ rootProject.name = 'durabletask-java' include ":client" include ":azurefunctions" +include ":azuremanaged" include ":samples" include ":samples-azure-functions" -include 'endtoendtests' +include ":endtoendtests"