BREAKING: remove reader and writer from Connection

connection.ts

@@ -1,8 +1,9 @@
-import { sendCommand } from "./protocol/mod.ts";
-import type { RedisReply, RedisValue } from "./protocol/mod.ts";
+import { readReply, sendCommand, sendCommands } from "./protocol/mod.ts";
+import type { Command, RedisReply, RedisValue } from "./protocol/mod.ts";
 import type { Backoff } from "./backoff.ts";
 import { exponentialBackoff } from "./backoff.ts";
 import { ErrorReplyError, isRetriableError } from "./errors.ts";
+import { kUnstablePipeline, kUnstableReadReply } from "./internal/symbols.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 {
@@ -22,8 +23,6 @@ export interface SendCommandOptions {
 }
 
 export interface Connection {
-  reader: BufReader;
-  writer: BufWriter;
   isClosed: boolean;
   isConnected: boolean;
   close(): void;
@@ -34,6 +33,16 @@ export interface Connection {
     args?: Array<RedisValue>,
     options?: SendCommandOptions,
   ): Promise<RedisReply>;
+  /**
+   * @private
+   */
+  [kUnstableReadReply](returnsUint8Arrays?: boolean): Promise<RedisReply>;
+  /**
+   * @private
+   */
+  [kUnstablePipeline](
+    commands: Array<Command>,
+  ): Promise<Array<RedisReply | ErrorReplyError>>;
 }
 
 export interface RedisConnectionOptions {
@@ -64,8 +73,8 @@ interface Command {
 
 export class RedisConnection implements Connection {
   name: string | null = null;
-  reader!: BufReader;
-  writer!: BufWriter;
+  private reader!: BufReader;
+  private writer!: BufWriter;
   private closer!: Closer;
   private maxRetryCount = 10;
 
@@ -149,6 +158,14 @@ export class RedisConnection implements Connection {
     return promise;
   }
 
+  [kUnstableReadReply](returnsUint8Arrays?: boolean): Promise<RedisReply> {
+    return readReply(this.reader, returnsUint8Arrays);
+  }
+
+  [kUnstablePipeline](commands: Array<Command>) {
+    return sendCommands(this.writer, this.reader, commands);
+  }
+
   /**
    * Connect to Redis server
    */

internal/symbols.ts

@@ -0,0 +1,9 @@
+/**
+ * @private
+ */
+export const kUnstableReadReply = Symbol("deno-redis.readReply");
+
+/**
+ * @private
+ */
+export const kUnstablePipeline = Symbol("deno-redis.pipeline");

pipeline.ts

@@ -9,6 +9,7 @@ import {
   sendCommands,
 } from "./protocol/mod.ts";
 import { create, Redis } from "./redis.ts";
+import { kUnstablePipeline } from "./internal/symbols.ts";
 import {
   Deferred,
   deferred,
@@ -92,7 +93,7 @@ export class PipelineExecutor implements CommandExecutor {
   private dequeue(): void {
     const [e] = this.queue;
     if (!e) return;
-    sendCommands(this.connection.writer, this.connection.reader, e.commands)
+    this.connection[kUnstablePipeline](e.commands)
       .then(e.d.resolve)
       .catch(e.d.reject)
       .finally(() => {

protocol/command.ts

@@ -98,20 +98,22 @@ export async function sendCommand(
   return readReply(reader, returnUint8Arrays);
 }
 
+export interface Command {
+  command: string;
+  args: RedisValue[];
+  returnUint8Arrays?: boolean;
+}
+
 export async function sendCommands(
   writer: BufWriter,
   reader: BufReader,
-  commands: {
-    command: string;
-    args: RedisValue[];
-    returnUint8Arrays?: boolean;
-  }[],
-): Promise<unknown[]> {
+  commands: Command[],
+): Promise<(RedisReply | ErrorReplyError)[]> {
   for (const { command, args } of commands) {
     await writeRequest(writer, command, args);
   }
   await writer.flush();
-  const ret: unknown[] = [];
+  const ret: (RedisReply | ErrorReplyError)[] = [];
   for (let i = 0; i < commands.length; i++) {
     try {
       const rep = await readReply(reader, commands[i].returnUint8Arrays);

protocol/mod.ts

@@ -14,4 +14,5 @@ export type {
 
 export { okReply, readArrayReply, readReply } from "./reply.ts";
 
+export type { Command } from "./command.ts";
 export { sendCommand, sendCommands } from "./command.ts";

pubsub.ts

@@ -3,6 +3,7 @@ import { InvalidStateError } from "./errors.ts";
 import type { Binary } from "./protocol/mod.ts";
 import { readArrayReply } from "./protocol/mod.ts";
 import { decoder } from "./protocol/_util.ts";
+import { kUnstableReadReply } from "./internal/symbols.ts";
 
 type DefaultMessageType = string;
 type ValidMessageType = string | string[];
@@ -96,11 +97,7 @@ class RedisSubscriptionImpl<
           T,
         ];
         try {
-          // TODO: `readArrayReply` should not be called directly here
-          rep = (await readArrayReply(
-            connection.reader,
-            binaryMode,
-          )) as typeof rep;
+          rep = await connection[kUnstableReadReply](binaryMode) as typeof rep;
         } catch (err) {
           if (err instanceof Deno.errors.BadResource) {
             // Connection already closed.