Merge pull request #3 from iuioiua/web-streams

feat: web streams

Author: iuioiua

README.md

@@ -3,7 +3,8 @@
 [![JSR](https://jsr.io/badges/@iuioiua/redis)](https://jsr.io/@iuioiua/redis)
 [![CI](https://github.com/iuioiua/redis/actions/workflows/ci.yml/badge.svg)](https://github.com/iuioiua/redis/actions/workflows/ci.yml)
 
-Minimal [Redis](https://redis.io/) client for [Deno](https://deno.land/).
+Lightning-fast, lightweight and reliable [Redis](https://redis.io/) client for
+Bun, Cloudflare Workers, Deno and web browsers.
 
 ```ts
 import { RedisClient } from "@iuioiua/redis";

bench.ts

@@ -1,3 +1,4 @@
+/// <reference lib="deno.ns" />
 import * as denoRedis from "jsr:@db/redis";
 import { Redis } from "npm:ioredis";
 import { createClient } from "npm:redis";

deno.json

@@ -6,7 +6,7 @@
     "@std/assert": "jsr:@std/assert@^1.0.11",
     "@std/bytes": "jsr:@std/bytes@^1.0.5",
     "@std/collections": "jsr:@std/collections@^1.0.10",
-    "@std/io": "jsr:@std/io@^0.225.2"
+    "@std/streams": "jsr:@std/streams@^1.0.9"
   },
   "tasks": {
     "redis:start": "redis-server --save \"\" --appendonly no --daemonize yes",
@@ -19,5 +19,8 @@
   },
   "exclude": [
     "/coverage"
-  ]
+  ],
+  "compilerOptions": {
+    "lib": ["dom", "dom.iterable", "dom.asynciterable", "esnext"]
+  }
 }

mod.ts

@@ -1,8 +1,6 @@
 // deno-lint-ignore-file no-explicit-any
 import { chunk } from "@std/collections/chunk";
 import { concat } from "@std/bytes/concat";
-import { writeAll } from "@std/io/write-all";
-import type { Reader, Writer } from "@std/io/types";
 
 /**
  * A Redis client for interacting with a Redis server.
@@ -85,13 +83,12 @@ function createRequest(command: Command): Uint8Array {
   return concat(lines);
 }
 
-async function* readLines(reader: Reader): AsyncIterableIterator<Uint8Array> {
-  const buffer = new Uint8Array(1024);
-  let chunks = new Uint8Array();
-  while (true) {
-    const result = await reader.read(buffer);
-    if (result === null) break;
-    chunks = concat([chunks, buffer.subarray(0, result)]);
+async function* readLines(
+  readable: ReadableStream<Uint8Array<ArrayBufferLike>>,
+) {
+  let chunks: Uint8Array<ArrayBufferLike> = new Uint8Array(new ArrayBuffer(0));
+  for await (const chunk of readable) {
+    chunks = concat([chunks, chunk]) as Uint8Array<ArrayBufferLike>;
     let index;
     while (
       (index = chunks.indexOf(CRLF_BYTES[0])) !== -1 &&
@@ -101,7 +98,6 @@ async function* readLines(reader: Reader): AsyncIterableIterator<Uint8Array> {
       chunks = chunks.subarray(index + 2);
     }
   }
-  yield chunks;
 }
 
 function readNReplies(
@@ -121,9 +117,6 @@ async function readReply(
   raw = false,
 ): Promise<Reply> {
   const { value } = await iterator.next();
-  if (value.length === 0) {
-    return Promise.reject(new TypeError("No reply received"));
-  }
   switch (value[0]) {
     case ARRAY_PREFIX:
     case PUSH_PREFIX: {
@@ -376,13 +369,18 @@ async function readReply(
  * ```
  */
 export class RedisClient {
-  #conn: Reader & Writer;
+  #writer: WritableStreamDefaultWriter<Uint8Array>;
   #lines: AsyncIterableIterator<Uint8Array>;
   #queue: Promise<any> = Promise.resolve();
 
-  constructor(conn: Reader & Writer) {
-    this.#conn = conn;
-    this.#lines = readLines(this.#conn);
+  constructor(
+    conn: {
+      readable: ReadableStream<Uint8Array>;
+      writable: WritableStream<Uint8Array>;
+    },
+  ) {
+    this.#writer = conn.writable.getWriter();
+    this.#lines = readLines(conn.readable);
   }
 
   #enqueue<T>(task: () => Promise<T>): Promise<T> {
@@ -410,8 +408,8 @@ export class RedisClient {
    * ```
    */
   sendCommand(command: Command, raw = false): Promise<Reply> {
-    return this.#enqueue(async () => {
-      await writeAll(this.#conn, createRequest(command));
+    return this.#enqueue(() => {
+      this.#writer.write(createRequest(command));
       return readReply(this.#lines, raw);
     });
   }
@@ -436,7 +434,7 @@ export class RedisClient {
    * ```
    */
   writeCommand(command: Command): Promise<void> {
-    return this.#enqueue(() => writeAll(this.#conn, createRequest(command)));
+    return this.#enqueue(() => this.#writer.write(createRequest(command)));
   }
 
   /**
@@ -494,9 +492,9 @@ export class RedisClient {
    * ```
    */
   pipelineCommands(commands: Command[], raw = false): Promise<Reply[]> {
-    return this.#enqueue(async () => {
+    return this.#enqueue(() => {
       const bytes = concat(commands.map(createRequest));
-      await writeAll(this.#conn, bytes);
+      this.#writer.write(bytes);
       return readNReplies(this.#lines, commands.length, raw);
     });
   }

test.ts

@@ -1,5 +1,6 @@
+/// <reference lib="deno.ns" />
 import { assertEquals, assertRejects } from "@std/assert";
-import { Buffer } from "@std/io/buffer";
+import { Buffer } from "@std/streams/buffer";
 import { type Command, RedisClient, type Reply } from "./mod.ts";
 
 const redisConn = await Deno.connect({ port: 6379 });