Skip to content

Add Logger interface, AzureLoggerAdapter, and fix gRPC client stop #85

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
YunchuWang merged 12 commits into from
Feb 3, 2026
Merged

Add Logger interface, AzureLoggerAdapter, and fix gRPC client stop #85

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
6d58e1b
feat: Add AzureLoggerAdapter for integrating Azure SDK logging
torosent Feb 3, 2026
2fe9c71
fix: Await gRPC stub close in TaskHubGrpcClient stop method
torosent Feb 3, 2026
2d555c1
Update packages/durabletask-js-azuremanaged/src/azure-logger-adapter.ts
torosent Feb 3, 2026
b2a92c4
Update packages/durabletask-js/src/client/client.ts
torosent Feb 3, 2026
5026862
Update packages/durabletask-js/src/worker/task-hub-grpc-worker.ts
torosent Feb 3, 2026
80529ea
fix: Remove non-null assertion in withTimeout by initializing timeout…
torosent Feb 3, 2026
1cd0503
fix: Catch unhandled promise rejections in worker start() method
torosent Feb 3, 2026
f2dffd3
fix: Reorder stream cleanup to prevent unhandled CANCELLED errors
torosent Feb 3, 2026
9e36c64
fix: Use event-driven synchronization instead of fixed delay in stop()
torosent Feb 3, 2026
73922cd
fix: Improve error logging format to preserve stack traces
torosent Feb 3, 2026
bde01d7
Update packages/durabletask-js/src/worker/task-hub-grpc-worker.ts
torosent Feb 3, 2026
b47ff2f
feat: Enhance TaskHubGrpcClient and TaskHubGrpcWorker with options in…
torosent Feb 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion .gitignore
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,7 +5,7 @@
!.vscode/launch.json
!.vscode/extensions.json
*.code-workspace

.DS_Store
### DotnetCore ###
# .NET Core build folders
bin/
Expand Down
62 changes: 35 additions & 27 deletions examples/azure-managed-dts.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,10 @@ import { DefaultAzureCredential } from "@azure/identity";
import {
createAzureManagedClient,
createAzureManagedWorkerBuilder,
// Logger types are re-exported for convenience
// ConsoleLogger is used by default
// Use createAzureLogger to integrate with @azure/logger
createAzureLogger,
} from "../extensions/durabletask-js-azuremanaged/build";
import { ActivityContext } from "../src/task/context/activity-context";
import { OrchestrationContext } from "../src/task/context/orchestration-context";
Expand All @@ -23,6 +27,10 @@ import { Task } from "../src/task/task";

