Cleanup

Author: kibertoad

README.md

@@ -286,6 +286,7 @@ const consumer = new MySnsSqsConsumer(dependencies, {
 | `pollingIntervalMs` | `number` | `5000` | Interval between availability checks (ms) |
 | `timeoutMs` | `number \| NO_TIMEOUT` | - (required) | Maximum wait time before throwing `StartupResourcePollingTimeoutError`. Use `NO_TIMEOUT` to poll indefinitely |
 | `throwOnTimeout` | `boolean` | `true` | When `true`, throws error on timeout. When `false`, reports error via errorReporter, resets timeout counter, and continues polling |
+| `nonBlocking` | `boolean` | `false` | When `true`, `init()` returns immediately if resource is not available, and polling continues in the background. A callback is invoked when the resource becomes available |
 
 ### Environment-Specific Configuration
 
@@ -338,6 +339,19 @@ import { NO_TIMEOUT } from '@message-queue-toolkit/core'
     }
   }
 }
+
+// Non-blocking mode - service starts immediately, polling continues in background
+// Useful when you want the service to be available even if dependencies are not ready
+{
+  locatorConfig: {
+    queueUrl: '...',
+    startupResourcePolling: {
+      enabled: true,
+      timeoutMs: 5 * 60 * 1000,
+      nonBlocking: true,  // init() resolves immediately
+    }
+  }
+}
 ```
 
 ### Error Handling

packages/sns/lib/utils/snsInitter.ts

@@ -47,7 +47,13 @@ export async function initSnsSqs(
   creationConfig?: SNSCreationConfig & SQSCreationConfig,
   subscriptionConfig?: SNSSubscriptionOptions,
   extraParams?: InitSnsSqsExtraParams,
-): Promise<{ subscriptionArn: string; topicArn: string; queueUrl: string; queueName: string }> {
+): Promise<{
+  subscriptionArn: string
+  topicArn: string
+  queueUrl: string
+  queueName: string
+  resourcesReady: boolean
+}> {
   if (!locatorConfig?.subscriptionArn) {
     if (!creationConfig?.topic && !locatorConfig?.topicArn && !locatorConfig?.topicName) {
       throw new Error(
@@ -303,61 +309,70 @@ export async function deleteSns(
   await deleteTopic(snsClient, stsClient, creationConfig.topic.Name)
 }
 
-export async function initSns(
+async function initSnsWithLocator(
   snsClient: SNSClient,
   stsClient: STSClient,
-  locatorConfig?: SNSTopicLocatorType,
-  creationConfig?: SNSCreationConfig,
+  locatorConfig: SNSTopicLocatorType,
   extraParams?: InitSnsExtraParams,
 ) {
-  if (locatorConfig) {
-    if (!locatorConfig.topicArn && !locatorConfig.topicName) {
-      throw new Error(
-        'When locatorConfig for the topic is specified, either topicArn or topicName must be specified',
-      )
-    }
+  if (!locatorConfig.topicArn && !locatorConfig.topicName) {
+    throw new Error(
+      'When locatorConfig for the topic is specified, either topicArn or topicName must be specified',
+    )
+  }
 
-    const topicArn =
-      locatorConfig.topicArn ?? (await buildTopicArn(stsClient, locatorConfig.topicName ?? ''))
+  const topicArn =
+    locatorConfig.topicArn ?? (await buildTopicArn(stsClient, locatorConfig.topicName ?? ''))
 
-    const startupResourcePolling = locatorConfig.startupResourcePolling
+  const startupResourcePolling = locatorConfig.startupResourcePolling
 
-    // If startup resource polling is enabled, poll for topic to become available
-    if (isStartupResourcePollingEnabled(startupResourcePolling)) {
-      const nonBlocking = startupResourcePolling.nonBlocking === true
+  // If startup resource polling is enabled, poll for topic to become available
+  if (isStartupResourcePollingEnabled(startupResourcePolling)) {
+    const nonBlocking = startupResourcePolling.nonBlocking === true
 
-      const topicResult = await waitForResource({
-        config: startupResourcePolling,
-        resourceName: `SNS topic ${topicArn}`,
-        logger: extraParams?.logger,
-        errorReporter: extraParams?.errorReporter,
-        onResourceAvailable: () => {
-          if (nonBlocking) {
-            extraParams?.onTopicReady?.({ topicArn })
-          }
-        },
-        checkFn: async () => {
-          const result = await getTopicAttributes(snsClient, topicArn)
-          if (result.error === 'not_found') {
-            return { isAvailable: false }
-          }
-          return { isAvailable: true, result: result.result }
-        },
-      })
-
-      // In non-blocking mode, return early if topic wasn't immediately available
-      if (nonBlocking && topicResult === undefined) {
-        return { topicArn, resourcesReady: false }
-      }
-    } else {
-      // Original behavior: check once and fail immediately if not found
-      const checkResult = await getTopicAttributes(snsClient, topicArn)
-      if (checkResult.error === 'not_found') {
-        throw new Error(`Topic with topicArn ${topicArn} does not exist.`)
-      }
+    const topicResult = await waitForResource({
+      config: startupResourcePolling,
+      resourceName: `SNS topic ${topicArn}`,
+      logger: extraParams?.logger,
+      errorReporter: extraParams?.errorReporter,
+      onResourceAvailable: () => {
+        if (nonBlocking) {
+          extraParams?.onTopicReady?.({ topicArn })
+        }
+      },
+      checkFn: async () => {
+        const result = await getTopicAttributes(snsClient, topicArn)
+        if (result.error === 'not_found') {
+          return { isAvailable: false }
+        }
+        return { isAvailable: true, result: result.result }
+      },
+    })
+
+    // In non-blocking mode, return early if topic wasn't immediately available
+    if (nonBlocking && topicResult === undefined) {
+      return { topicArn, resourcesReady: false }
+    }
+  } else {
+    // Original behavior: check once and fail immediately if not found
+    const checkResult = await getTopicAttributes(snsClient, topicArn)
+    if (checkResult.error === 'not_found') {
+      throw new Error(`Topic with topicArn ${topicArn} does not exist.`)
     }
+  }
+
+  return { topicArn, resourcesReady: true }
+}
 
-    return { topicArn, resourcesReady: true }
+export async function initSns(
+  snsClient: SNSClient,
+  stsClient: STSClient,
+  locatorConfig?: SNSTopicLocatorType,
+  creationConfig?: SNSCreationConfig,
+  extraParams?: InitSnsExtraParams,
+) {
+  if (locatorConfig) {
+    return initSnsWithLocator(snsClient, stsClient, locatorConfig, extraParams)
   }
 
   // create new topic if it does not exist

packages/sqs/lib/sqs/AbstractSqsService.ts

@@ -95,8 +95,7 @@ export abstract class AbstractSqsService<
   protected queueName: string
   // @ts-expect-error
   protected queueUrl: string
-  // @ts-expect-error
-  protected queueArn: string
+  protected queueArn: string | undefined
   protected readonly isFifoQueue: boolean
 
   constructor(dependencies: DependenciesType, options: SQSOptionsType) {