feat: receive message

Author: lucacasonato

mod.ts

@@ -1,4 +1,10 @@
 export { SQSQueue } from "./src/queue.ts";
 export type { SQSQueueConfig } from "./src/queue.ts";
-export type { SendMessageOptions } from "./src/types.ts";
+export type {
+  SendMessageOptions,
+  SendMessageResponse,
+  ReceiveMessageOptions,
+  ReceiveMessageResponse,
+  Message,
+} from "./src/types.ts";
 export { SQSError } from "./src/error.ts";

src/messages.ts

@@ -1,5 +1,9 @@
 import { parseXML } from "../deps.ts";
-import type { SendMessageResponse } from "./types.ts";
+import type {
+  SendMessageResponse,
+  ReceiveMessageResponse,
+  Message,
+} from "./types.ts";
 import { SQSError } from "./error.ts";
 
 interface Document {
@@ -17,40 +21,69 @@ interface Xml {
 }
 
 export function parseSendMessageResponse(xml: string): SendMessageResponse {
-  const data: Document = parseXML(xml);
-  const { root } = data;
-  if (!root || root.name !== "SendMessageResponse") {
-    throw new SQSError(
-      "Malformed sendMessage response. Missing SendMessageResponse field.",
-      xml,
-    );
-  }
-  const sendMessageResult = root.children.find((d) =>
-    d.name === "SendMessageResult"
-  );
-  if (!sendMessageResult) {
+  const doc: Document = parseXML(xml);
+  const root = extractRoot(doc, "SendMessageResponse");
+  const sendMessageResult = extractField(root, "SendMessageResult");
+
+  const messageID = extractContent(sendMessageResult, "MessageId");
+  const md5OfBody = extractContent(sendMessageResult, "MD5OfMessageBody");
+
+  return { messageID, md5OfBody };
+}
+
+export function parseReceiveMessageBody(xml: string): ReceiveMessageResponse {
+  const doc: Document = parseXML(xml);
+  const root = extractRoot(doc, "ReceiveMessageResponse");
+  const receiveMessageResult = extractField(root, "ReceiveMessageResult");
+
+  const messages = receiveMessageResult.children.map<Message>((message) => {
+    if (message.name !== "Message") {
+      throw new SQSError(
+        "Malformed field. Field type is not Message.",
+        JSON.stringify(message, undefined, 2),
+      );
+    }
+
+    const messageID = extractContent(message, "MessageId");
+    const receiptHandle = extractContent(message, "ReceiptHandle");
+    const md5OfBody = extractContent(message, "MD5OfBody");
+    const body = extractContent(message, "Body");
+
+    return { messageID, md5OfBody, receiptHandle, body };
+  });
+
+  return { messages };
+}
+
+function extractRoot(doc: Document, name: string): Xml {
+  if (!doc.root || doc.root.name !== name) {
     throw new SQSError(
-      "Malformed sendMessage response. Missing SendMessageResult field.",
-      xml,
+      `Malformed XML document. Missing ${name} field.`,
+      JSON.stringify(doc, undefined, 2),
     );
   }
-  const messageIDField = sendMessageResult.children.find((d) =>
-    d.name === "MessageId"
-  );
-  if (!messageIDField) {
+  return doc.root;
+}
+
+function extractField(node: Xml, name: string): Xml {
+  const bodyField = node.children.find((node) => node.name === name);
+  if (!bodyField) {
     throw new SQSError(
-      "Malformed sendMessage response. Missing MessageId field.",
-      xml,
+      `Missing ${name} field in ${node.name} node.`,
+      JSON.stringify(node, undefined, 2),
     );
   }
-  const messageID = messageIDField.content;
-  if (!messageID) {
+  return bodyField;
+}
+
+function extractContent(node: Xml, name: string): string {
+  const field = extractField(node, name);
+  const content = field.content;
+  if (!content) {
     throw new SQSError(
-      "Malformed sendMessage response. Missing content in MessageId field.",
-      xml,
+      `Missing content in ${node.name} node.`,
+      JSON.stringify(node, undefined, 2),
     );
   }
-  return {
-    messageID,
-  };
+  return content;
 }

src/queue.ts

@@ -1,7 +1,15 @@
 import { AWSSignerV4, sha256, parseXML } from "../deps.ts";
-import { parseSendMessageResponse } from "./messages.ts";
+import {
+  parseSendMessageResponse,
+  parseReceiveMessageBody,
+} from "./messages.ts";
 import { SQSError } from "./error.ts";
-import type { SendMessageOptions, SendMessageResponse } from "./types.ts";
+import type {
+  SendMessageOptions,
+  SendMessageResponse,
+  ReceiveMessageOptions,
+  ReceiveMessageResponse,
+} from "./types.ts";
 
 export interface SQSQueueConfig {
   queueURL: string;
@@ -11,27 +19,41 @@ export interface SQSQueueConfig {
   sessionToken?: string;
 }
 
+interface Params {
+  [key: string]: string;
+}
+
 export class SQSQueue {
   #signer: AWSSignerV4;
-  #host: string;
+  #queueURL: string;
 
   constructor(config: SQSQueueConfig) {
     this.#signer = new AWSSignerV4(config.region, {
       awsAccessKeyId: config.accessKeyID,
       awsSecretKey: config.secretKey,
       sessionToken: config.sessionToken,
     });
-    this.#host = config.queueURL;
+    this.#queueURL = config.queueURL;
   }
 
   private _doRequest(
     path: string,
+    params: Params,
     method: string,
     headers: { [key: string]: string },
     body?: Uint8Array | undefined,
   ): Promise<Response> {
-    const url = `${this.#host}${path}`;
-    const signedHeaders = this.#signer.sign("sqs", url, method, headers, body);
+    const url = new URL(this.#queueURL + path);
+    for (const key in params) {
+      url.searchParams.set(key, params[key]);
+    }
+    const signedHeaders = this.#signer.sign(
+      "sqs",
+      url.toString(),
+      method,
+      headers,
+      body,
+    );
     signedHeaders["x-amz-content-sha256"] = sha256(
       body ?? "",
       "utf8",
@@ -47,10 +69,12 @@ export class SQSQueue {
   async sendMessage(
     options: SendMessageOptions,
   ): Promise<SendMessageResponse> {
-    const url = `/?Action=SendMessage&MessageBody=${
-      encodeURIComponent(options.body)
-    }`;
-    const res = await this._doRequest(url, "GET", {});
+    const res = await this._doRequest(
+      "/",
+      { Action: "SendMessage", MessageBody: encodeURIComponent(options.body) },
+      "GET",
+      {},
+    );
     if (!res.ok) {
       throw new SQSError(
         `Failed to send message: ${res.status} ${res.statusText}`,
@@ -60,4 +84,35 @@ export class SQSQueue {
     const xml = await res.text();
     return parseSendMessageResponse(xml);
   }
+
+  async receiveMessage(
+    options?: ReceiveMessageOptions,
+  ): Promise<ReceiveMessageResponse> {
+    const params: Params = { Action: "ReceiveMessage" };
+    if (options) {
+      if (options.maxNumberOfMessages) {
+        params["MaxNumberOfMessages"] = options.maxNumberOfMessages.toString();
+      }
+      if (options.visibilityTimeout) {
+        params["VisibilityTimeout"] = options.visibilityTimeout.toString();
+      }
+      if (options.waitTimeSeconds) {
+        params["WaitTimeSeconds"] = options.waitTimeSeconds.toString();
+      }
+    }
+    const res = await this._doRequest(
+      "/",
+      params,
+      "GET",
+      {},
+    );
+    if (!res.ok) {
+      throw new SQSError(
+        `Failed to send message: ${res.status} ${res.statusText}`,
+        await res.text(),
+      );
+    }
+    const xml = await res.text();
+    return parseReceiveMessageBody(xml);
+  }
 }

src/queue_test.ts

@@ -18,3 +18,17 @@ Deno.test({
     assertEquals(typeof res.messageID, "string");
   },
 });
+
+Deno.test({
+  name: "receive message",
+  async fn() {
+    const res = await queue.receiveMessage();
+    assert(res);
+    assertEquals(res.messages.length, 1);
+    const message = res.messages[0];
+    assertEquals(typeof message.messageID, "string");
+    assertEquals(typeof message.receiptHandle, "string");
+    assertEquals(typeof message.md5OfBody, "string");
+    assertEquals(message.body, "test");
+  },
+});

src/types.ts

@@ -1,8 +1,75 @@
 export interface SendMessageOptions {
+  /**
+   * The message to send. The minimum size is one character. The maximum size
+   * is 256 KB.
+   */
   body: string;
 }
 
 export interface SendMessageResponse {
-  /** An attribute containing the MessageId of the message sent to the queue. */
+  /**
+   * An attribute containing the MessageId of the message sent to the queue.
+   */
   messageID: string;
+
+  /**
+   * An MD5 digest of the non-URL-encoded message body string.
+   */
+  md5OfBody: string;
+}
+
+export interface ReceiveMessageOptions {
+  /**
+   * The maximum number of messages to return. Amazon SQS never returns more
+   * messages than this value (however, fewer messages might be returned).
+   * Valid values: 1 to 10. Default: 1.
+   */
+  maxNumberOfMessages?: number;
+
+  /**
+   * The duration (in seconds) that the received messages are hidden from
+   * subsequent retrieve requests after being retrieved by a ReceiveMessage
+   * request.
+   */
+  visibilityTimeout?: number;
+
+  /**
+   * The duration (in seconds) for which the call waits for a message to arrive
+   * in the queue before returning. If a message is available, the call returns
+   * sooner than WaitTimeSeconds. If no messages are available and the wait time
+   * expires, the call returns successfully with an empty list of messages.
+   */
+  waitTimeSeconds?: number;
+}
+
+export interface ReceiveMessageResponse {
+  /**
+   * An attribute containing the MessageId of the message sent to the queue.
+   */
+  messages: Message[];
+}
+
+export interface Message {
+  /**
+   * A unique identifier for the message. A MessageIdis considered unique across
+   * all AWS accounts for an extended period of time.
+   */
+  messageID: string;
+
+  /**
+   * An MD5 digest of the non-URL-encoded message body string.
+   */
+  md5OfBody: string;
+
+  /**
+   * An identifier associated with the act of receiving the message. A new receipt
+   * handle is returned every time you receive a message. When deleting a message,
+   * you provide the last received receipt handle to delete the message.
+   */
+  receiptHandle: string;
+
+  /**
+   * The message's contents (not URL-encoded).
+   */
+  body: string;
 }