feat: update to deno v2

deno.json

@@ -12,5 +12,9 @@
     "codegen:codec": "deno run codegen/generate_codec.ts | deno fmt - > src/amqp_codec.ts",
     "codegen:types": "deno run codegen/generate_types.ts | deno fmt - > src/amqp_types.ts",
     "codegen:constants": "deno run codegen/generate_constants.ts | deno fmt - > src/amqp_constants.ts"
+  },
+  "imports": {
+    "@std/io": "jsr:@std/io@^0.225.0",
+    "@std/streams": "jsr:@std/streams@^1.0.7"
   }
 }

deps.ts

@@ -1 +1 @@
-export { Buffer, BufReader, writeAll } from "jsr:@std/io";
+export { writeAll, Buffer } from "jsr:@std/io";

src/amqp_frame_reader.ts

@@ -1,24 +1,25 @@
 import { FrameError } from "./frame_error.ts";
 import { decodeHeader, decodeMethod } from "./amqp_codec.ts";
 import type { IncomingFrame } from "./amqp_frame.ts";
-import { BufReader } from "../deps.ts";
+import type { Reader } from "jsr:@std/io/types";
 
 export class AmqpFrameReader {
   #timer: null | number = null;
-  #reader: BufReader;
+  #reader: Reader;
 
-  constructor(r: Deno.Reader) {
-    this.#reader = BufReader.create(r);
+  constructor(r: Reader) {
+    this.#reader = r;
   }
 
   #readBytes = async (length: number): Promise<Uint8Array> => {
-    const n = await this.#reader.readFull(new Uint8Array(length));
+    const buf = new Uint8Array(length);
+    const n = await this.#reader.read(buf);
 
     if (n === null) {
       throw new FrameError("EOF");
     }
 
-    return n;
+    return buf;
   };
 
   #readFrame = async (): Promise<IncomingFrame> => {
@@ -80,9 +81,7 @@ export class AmqpFrameReader {
     }
   }
 
-  read(
-    timeout: number,
-  ): Promise<IncomingFrame> {
+  read(timeout: number): Promise<IncomingFrame> {
     this.abort();
 
     if (timeout <= 0) {

src/amqp_multiplexer.ts

@@ -109,7 +109,7 @@ function createSocketDemux(
         emit(frame);
       }
     } catch (error) {
-      handleError(error);
+      handleError(error as Error);
     }
   }
 
@@ -134,7 +134,7 @@ function createSocketDemux(
           removeSubscriber(subscriber);
         }
       } catch (error) {
-        subscriber.error(error);
+        subscriber.error(error as Error);
         removeSubscriber(subscriber);
       }
     }

src/amqp_socket.ts

@@ -2,7 +2,7 @@ import { encodeHeader, encodeMethod } from "./amqp_codec.ts";
 import { AmqpFrameReader } from "./amqp_frame_reader.ts";
 import type { IncomingFrame, OutgoingFrame } from "./amqp_frame.ts";
 import { writeAll } from "../deps.ts";
-
+import type { Reader,Writer, Closer } from "jsr:@std/io/types";
 export interface AmqpSocketWriter {
   write(frames: Array<OutgoingFrame>): Promise<void>;
 }
@@ -49,16 +49,7 @@ function encodeFrame(frame: OutgoingFrame): Uint8Array {
   return data;
 }
 
