feat: implement web streams based Connection

command.ts

@@ -8,7 +8,7 @@ import type {
   Raw,
   RedisValue,
   SimpleString,
-} from "./protocol/mod.ts";
+} from "./protocol/shared/types.ts";
 import type { RedisPipeline } from "./pipeline.ts";
 import type { RedisSubscription } from "./pubsub.ts";
 import type {

connection.ts

@@ -1,6 +1,10 @@
-import { readReply, sendCommand, sendCommands } from "./protocol/mod.ts";
-import type { RedisReply, RedisValue } from "./protocol/mod.ts";
-import type { Command } from "./protocol/command.ts";
+import {
+  readReply,
+  sendCommand,
+  sendCommands,
+} from "./protocol/deno_streams/mod.ts";
+import type { RedisReply, RedisValue } from "./protocol/shared/types.ts";
+import type { Command } from "./protocol/deno_streams/command.ts";
 import type { Backoff } from "./backoff.ts";
 import { exponentialBackoff } from "./backoff.ts";
 import { ErrorReplyError, isRetriableError } from "./errors.ts";

executor.ts

@@ -1,5 +1,5 @@
 import type { Connection, SendCommandOptions } from "./connection.ts";
-import type { RedisReply, RedisValue } from "./protocol/mod.ts";
+import type { RedisReply, RedisValue } from "./protocol/shared/types.ts";
 
 export interface CommandExecutor {
   readonly connection: Connection;

experimental/cluster/mod.ts

@@ -28,7 +28,7 @@ import type { RedisConnectOptions } from "../../redis.ts";
 import type { CommandExecutor } from "../../executor.ts";
 import type { Connection, SendCommandOptions } from "../../connection.ts";
 import type { Redis } from "../../redis.ts";
-import type { RedisReply, RedisValue } from "../../protocol/mod.ts";
+import type { RedisReply, RedisValue } from "../../protocol/shared/types.ts";
 import { ErrorReplyError } from "../../errors.ts";
 import { delay } from "../../vendor/https/deno.land/std/async/delay.ts";
 import calculateSlot from "../../vendor/https/esm.sh/cluster-key-slot/lib/index.js";

mod.ts

@@ -1,5 +1,5 @@
 // Generated by tools/make_mod.ts. Don't edit.
-export { okReply } from "./protocol/mod.ts";
+export { okReply } from "./protocol/shared/types.ts";
 export { connect, create, createLazyClient, parseURL } from "./redis.ts";
 export {
   ConnectionClosedError,
@@ -72,7 +72,7 @@ export type {
   RedisReply,
   RedisValue,
   SimpleString,
-} from "./protocol/mod.ts";
+} from "./protocol/shared/types.ts";
 export type { RedisPubSubMessage, RedisSubscription } from "./pubsub.ts";
 export type { Redis, RedisConnectOptions } from "./redis.ts";
 export type {

pipeline.ts

@@ -1,7 +1,12 @@
 import type { Connection, SendCommandOptions } from "./connection.ts";
 import { kEmptyRedisArgs } from "./connection.ts";
 import { CommandExecutor } from "./executor.ts";
-import { okReply, RawOrError, RedisReply, RedisValue } from "./protocol/mod.ts";
+import {
+  okReply,
+  RawOrError,
+  RedisReply,
+  RedisValue,
+} from "./protocol/shared/types.ts";
 import { create, Redis } from "./redis.ts";
 import { kUnstablePipeline } from "./internal/symbols.ts";
 import {

protocol/command.ts → protocol/deno_streams/command.ts

@@ -1,9 +1,9 @@
-import { BufReader } from "../vendor/https/deno.land/std/io/buf_reader.ts";
-import { BufWriter } from "../vendor/https/deno.land/std/io/buf_writer.ts";
+import { BufReader } from "../../vendor/https/deno.land/std/io/buf_reader.ts";
+import { BufWriter } from "../../vendor/https/deno.land/std/io/buf_writer.ts";
 import { readReply } from "./reply.ts";
-import { ErrorReplyError } from "../errors.ts";
-import { encoder } from "./_util.ts";
-import type { RedisReply, RedisValue } from "./types.ts";
+import { ErrorReplyError } from "../../errors.ts";
+import { encoder } from "../../internal/encoding.ts";
+import type { RedisReply, RedisValue } from "../shared/types.ts";
 
 const CRLF = encoder.encode("\r\n");
 const ArrayCode = encoder.encode("*");

protocol/deno_streams/mod.ts

@@ -0,0 +1,3 @@
+export { readArrayReply, readReply } from "./reply.ts";
+
+export { sendCommand, sendCommands } from "./command.ts";

protocol/reply.ts → protocol/deno_streams/reply.ts

@@ -1,7 +1,7 @@
-import { BufReader } from "../vendor/https/deno.land/std/io/buf_reader.ts";
-import type * as types from "./types.ts";
-import { EOFError, ErrorReplyError, InvalidStateError } from "../errors.ts";
-import { decoder } from "./_util.ts";
+import { BufReader } from "../../vendor/https/deno.land/std/io/buf_reader.ts";
+import type * as types from "../shared/types.ts";
+import { EOFError, ErrorReplyError, InvalidStateError } from "../../errors.ts";
+import { decoder } from "../../internal/encoding.ts";
 
 const IntegerReplyCode = ":".charCodeAt(0);
 const BulkReplyCode = "$".charCodeAt(0);
@@ -113,8 +113,6 @@ export async function readArrayReply(
   return array;
 }
 
-export const okReply = "OK";
-
 function tryParseErrorReply(line: Uint8Array): never {
   const code = line[0];
   if (code === ErrorReplyCode) {

protocol/types.ts → protocol/shared/types.ts

@@ -1,4 +1,4 @@
-import { ErrorReplyError } from "../errors.ts";
+import { ErrorReplyError } from "../../errors.ts";
 
 /**
  * @see https://redis.io/topics/protocol
@@ -46,3 +46,5 @@ export type ConditionalArray = Raw[];
 export type RedisReply = Raw | ConditionalArray;
 
 export type RawOrError = Raw | ErrorReplyError;
+
+export const okReply = "OK";

pubsub.ts

@@ -1,7 +1,7 @@
 import type { CommandExecutor } from "./executor.ts";
 import { EOFError, InvalidStateError } from "./errors.ts";
-import type { Binary } from "./protocol/mod.ts";
-import { decoder } from "./protocol/_util.ts";
+import type { Binary } from "./protocol/shared/types.ts";
+import { decoder } from "./internal/encoding.ts";
 import { kUnstableReadReply } from "./internal/symbols.ts";
 
 type DefaultMessageType = string;

redis.ts

@@ -56,7 +56,7 @@ import type {
   RedisReply,
   RedisValue,
   SimpleString,
-} from "./protocol/mod.ts";
+} from "./protocol/shared/types.ts";
 import { createRedisPipeline } from "./pipeline.ts";
 import { psubscribe, subscribe } from "./pubsub.ts";
 import {

stream.ts

@@ -1,4 +1,8 @@
-import type { ConditionalArray, Raw, RedisValue } from "./protocol/mod.ts";
+import type {
+  ConditionalArray,
+  Raw,
+  RedisValue,
+} from "./protocol/shared/types.ts";
 
 export interface XId {
   unixMs: number;

tools/make_mod.ts

@@ -45,12 +45,13 @@ function collectSourceFilesInRootDir(): Array<string> {
   }).map((f) => f.name);
 }
 
-const files = [...collectSourceFilesInRootDir(), "protocol/mod.ts"].sort();
+const files = [...collectSourceFilesInRootDir(), "protocol/shared/types.ts"]
+  .sort();
 let content = `// Generated by tools/make_mod.ts. Don't edit.\n`;
 
-// Expose public variables from protocol/mod.ts
+// Expose public variables from protocol/shared/types.ts
 {
-  const fileName = "protocol/mod.ts";
+  const fileName = "protocol/shared/types.ts";
   const variables = (await doc(fileName)).filter((node) => {
     return node.kind === "variable";
   }).map((node) => node.name);