refactor: move line read logic in-house

mod.ts

@@ -1,7 +1,7 @@
 // deno-lint-ignore-file no-explicit-any
 import { chunk } from "@std/collections/chunk";
 import { concat } from "@std/bytes/concat";
-import { readDelim } from "@std/io/read_delim";
+import { readLines } from "./read_delim.ts";
 import { writeAll } from "@std/io/write_all";
 import type { Writer } from "@std/io/types";
 
@@ -248,7 +248,7 @@ async function sendCommand(
   raw = false,
 ): Promise<Reply> {
   await writeCommand(redisConn, command);
-  return await readReply(readDelim(redisConn, CRLF_RAW), raw);
+  return await readReply(readLines(redisConn), raw);
 }
 
 async function pipelineCommands(
@@ -257,14 +257,14 @@ async function pipelineCommands(
 ): Promise<Reply[]> {
   const bytes = commands.map(createRequest);
   await writeAll(redisConn, concat(bytes));
-  return readNReplies(commands.length, readDelim(redisConn, CRLF_RAW));
+  return readNReplies(commands.length, readLines(redisConn));
 }
 
 async function* readReplies(
   redisConn: Deno.Conn,
   raw = false,
 ): AsyncIterableIterator<Reply> {
-  const iterator = readDelim(redisConn, CRLF_RAW);
+  const iterator = readLines(redisConn);
   while (true) {
     yield await readReply(iterator, raw);
   }

read_delim.ts

@@ -0,0 +1,35 @@
+// Copyright 2018-2024 the Deno authors. All rights reserved. MIT license.
+// This module is browser compatible.
+
+import type { Reader } from "@std/io/types";
+
+const CRLF = new TextEncoder().encode("\r\n");
+const BUFFER_LENGTH = 1_024;
+
+function concat(a: Uint8Array, b: Uint8Array): Uint8Array {
+  const output = new Uint8Array(a.length + b.length);
+  output.set(a, 0);
+  output.set(b, a.length);
+  return output;
+}
+
+export async function* readLines(
+  reader: Reader,
+): AsyncIterableIterator<Uint8Array> {
+  let chunks = new Uint8Array();
+  while (true) {
+    const buffer = new Uint8Array(BUFFER_LENGTH);
+    const result = await reader.read(buffer);
+    if (result === null) {
+      yield chunks;
+      return;
+    }
+    chunks = concat(chunks, buffer.slice(0, result));
+    const crlfIndex = chunks.indexOf(CRLF[0]);
+    if (crlfIndex !== -1 && chunks[crlfIndex + 1] === CRLF[1]) {
+      const line = chunks.slice(0, crlfIndex);
+      yield line;
+      chunks = chunks.slice(crlfIndex + 2);
+    }
+  }
+}

test.ts

@@ -161,8 +161,11 @@ async function sendCommandTest(
 
 Deno.test("redisClient.sendCommand() - transactions", async () => {
   await sendCommandTest(["MULTI"], "OK");
+  console.log(1);
   await sendCommandTest(["INCR", "FOO"], "QUEUED");
+  console.log(1);
   await sendCommandTest(["INCR", "BAR"], "QUEUED");
+  console.log(1);
   await sendCommandTest(["EXEC"], [1, 1]);
 });