-const HEARTBEAT_FRAME = new Uint8Array([
-  8,
-  0,
-  0,
-  0,
-  0,
-  0,
-  0,
-  206,
-]);
+const HEARTBEAT_FRAME = new Uint8Array([8, 0, 0, 0, 0, 0, 0, 206]);
 
 function splitArray(arr: Uint8Array, size: number): Uint8Array[] {
   const chunks: Uint8Array[] = [];
@@ -78,16 +69,18 @@ interface AmqpSocketOptions {
   frameMax?: number;
 }
 
-export class AmqpSocket implements AmqpSocketWriter, AmqpSocketReader, AmqpSocketCloser {
-  #conn: Deno.Reader & Deno.Writer & Deno.Closer;
+export class AmqpSocket
+  implements AmqpSocketWriter, AmqpSocketReader, AmqpSocketCloser
+{
+  #conn: Reader & Writer & Closer;
   #reader: AmqpFrameReader;
   #sendTimer: number | null = null;
   #sendTimeout = 0;
   #readTimeout = 0;
   #frameMax = -1;
   #guard: Promise<void>;
 
-  constructor(conn: Deno.Reader & Deno.Writer & Deno.Closer) {
+  constructor(conn: Reader & Writer & Closer) {
     this.#conn = conn;
     this.#reader = new AmqpFrameReader(conn);
     this.#guard = Promise.resolve();
@@ -108,9 +101,16 @@ export class AmqpSocket implements AmqpSocketWriter, AmqpSocketReader, AmqpSocke
   };
 
   tune(options: AmqpSocketOptions) {
-    this.#readTimeout = options.readTimeout !== undefined ? options.readTimeout : this.#readTimeout;
-    this.#sendTimeout = options.sendTimeout !== undefined ? options.sendTimeout : this.#sendTimeout;
-    this.#frameMax = options.frameMax !== undefined ? options.frameMax : this.#frameMax;
+    this.#readTimeout =
+      options.readTimeout !== undefined
+        ? options.readTimeout
+        : this.#readTimeout;
+    this.#sendTimeout =
+      options.sendTimeout !== undefined
+        ? options.sendTimeout
+        : this.#sendTimeout;
+    this.#frameMax =
+      options.frameMax !== undefined ? options.frameMax : this.#frameMax;
     this.#resetSendTimer();
   }
 
@@ -124,23 +124,23 @@ export class AmqpSocket implements AmqpSocketWriter, AmqpSocketReader, AmqpSocke
     this.#resetSendTimer();
     for (const frame of frames) {
       if (frame.type === "content") {
-        const chunks = this.#frameMax > 8 && frame.payload.length > this.#frameMax - 8
-          ? splitArray(
-            frame.payload,
-            this.#frameMax - 8,
-          ).map((chunk) =>
-            encodeFrame({
-              type: "content",
-              channel: frame.channel,
-              payload: chunk,
-            })
-          )
-          : [encodeFrame(frame)];
+        const chunks =
+          this.#frameMax > 8 && frame.payload.length > this.#frameMax - 8
+            ? splitArray(frame.payload, this.#frameMax - 8).map((chunk) =>
+                encodeFrame({
+                  type: "content",
+                  channel: frame.channel,
+                  payload: chunk,
+                }),
+              )
+            : [encodeFrame(frame)];
         for (const chunk of chunks) {
           this.#guard = this.#guard.then(() => writeAll(this.#conn, chunk));
         }
       } else {
-        this.#guard = this.#guard.then(() => writeAll(this.#conn, encodeFrame(frame)));
+        this.#guard = this.#guard.then(() =>
+          writeAll(this.#conn, encodeFrame(frame)),
+        );
       }
     }
 

src/amqp_socket_test.ts

@@ -3,13 +3,14 @@ import { arrayOf, assertEquals, assertRejects, test } from "./testing.ts";
 import { mock } from "./mock.ts";
 import { FrameError } from "./frame_error.ts";
 import { createResolvable } from "./resolvable.ts";
+import type { Reader,Writer, Closer } from "jsr:@std/io/types";
 
 function sleep(ms: number) {
   return new Promise((resolve) => setTimeout(resolve, ms));
 }
 
 function createConn() {
-  return mock.obj<Deno.Reader & Deno.Writer & Deno.Closer>({
+  return mock.obj<Reader & Writer & Closer>({
     read: mock.fn(),
     write: mock.fn(async () => {}),
     close: mock.fn(() => {}),

src/encoding/utils.ts

@@ -1,3 +1,5 @@
+import type { Reader, ReaderSync } from "@std/io/types";
+
 export function splitArray<T>(arr: T[], size: number): T[][] {
   const chunks: T[][] = [];
   let index = 0;
@@ -22,7 +24,7 @@ export function assertLength(arr: Uint8Array, length: number) {
 }
 
 export async function readBytes(
-  r: Deno.Reader,
+  r: Reader,
   length: number,
 ): Promise<Uint8Array | null> {
   const data = new Uint8Array(length);
@@ -39,7 +41,7 @@ export async function readBytes(
   return data;
 }
 
-export function readBytesSync(r: Deno.ReaderSync, length: number): Uint8Array {
+export function readBytesSync(r: ReaderSync, length: number): Uint8Array {
   const data = new Uint8Array(length);
   const result = r.readSync(data);