feat(autopipeline): AutoPipeliner implements UniversalClient (drop-in)

Widen the internal cmdableClient to UniversalClient and delegate the non-batched
surface (AddHook, Watch, Subscribe/PSubscribe/SSubscribe, PoolStats, Do,
AutoPipeline/AsyncAutoPipeline) to the underlying client, so an *AutoPipeliner
can be used anywhere a redis.UniversalClient is expected: data commands are
batched, everything else runs on the underlying client. Narrow Do to return *Cmd
to match the interface; drop the now-redundant .(*redis.Cmd) assertions in the
tests. Close() still closes only the autopipeliner, not the underlying client.

This lets tests (the command-suite matrix and maintnotifications e2e) parametrize
a single UniversalClient subject that includes the autopipeliner.

Author: ndyakov

autopipeline.go

@@ -202,10 +202,12 @@ func (cfg *AutoPipelineConfig) Validate() error {
 }
 
 // cmdableClient is an interface for clients that support pipelining.
-// Both Client and ClusterClient implement this interface.
+// Both Client and ClusterClient implement this interface. It embeds
+// UniversalClient (Cmdable + Process + Do + AddHook + Watch + Subscribe... +
+// Close + PoolStats) so the AutoPipeliner can delegate the non-batched surface
+// back to the underlying client and itself satisfy UniversalClient.
 type cmdableClient interface {
-	Cmdable
-	Process(ctx context.Context, cmd Cmder) error
+	UniversalClient
 	// processPipelineHook is the hook-wrapped []Cmder pipeline entry — the same
 	// method Pipeline.Exec is wired to (see Client.Pipeline). The flusher
 	// dispatches drained batches through it directly, skipping the per-batch
@@ -586,7 +588,7 @@ func newAutoPipeliner(pipeliner cmdableClient, config *AutoPipelineConfig, block
 // a blocking autopipeliner the call blocks until the command has executed; on a
 // deferred (async) one it returns immediately and the command's result
 // accessors (Err/Val/Result) block until it completes.
-func (ap *AutoPipeliner) Do(ctx context.Context, args ...interface{}) Cmder {
+func (ap *AutoPipeliner) Do(ctx context.Context, args ...interface{}) *Cmd {
 	cmd := NewCmd(ctx, args...)
 	if len(args) == 0 {
 		cmd.SetErr(errDoNoArgs)
@@ -624,6 +626,61 @@ func (ap *AutoPipeliner) Process(ctx context.Context, cmd Cmder) error {
 	return ap.cmdable(ctx, cmd)
 }
 
+// The methods below complete the UniversalClient surface by delegating to the
+// underlying client. They are NOT autopipelined — pub/sub, transactions (Watch),
+// hooks, Do and pool stats cannot be batched — so an AutoPipeliner used as a
+// UniversalClient batches only the typed data commands; everything here runs on
+// the underlying client exactly as it would there.
+//
+// Note on lifecycle: Close() (defined elsewhere) closes the AUTOPIPELINER —
+// drains in-flight batches and stops flushers — but does NOT close the
+// underlying client, whose lifecycle is owned by whoever created it.
+
+// AddHook adds a hook to the underlying client. Autopipelined batches are hooked
+// too, since dispatch goes through the hook-wrapped pipeline entry.
+func (ap *AutoPipeliner) AddHook(hook Hook) { ap.pipeliner.AddHook(hook) }
+
+// Watch runs a transactional function on the underlying client (not batched).
+func (ap *AutoPipeliner) Watch(ctx context.Context, fn func(*Tx) error, keys ...string) error {
+	return ap.pipeliner.Watch(ctx, fn, keys...)
+}
+
+// Subscribe opens a pub/sub on the underlying client (not batched — pub/sub
+// needs a dedicated connection).
+func (ap *AutoPipeliner) Subscribe(ctx context.Context, channels ...string) *PubSub {
+	return ap.pipeliner.Subscribe(ctx, channels...)
+}
+
+// PSubscribe opens a pattern pub/sub on the underlying client (not batched).
+func (ap *AutoPipeliner) PSubscribe(ctx context.Context, channels ...string) *PubSub {
+	return ap.pipeliner.PSubscribe(ctx, channels...)
+}
+
+// SSubscribe opens a sharded pub/sub on the underlying client (not batched).
+func (ap *AutoPipeliner) SSubscribe(ctx context.Context, channels ...string) *PubSub {
+	return ap.pipeliner.SSubscribe(ctx, channels...)
+}
+
+// PoolStats returns the underlying client's connection pool statistics.
+func (ap *AutoPipeliner) PoolStats() *PoolStats { return ap.pipeliner.PoolStats() }
+
+// AutoPipeline delegates to the underlying client, which returns its cached
+// autopipeliner (typically this same instance). Present to satisfy the
+// UniversalClient surface.
+func (ap *AutoPipeliner) AutoPipeline(config ...*AutoPipelineConfig) (*AutoPipeliner, error) {
+	return ap.pipeliner.AutoPipeline(config...)
+}
+
+// AsyncAutoPipeline delegates to the underlying client. Present to satisfy the
+// UniversalClient surface.
+func (ap *AutoPipeliner) AsyncAutoPipeline(config ...*AutoPipelineConfig) (*AutoPipeliner, error) {
+	return ap.pipeliner.AsyncAutoPipeline(config...)
+}
+
+// validate AutoPipeliner implements UniversalClient (drop-in for the real
+// clients; non-data operations delegate to the underlying client).
+var _ UniversalClient = (*AutoPipeliner)(nil)
+
 // AutoFuture is the handle returned by Submit. Call Wait (or Result on the
 // command after Wait) once the result is needed; it blocks only until the
 // command's batch has executed.

autopipeline_accessor_test.go

@@ -66,10 +66,10 @@ func TestAutoPipelineSecondaryAccessors(t *testing.T) {
 			fail(err != nil || u != 1)
 
 			// Generic Cmd via Do() — the whole *Cmd type was missing await().
-			doCmd := ap.Do(ctx, "GET", sk).(*redis.Cmd)
+			doCmd := ap.Do(ctx, "GET", sk)
 			v, err := doCmd.Int()
 			fail(err != nil || v != 12345)
-			txt, err := ap.Do(ctx, "GET", sk).(*redis.Cmd).Text()
+			txt, err := ap.Do(ctx, "GET", sk).Text()
 			fail(err != nil || txt != "12345")
 
 			// SliceCmd.Val()/Result() — previously missing await().

autopipeline_blocking_test.go

@@ -41,7 +41,7 @@ var _ = Describe("AutoPipeline Blocking Commands", func() {
 		// ap.Do returns a generic *redis.Cmd; use its typed accessors.
 		start := time.Now()
 		result := ap.Do(ctx, "BLPOP", "list", "1")
-		val, err := result.(*redis.Cmd).StringSlice()
+		val, err := result.StringSlice()
 		elapsed := time.Since(start)
 
 		Expect(err).NotTo(HaveOccurred())
@@ -61,15 +61,15 @@ var _ = Describe("AutoPipeline Blocking Commands", func() {
 		brpopCmd := ap.Do(ctx, "BRPOP", "list3", "1")
 
 		// Get results — ap.Do returns generic *redis.Cmd; use typed accessors.
-		blpopVal, err := blpopCmd.(*redis.Cmd).StringSlice()
+		blpopVal, err := blpopCmd.StringSlice()
 		Expect(err).NotTo(HaveOccurred())
 		Expect(blpopVal).To(Equal([]string{"list3", "a"}))
 
-		getVal, err := getCmd.(*redis.Cmd).Text()
+		getVal, err := getCmd.Text()
 		Expect(err).NotTo(HaveOccurred())
 		Expect(getVal).To(Equal("value1"))
 
-		brpopVal, err := brpopCmd.(*redis.Cmd).StringSlice()
+		brpopVal, err := brpopCmd.StringSlice()
 		Expect(err).NotTo(HaveOccurred())
 		Expect(brpopVal).To(Equal([]string{"list3", "c"}))
 	})

autopipeline_cmdable_test.go

@@ -131,11 +131,11 @@ var _ = Describe("AutoPipeline Cmdable Interface", func() {
 		getCmd := ap.Do(ctx, "GET", "custom_key")
 
 		// Get results
-		setVal, err := setCmd.(*redis.Cmd).Result()
+		setVal, err := setCmd.Result()
 		Expect(err).NotTo(HaveOccurred())
 		Expect(setVal).To(Equal("OK"))
 
-		getVal, err := getCmd.(*redis.Cmd).Result()
+		getVal, err := getCmd.Result()
 		Expect(err).NotTo(HaveOccurred())
 		Expect(getVal).To(Equal("custom_value"))
 	})

autopipeline_universal_test.go

@@ -50,3 +50,39 @@ func TestUniversalClientAutoPipeline(t *testing.T) {
 		t.Fatal("Ring via UniversalClient.AutoPipeline should return an error")
 	}
 }
+
+// TestAutoPipelinerAsUniversalClient verifies an *AutoPipeliner is a drop-in
+// UniversalClient: data commands are batched through it, while the non-batched
+// surface (PoolStats, Do, ...) delegates to the underlying client.
+func TestAutoPipelinerAsUniversalClient(t *testing.T) {
+	ctx := context.Background()
+	client := redis.NewClient(&redis.Options{Addr: ":6379"})
+	defer client.Close()
+	if err := client.Ping(ctx).Err(); err != nil {
+		t.Skipf("no redis: %v", err)
+	}
+	apc, err := client.AutoPipeline(nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer apc.Close()
+
+	// Use the autopipeliner strictly through the UniversalClient interface.
+	var uc redis.UniversalClient = apc
+
+	// Data commands: batched by the autopipeliner.
+	if v := uc.Set(ctx, "ucap:k", "v", 0).Val(); v != "OK" {
+		t.Fatalf("Set = %q, want OK", v)
+	}
+	if v := uc.Get(ctx, "ucap:k").Val(); v != "v" {
+		t.Fatalf("Get = %q, want v", v)
+	}
+	// Non-batched surface delegates to the underlying client.
+	if uc.PoolStats() == nil {
+		t.Fatal("PoolStats() returned nil via UniversalClient")
+	}
+	if err := uc.Do(ctx, "PING").Err(); err != nil {
+		t.Fatalf("Do(PING): %v", err)
+	}
+	uc.Del(ctx, "ucap:k")
+}