feat: break out (de)serializer (#2)

denosaurs · May 6, 2023 · 5f800b3 · 5f800b3
1 parent 505d3ce
commit 5f800b3
Show file tree

Hide file tree

Showing 5 changed files with 48 additions and 35 deletions.
diff --git a/deps.ts b/deps.ts
@@ -1,2 +1,2 @@
-export * from "https://deno.land/std@0.170.0/async/mod.ts";
+export * from "https://deno.land/std@0.186.0/async/mod.ts";
 export * from "https://deno.land/x/[email protected]/mod.ts";
diff --git a/message_queue/amqp.ts b/message_queue/amqp.ts
@@ -5,7 +5,7 @@ import {
   BasicDeliver,
   BasicProperties,
   connect,
-} from "https://deno.land/x/amqp@v0.21.0/mod.ts";
+} from "https://deno.land/x/amqp@v0.23.1/mod.ts";
 
 import { MessageEvent } from "./message_event.ts";
 import { MessageQueue, MessageQueueOptions } from "./message_queue.ts";
@@ -52,7 +52,7 @@ export class AMQPMessageQueue<T = any> extends MessageQueue<T> {
   async #consume(
     args: BasicDeliver,
     props: BasicProperties,
-    rawData: Uint8Array,
+    data: Uint8Array,
   ): Promise<void> {
     if (
       props.headers &&
@@ -62,7 +62,6 @@ export class AMQPMessageQueue<T = any> extends MessageQueue<T> {
       return;
     }
 
-    // TODO: Break the message decoding out into a helper function
     if (this.encoderDecoder?.contentEncoding !== props.contentEncoding) {
       throw new TypeError(
         `Expected message to have the contentEncoding ${this.encoderDecoder?.contentEncoding} but found ${props.contentEncoding}`,
@@ -75,17 +74,8 @@ export class AMQPMessageQueue<T = any> extends MessageQueue<T> {
       );
     }
 
-    let data: T | Uint8Array = rawData;
-    if (this.encoderDecoder) {
-      data = this.encoderDecoder.decode(data);
-    }
-
-    if (this.serializerDeserializer) {
-      data = this.serializerDeserializer.deserialize(data);
-    }
-
     const event = new MessageEvent<T>({
-      data: data as T,
+      data: this.decode(data),
       origin: args.consumerTag,
     });
 
@@ -103,7 +93,10 @@ export class AMQPMessageQueue<T = any> extends MessageQueue<T> {
           break;
         }
         case "rejected": {
-          await this.#channel!.nack({ deliveryTag: args.deliveryTag, requeue });
+          await this.#channel!.reject({
+            deliveryTag: args.deliveryTag,
+            requeue,
+          });
           break;
         }
         case "pending":
@@ -124,24 +117,7 @@ export class AMQPMessageQueue<T = any> extends MessageQueue<T> {
   }
 
   async queueMessage(message: T): Promise<void> {
-    // TODO: Break the message encoding out into a helper function
-    let data: Uint8Array;
-
-    if (this.serializerDeserializer) {
-      data = this.serializerDeserializer.serialize(message);
-    } else {
-      if (message instanceof Uint8Array) {
-        data = new Uint8Array(message);
-      } else {
-        throw new TypeError(
-          "Expected an Uint8Array when calling queueMessage without an serializerDeserializer configured",
-        );
-      }
-    }
-
-    if (this.encoderDecoder) {
-      data = this.encoderDecoder.encode(data);
-    }
+    const data = this.encode(message);
 
     if (this.#closed) {
       throw new Error("This AMQPMessageQueue has already been closed");

diff --git a/message_queue/message_queue.ts b/message_queue/message_queue.ts
@@ -55,6 +55,42 @@ export abstract class MessageQueue<T = any>
     this.#id = options.id ?? crypto.randomUUID();
   }
 
+  protected encode(message: T): Uint8Array {
+    let data: Uint8Array;
+
+    if (this.serializerDeserializer) {
+      data = this.serializerDeserializer.serialize(message);
+    } else {
+      if (message instanceof Uint8Array) {
+        data = new Uint8Array(message);
+      } else {
+        throw new TypeError(
+          "Expected an Uint8Array when calling queueMessage without an serializerDeserializer configured",
+        );
+      }
+    }
+
+    if (this.encoderDecoder) {
+      data = this.encoderDecoder.encode(data);
+    }
+
+    return data;
+  }
+
+  protected decode(data: Uint8Array): T {
+    let message: T | Uint8Array = data;
+
+    if (this.encoderDecoder) {
+      message = this.encoderDecoder.decode(data);
+    }
+
+    if (this.serializerDeserializer) {
+      message = this.serializerDeserializer.deserialize(data);
+    }
+
+    return message as T;
+  }
+
   abstract close(): Promise<void>;
   abstract queueMessage(message: T): Promise<void>;
 }
diff --git a/serializer_deserializer/types.ts b/serializer_deserializer/types.ts
@@ -1,4 +1,5 @@
-export interface SerializerDeserializer<T> {
+// deno-lint-ignore no-explicit-any
+export interface SerializerDeserializer<T = any> {
   contentType?: string;
   serialize(value: T): Uint8Array;
   deserialize(data: Uint8Array): T;

diff --git a/serializer_deserializer/v8.ts b/serializer_deserializer/v8.ts
@@ -1,7 +1,7 @@
 import {
   deserializeAny,
   serializeAny,
-} from "https://raw.githubusercontent.com/MierenManz/v8_format/main/references/mod.ts";
+} from "https://raw.githubusercontent.com/MierenManz/v8_format/8352edf/src/mod.ts";
 import { SerializerDeserializer } from "./types.ts";
 
 const v8SerializerDeserializer = {