// Wrap the entire code in an immediately-invoked async function
(async () => {
// Create a logger for this example
// This uses Azure SDK's logging infrastructure - set AZURE_LOG_LEVEL=verbose to see all logs
const logger = createAzureLogger("example");

// Configuration for Azure Managed DTS
// These values should be set as environment variables
const endpoint = process.env.AZURE_DTS_ENDPOINT;
Expand All @@ -31,31 +39,31 @@ import { Task } from "../src/task/task";

// Validate configuration
if (!connectionString && (!endpoint || !taskHubName)) {
console.error(
logger.error(
"Error: Either AZURE_DTS_CONNECTION_STRING or both AZURE_DTS_ENDPOINT and AZURE_DTS_TASKHUB must be set.",
);
console.log("\nUsage:");
console.log(" Option 1: Create a .env file in the examples directory (recommended):");
console.log(
logger.info("\nUsage:");
logger.info(" Option 1: Create a .env file in the examples directory (recommended):");
logger.info(
" AZURE_DTS_CONNECTION_STRING=Endpoint=https://myservice.durabletask.io;Authentication=DefaultAzure;TaskHub=myTaskHub",
);
console.log(" or");
console.log(" AZURE_DTS_ENDPOINT=https://myservice.durabletask.io");
console.log(" AZURE_DTS_TASKHUB=myTaskHub");
console.log("\n Option 2: Set environment variables directly");
console.log(" export AZURE_DTS_CONNECTION_STRING=...");
logger.info(" or");
logger.info(" AZURE_DTS_ENDPOINT=https://myservice.durabletask.io");
logger.info(" AZURE_DTS_TASKHUB=myTaskHub");
logger.info("\n Option 2: Set environment variables directly");
logger.info(" export AZURE_DTS_CONNECTION_STRING=...");
process.exit(1);
}

// Define an activity function that greets a city
const greetCity = async (_: ActivityContext, city: string): Promise<string> => {
console.log(`Activity executing: greeting ${city}`);
logger.info(`Activity executing: greeting ${city}`);
return `Hello, ${city}!`;
};

// Define an activity function that processes work items
const processWorkItem = async (_: ActivityContext, item: string): Promise<number> => {
console.log(`Activity executing: processing ${item}`);
logger.info(`Activity executing: processing ${item}`);
// Simulate some processing time
await new Promise((resolve) => setTimeout(resolve, 500));
return item.length;
Expand Down Expand Up @@ -98,7 +106,7 @@ import { Task } from "../src/task/task";
try {
// Create client and worker using connection string or explicit parameters
if (connectionString) {
console.log("Using connection string authentication...");
logger.info("Using connection string authentication...");
client = createAzureManagedClient(connectionString);
worker = createAzureManagedWorkerBuilder(connectionString)
.addOrchestrator(sequenceOrchestrator)
Expand All @@ -107,7 +115,7 @@ import { Task } from "../src/task/task";
.addActivity(processWorkItem)
.build();
} else {
console.log("Using DefaultAzureCredential authentication...");
logger.info("Using DefaultAzureCredential authentication...");
const credential = new DefaultAzureCredential();
client = createAzureManagedClient(endpoint!, taskHubName!, credential);
worker = createAzureManagedWorkerBuilder(endpoint!, taskHubName!, credential)
Expand All @@ -119,42 +127,42 @@ import { Task } from "../src/task/task";
}

// Start the worker
console.log("Starting worker...");
logger.info("Starting worker...");
await worker.start();
console.log("Worker started successfully!");
logger.info("Worker started successfully!");

// Run the sequence orchestrator
console.log("\n--- Running Sequence Orchestrator ---");
logger.info("\n--- Running Sequence Orchestrator ---");
const sequenceId = await client.scheduleNewOrchestration(sequenceOrchestrator);
console.log(`Orchestration scheduled with ID: ${sequenceId}`);
logger.info(`Orchestration scheduled with ID: ${sequenceId}`);

const sequenceState = await client.waitForOrchestrationCompletion(sequenceId, undefined, 60);
console.log(`Sequence orchestration completed!`);
console.log(`Result: ${sequenceState?.serializedOutput}`);
logger.info(`Sequence orchestration completed!`);
logger.info(`Result: ${sequenceState?.serializedOutput}`);

// Run the fan-out/fan-in orchestrator
console.log("\n--- Running Fan-Out/Fan-In Orchestrator ---");
logger.info("\n--- Running Fan-Out/Fan-In Orchestrator ---");
const fanOutId = await client.scheduleNewOrchestration(fanOutFanInOrchestrator);
console.log(`Orchestration scheduled with ID: ${fanOutId}`);
logger.info(`Orchestration scheduled with ID: ${fanOutId}`);

const fanOutState = await client.waitForOrchestrationCompletion(fanOutId, undefined, 60);
console.log(`Fan-out/fan-in orchestration completed!`);
console.log(`Result: ${fanOutState?.serializedOutput}`);
logger.info(`Fan-out/fan-in orchestration completed!`);
logger.info(`Result: ${fanOutState?.serializedOutput}`);

console.log("\n--- All orchestrations completed successfully! ---");
logger.info("\n--- All orchestrations completed successfully! ---");
} catch (error) {
console.error("Error:", error);
logger.error("Error:", error);
process.exit(1);
} finally {
// Cleanup: stop worker and client
console.log("\nStopping worker and client...");
logger.info("\nStopping worker and client...");
if (worker) {
await worker.stop();
}
if (client) {
await client.stop();
}
console.log("Cleanup complete.");
logger.info("Cleanup complete.");
process.exit(0);
}
})();
62 changes: 35 additions & 27 deletions examples/azure-managed/index.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,10 @@ import { DefaultAzureCredential } from "@azure/identity";
import {
createAzureManagedClient,
createAzureManagedWorkerBuilder,
// Logger types are re-exported for convenience
// ConsoleLogger is used by default
// Use createAzureLogger to integrate with @azure/logger
createAzureLogger,
} from "@microsoft/durabletask-js-azuremanaged";
import { ActivityContext } from "@microsoft/durabletask-js/dist/task/context/activity-context";
import { OrchestrationContext } from "@microsoft/durabletask-js/dist/task/context/orchestration-context";
Expand All @@ -23,6 +27,10 @@ import { Task } from "@microsoft/durabletask-js/dist/task/task";

// Wrap the entire code in an immediately-invoked async function
(async () => {
// Create a logger for this example
// This uses Azure SDK's logging infrastructure - set AZURE_LOG_LEVEL=verbose to see all logs
const logger = createAzureLogger("example");

// Configuration for Azure Managed DTS
// These values should be set as environment variables
const endpoint = process.env.AZURE_DTS_ENDPOINT;
Expand All @@ -31,31 +39,31 @@ import { Task } from "@microsoft/durabletask-js/dist/task/task";

// Validate configuration
if (!connectionString && (!endpoint || !taskHubName)) {
console.error(
logger.error(
"Error: Either AZURE_DTS_CONNECTION_STRING or both AZURE_DTS_ENDPOINT and AZURE_DTS_TASKHUB must be set.",
);
console.log("\nUsage:");
console.log(" Option 1: Create a .env file in the examples directory (recommended):");
console.log(
logger.info("\nUsage:");
logger.info(" Option 1: Create a .env file in the examples directory (recommended):");
logger.info(
" AZURE_DTS_CONNECTION_STRING=Endpoint=https://myservice.durabletask.io;Authentication=DefaultAzure;TaskHub=myTaskHub",
);
console.log(" or");
console.log(" AZURE_DTS_ENDPOINT=https://myservice.durabletask.io");
console.log(" AZURE_DTS_TASKHUB=myTaskHub");
console.log("\n Option 2: Set environment variables directly");
console.log(" export AZURE_DTS_CONNECTION_STRING=...");
logger.info(" or");
logger.info(" AZURE_DTS_ENDPOINT=https://myservice.durabletask.io");
logger.info(" AZURE_DTS_TASKHUB=myTaskHub");
logger.info("\n Option 2: Set environment variables directly");
logger.info(" export AZURE_DTS_CONNECTION_STRING=...");
process.exit(1);
}

// Define an activity function that greets a city
const greetCity = async (_: ActivityContext, city: string): Promise<string> => {
console.log(`Activity executing: greeting ${city}`);
logger.info(`Activity executing: greeting ${city}`);
return `Hello, ${city}!`;
};

// Define an activity function that processes work items
const processWorkItem = async (_: ActivityContext, item: string): Promise<number> => {
console.log(`Activity executing: processing ${item}`);
logger.info(`Activity executing: processing ${item}`);
// Simulate some processing time
await new Promise((resolve) => setTimeout(resolve, 500));
return item.length;
Expand Down Expand Up @@ -98,7 +106,7 @@ import { Task } from "@microsoft/durabletask-js/dist/task/task";
try {
// Create client and worker using connection string or explicit parameters
if (connectionString) {
console.log("Using connection string authentication...");
logger.info("Using connection string authentication...");
client = createAzureManagedClient(connectionString);
worker = createAzureManagedWorkerBuilder(connectionString)
.addOrchestrator(sequenceOrchestrator)
Expand All @@ -107,7 +115,7 @@ import { Task } from "@microsoft/durabletask-js/dist/task/task";
.addActivity(processWorkItem)
.build();
} else {
console.log("Using DefaultAzureCredential authentication...");
logger.info("Using DefaultAzureCredential authentication...");
const credential = new DefaultAzureCredential();
client = createAzureManagedClient(endpoint!, taskHubName!, credential);
worker = createAzureManagedWorkerBuilder(endpoint!, taskHubName!, credential)
Expand All @@ -119,42 +127,42 @@ import { Task } from "@microsoft/durabletask-js/dist/task/task";
}

// Start the worker
console.log("Starting worker...");
logger.info("Starting worker...");
await worker.start();
console.log("Worker started successfully!");
logger.info("Worker started successfully!");

// Run the sequence orchestrator
console.log("\n--- Running Sequence Orchestrator ---");
logger.info("\n--- Running Sequence Orchestrator ---");
const sequenceId = await client.scheduleNewOrchestration(sequenceOrchestrator);
console.log(`Orchestration scheduled with ID: ${sequenceId}`);
logger.info(`Orchestration scheduled with ID: ${sequenceId}`);

const sequenceState = await client.waitForOrchestrationCompletion(sequenceId, undefined, 60);
console.log(`Sequence orchestration completed!`);
console.log(`Result: ${sequenceState?.serializedOutput}`);
logger.info(`Sequence orchestration completed!`);
logger.info(`Result: ${sequenceState?.serializedOutput}`);

// Run the fan-out/fan-in orchestrator
console.log("\n--- Running Fan-Out/Fan-In Orchestrator ---");
logger.info("\n--- Running Fan-Out/Fan-In Orchestrator ---");
const fanOutId = await client.scheduleNewOrchestration(fanOutFanInOrchestrator);
console.log(`Orchestration scheduled with ID: ${fanOutId}`);
logger.info(`Orchestration scheduled with ID: ${fanOutId}`);

const fanOutState = await client.waitForOrchestrationCompletion(fanOutId, undefined, 60);
console.log(`Fan-out/fan-in orchestration completed!`);
console.log(`Result: ${fanOutState?.serializedOutput}`);
logger.info(`Fan-out/fan-in orchestration completed!`);
logger.info(`Result: ${fanOutState?.serializedOutput}`);

console.log("\n--- All orchestrations completed successfully! ---");
logger.info("\n--- All orchestrations completed successfully! ---");
} catch (error) {
console.error("Error:", error);
logger.error("Error:", error);
process.exit(1);
} finally {
// Cleanup: stop worker and client
console.log("\nStopping worker and client...");
logger.info("\nStopping worker and client...");
if (worker) {
await worker.stop();
}
if (client) {
await client.stop();
}
console.log("Cleanup complete.");
logger.info("Cleanup complete.");
process.exit(0);
}
})();
40 changes: 33 additions & 7 deletions examples/hello-world/activity-sequence.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,14 +7,40 @@ import {
ActivityContext,
OrchestrationContext,
TOrchestrator,
// Logger types for custom logging
// ConsoleLogger (default) - logs to console
// NoOpLogger - silent mode, useful for testing
// You can also implement your own Logger interface
ConsoleLogger,
} from "@microsoft/durabletask-js";

// Wrap the entire code in an immediately-invoked async function
(async () => {
// Update the gRPC client and worker to use a local address and port
const grpcServerAddress = "localhost:4001";
const taskHubClient: TaskHubGrpcClient = new TaskHubGrpcClient(grpcServerAddress);
const taskHubWorker: TaskHubGrpcWorker = new TaskHubGrpcWorker(grpcServerAddress);

// Optional: Create a custom logger (defaults to ConsoleLogger if not provided)
// You can implement your own Logger interface to integrate with Winston, Pino, etc.
const logger = new ConsoleLogger();

// Pass the logger as the 6th parameter (after metadataGenerator)
// Parameters: hostAddress, options, useTLS, credentials, metadataGenerator, logger
const taskHubClient: TaskHubGrpcClient = new TaskHubGrpcClient(
grpcServerAddress,
undefined,
undefined,
undefined,
undefined,
logger,
);
const taskHubWorker: TaskHubGrpcWorker = new TaskHubGrpcWorker(
grpcServerAddress,
undefined,
undefined,
undefined,
undefined,
logger,
);

const hello = async (_: ActivityContext, name: string) => {
return `Hello ${name}!`;
Expand All @@ -39,22 +65,22 @@ import {
// Wrap the worker startup in a try-catch block to handle any errors during startup
try {
await taskHubWorker.start();
console.log("Worker started successfully");
logger.info("Worker started successfully");
} catch (error) {
console.error("Error starting worker:", error);
logger.error("Error starting worker:", error);
}

// Schedule a new orchestration
try {
const id = await taskHubClient.scheduleNewOrchestration(sequence);
console.log(`Orchestration scheduled with ID: ${id}`);
logger.info(`Orchestration scheduled with ID: ${id}`);

// Wait for orchestration completion
const state = await taskHubClient.waitForOrchestrationCompletion(id, undefined, 30);

console.log(`Orchestration completed! Result: ${state?.serializedOutput}`);
logger.info(`Orchestration completed! Result: ${state?.serializedOutput}`);
} catch (error) {
console.error("Error scheduling or waiting for orchestration:", error);
logger.error("Error scheduling or waiting for orchestration:", error);
}

// stop worker and client
Expand Down
Loading
Loading