feat(BrokerUtil): support sending bigints

Author: didinele

libs/brokers/src/brokers/BrokerUtil.ts

@@ -1,6 +1,6 @@
 import { randomBytes } from 'crypto';
 import { getMissingProps } from '@cordis/common';
-import { encode, decode } from '@msgpack/msgpack';
+import { encode, decode, ExtensionCodec } from '@msgpack/msgpack';
 import { CordisBrokerTypeError } from '../error';
 import type * as amqp from 'amqplib';
 import type { Broker } from './Broker';
@@ -88,7 +88,21 @@ export class BrokerUtil {
      * Broker instance
      */
     public readonly broker: Broker
-  ) {}
+  ) {
+    this.packBigintExtensionCodec.register({
+      type: 0,
+      encode: (input: unknown) => {
+        if (typeof input === 'bigint') {
+          return encode(input.toString());
+        }
+
+        return null;
+      },
+      decode: (data: Uint8Array) => BigInt(decode(data) as string)
+    });
+  }
+
+  public readonly packBigintExtensionCodec = new ExtensionCodec();
 
   /**
    * Generates a base64 string with the given length using Node.js' Crypto
@@ -118,7 +132,7 @@ export class BrokerUtil {
         }
 
         try {
-          await cb(decode(msg.content) as T, msg);
+          await cb(decode(msg.content, { extensionCodec: this.packBigintExtensionCodec }) as T, msg);
         } catch (e) {
           this.broker.emit('error', e);
           this.broker.channel.reject(msg, true);
@@ -145,7 +159,7 @@ export class BrokerUtil {
   public sendToQueue<T>(options: SendToQueueOptions<T>) {
     const { to, content, options: amqpOptions } = options;
 
-    const encoded = encode(content);
+    const encoded = encode(content, { extensionCodec: this.packBigintExtensionCodec });
     const data = Buffer.from(encoded.buffer, encoded.byteOffset, encoded.byteLength);
     return this.broker.channel.sendToQueue(to, data, amqpOptions);
   }
@@ -158,7 +172,7 @@ export class BrokerUtil {
   public sendToExchange<T>(options: SendToExchangeOptions<T>) {
     const { to, content, key, options: amqpOptions } = options;
 
-    const encoded = encode(content);
+    const encoded = encode(content, { extensionCodec: this.packBigintExtensionCodec });
     const data = Buffer.from(encoded.buffer, encoded.byteOffset, encoded.byteLength);
     return this.broker.channel.publish(to, key, data, amqpOptions);
   }