AWS service endpoints

packages/aws-config/README.md

@@ -45,13 +45,23 @@ then separate sections will be specified within the application code.
   aws_secret_access_key: ${AWS_SECRET_ACCESS_KEY_SES1}
 ```
 
+Most AWS services support endpoints.  The endpoint may be included in the configuration as desired:
+```yaml
+ sqs_for_shipping:
+  aws_region: ${AWS_REGION}
+  aws_endpoint: ${AWS_ENDPOINT_URL_SQS}
+  aws_access_key_id: 'notneededintest'
+  aws_secret_access_key: 'notneededintest'
+```
+
 ## Interfaces
 
 The subsection of `application` in the configuration file should provide:
 ```typescript
 export interface AWSCfgFileItem
 {
     aws_region?: string,
+    aws_endpoint?: string,
     aws_access_key_id?: string,
     aws_secret_access_key?: string,
 }
@@ -63,6 +73,7 @@ export interface AWSServiceConfig
 {
     name: string,
     region: string,
+    endpoint?: string,
     credentials: {
       accessKeyId: string,
       secretAccessKey: string,

packages/aws-config/index.ts

@@ -36,6 +36,7 @@ export interface AWSServiceConfig
 {
     name: string,
     region: string,
+    endpoint?: string,
     credentials: {
       accessKeyId: string,
       secretAccessKey: string,
@@ -50,6 +51,7 @@ export interface AWSServiceConfig
 export interface AWSCfgFileItem
 {
     aws_region?: string,
+    aws_endpoint?: string,
     aws_access_key_id?: string,
     aws_secret_access_key?: string,
 }
@@ -82,8 +84,14 @@ export function loadAWSConfigByName(ctx: ConfigProvider, cfgname: string): AWSSe
         throw new DBOSError.DBOSConfigKeyTypeError(`${cfgname}.aws_secret_access_key`, 'string', typeof(cfgstrs.aws_secret_access_key));
     }
 
+    if (cfgstrs.aws_endpoint && typeof(cfgstrs.aws_region) !== 'string') {
+        throw new DBOSError.DBOSConfigKeyTypeError(`${cfgname}.aws_endpoint`, 'string', typeof(cfgstrs.aws_endpoint));
+    }
+
     return {
-        name: cfgname, region: cfgstrs.aws_region.toString(),
+        name: cfgname,
+        region: cfgstrs.aws_region.toString(),
+        endpoint: cfgstrs.aws_endpoint ? cfgstrs.aws_endpoint : undefined,
         credentials: {
             accessKeyId: cfgstrs.aws_access_key_id.toString(),
             secretAccessKey: cfgstrs.aws_secret_access_key.toString()

packages/communicator-email-ses/index.ts

@@ -99,6 +99,7 @@ class SendEmailCommunicator extends ConfiguredInstance
 
     static createSES(cfg: AWSServiceConfig) {
         return new SESv2({
+            endpoint: cfg.endpoint,
             region: cfg.region,
             credentials: cfg.credentials,
             maxAttempts: cfg.maxRetries,

packages/component-aws-s3/src/s3_utils.ts

@@ -81,6 +81,7 @@ export class S3Ops extends ConfiguredInstance {
 
     static createS3Client(cfg: AWSServiceConfig) {
         return new S3Client({
+            endpoint: cfg.endpoint,
             region: cfg.region,
             credentials: cfg.credentials,
             maxAttempts: cfg.maxRetries,

packages/dbos-sqs/README.md

@@ -80,7 +80,7 @@ The `@SQSConfigure` decorator should be applied at the class level to identify t
 interface SQSConfig {
     awscfgname?: string;
     awscfg?: AWSServiceConfig;
-    queueURL?: string;
+    queueUrl?: string;
     getWFKey?: (m: Message) => string; // Calculate workflow OAOO key for each message
 }
 
@@ -94,7 +94,7 @@ Then, within the class, one or more methods should be decorated to handle SQS me
 ```typescript
 @SQSConfigure({awscfgname: 'sqs_receiver'})
 class SQSEventProcessor {
-  @SQSMessageConsumer({queueURL: process.env['SQS_QUEUE_URL']})
+  @SQSMessageConsumer({queueUrl: process.env['SQS_QUEUE_URL']})
   @Workflow()
   static async recvMessage(ctx: WorkflowContext, msg: Message) {
     // Workflow code goes here...

packages/dbos-sqs/index.ts

@@ -26,7 +26,8 @@ type MessageWithOptionalQueueUrl = MessageWithoutQueueUrl & { QueueUrl?: string
 interface SQSConfig {
     awscfgname?: string;
     awscfg?: AWSServiceConfig;
-    queueURL?: string;
+    queueUrl?: string;
+    queueURL?: string; // disfavored vs. queueUrl
     getWFKey?: (m: Message) => string;
 }
 
@@ -74,7 +75,7 @@ class SQSCommunicator extends ConfiguredInstance
     )
     {
         try {
-            const smsg = {...msg, QueueUrl: msg.QueueUrl || this.config.queueURL};
+            const smsg = {...msg, QueueUrl: msg.QueueUrl || this.config.queueUrl || this.config.queueURL};
             return await this.client!.send(new SendMessageCommand(smsg));
         }
         catch (e) {
@@ -88,6 +89,7 @@ class SQSCommunicator extends ConfiguredInstance
             region: cfg.region,
             credentials: cfg.credentials,
             maxAttempts: cfg.maxRetries,
+            endpoint: cfg.endpoint,
             //logger: console,
         });
     }
@@ -133,7 +135,12 @@ class SQSReceiver implements DBOSEventReceiver
         for (const registeredOperation of regops) {
             const cro = registeredOperation.classConfig as SQSReceiverClassDefaults;
             const mro = registeredOperation.methodConfig as SQSReceiverMethodSpecifics;
-            const url = cro.config?.queueURL ?? mro.config?.queueURL;
+            const url =
+                cro.config?.queueUrl
+                ?? cro.config?.queueURL
+                ?? mro.config?.queueUrl
+                ?? mro.config?.queueURL;
+
             if (url) {
                 const method = registeredOperation.methodReg;
                 const cname = method.className;
@@ -166,7 +173,7 @@ class SQSReceiver implements DBOSEventReceiver
                 catch (e) { 
                     const err = e as Error;
                     executor.logger.error(err);
-                    throw new DBOSError.DBOSError(`SQS Receiver for ${cname}.${mname} was unable to connect: ${err.message}`);
+                    throw new DBOSError.DBOSError(`SQS Receiver for ${cname}.${mname} was unable to connect to ${url}: ${err.message}`);
                 }
                 this.listeners.push((async (client: SQSClient, url: string) =>
                 {
@@ -220,7 +227,11 @@ class SQSReceiver implements DBOSEventReceiver
         regops.forEach((registeredOperation) => {
             const co = registeredOperation.classConfig as SQSReceiverClassDefaults;
             const mo = registeredOperation.methodConfig as SQSReceiverMethodSpecifics;
-            const url = co.config?.queueURL ?? mo.config?.queueURL;
+            const url =
+                co.config?.queueUrl
+                ?? co.config?.queueURL
+                ?? mo.config?.queueUrl
+                ?? mo.config?.queueURL;
             if (url) {
                 const cname = registeredOperation.methodReg.className;
                 const mname = registeredOperation.methodReg.name;

packages/dbos-sqs/sqs-test-dbos-config.yaml

@@ -19,6 +19,7 @@ application:
   aws_sqs_configuration: 'aws_config'
   sqs_queue: ${SQS_QUEUE_URL}
   aws_config:
+    aws_endpoint: ${AWS_ENDPOINT_URL_SQS}
     aws_region: ${AWS_REGION}
     aws_access_key_id: ${AWS_ACCESS_KEY_ID}
     aws_secret_access_key: ${AWS_SECRET_ACCESS_KEY}

packages/dbos-sqs/sqs.test.ts

@@ -13,7 +13,7 @@ class SQSReceiver
 {
   static msgRcvCount: number = 0;
   static msgValueSum: number = 0;
-  @SQSMessageConsumer({queueURL: process.env['SQS_QUEUE_URL']})
+  @SQSMessageConsumer({queueUrl: process.env['SQS_QUEUE_URL']})
   @Workflow()
   static async recvMessage(_ctx: WorkflowContext, msg: Message) {
     const ms = msg.Body!;
@@ -36,7 +36,7 @@ describe("sqs-tests", () => {
     }
     else {
       // This would normally be a global or static or something
-      sqsCfg = configureInstance(SQSCommunicator, 'default', {awscfgname: 'aws_config', queueURL: process.env['SQS_QUEUE_URL']});
+      sqsCfg = configureInstance(SQSCommunicator, 'default', {awscfgname: 'aws_config', queueUrl: process.env['SQS_QUEUE_URL']});
     }
   });