allow adding interceptors via request options (#70)

pkg/client/multi.go

@@ -55,8 +55,13 @@ func RequestMulti[ResponseType proto.Message](
 		requestID: rand.NewRequestID(),
 		resChan:   resChan,
 	}
+
+	reqInterceptors := getRequestInterceptors(
+		c.MultiRPCInterceptors,
+		getRequestOpts(i, c.ClientOpts, opts...).Interceptors,
+	)
 	m.handler = interceptors.ChainClientInterceptors[psrpc.ClientMultiRPCHandler](
-		c.MultiRPCInterceptors, i, m,
+		reqInterceptors, i, m,
 	)
 
 	if err = m.handler.Send(ctx, request, opts...); err != nil {

pkg/client/options.go

@@ -15,6 +15,8 @@
 package client
 
 import (
+	"fmt"
+
 	"github.com/livekit/psrpc"
 	"github.com/livekit/psrpc/internal/bus"
 	"github.com/livekit/psrpc/pkg/info"
@@ -58,3 +60,21 @@ func getRequestOpts(i *info.RequestInfo, options psrpc.ClientOpts, opts ...psrpc
 
 	return *o
 }
+
+func getRequestInterceptors[T psrpc.RequestInterceptor](base []T, as []any) []T {
+	if as == nil {
+		return base
+	}
+
+	c := make([]T, len(base), len(base)+len(as))
+	copy(c, base)
+	for _, a := range as {
+		i, ok := a.(T)
+		if !ok {
+			var ex T
+			panic(fmt.Sprintf("cannot use %T as %T in psrpc request", a, ex))
+		}
+		c = append(c, i)
+	}
+	return c
+}

pkg/client/rpc.go

@@ -57,8 +57,12 @@ func RequestSingle[ResponseType proto.Message](
 		hook(ctx, request, i.RPCInfo)
 	}
 
+	reqInterceptors := getRequestInterceptors(
+		c.RpcInterceptors,
+		getRequestOpts(i, c.ClientOpts, opts...).Interceptors,
+	)
 	handler := interceptors.ChainClientInterceptors[psrpc.ClientRPCHandler](
-		c.RpcInterceptors, i, newRPC[ResponseType](c, i),
+		reqInterceptors, i, newRPC[ResponseType](c, i),
 	)
 
 	res, err := handler(ctx, request, opts...)

pkg/client/stream.go

@@ -78,7 +78,7 @@ func OpenStream[SendType, RecvType proto.Message](
 		streamID,
 		c.Timeout,
 		&clientStream{c: c, i: i},
-		c.StreamInterceptors,
+		getRequestInterceptors(c.StreamInterceptors, o.Interceptors),
 		make(chan RecvType, c.ChannelSize),
 		map[string]chan struct{}{requestID: ackChan},
 	)

request.go

@@ -14,13 +14,18 @@
 
 package psrpc
 
-import "time"
+import (
+	"time"
+
+	"golang.org/x/exp/slices"
+)
 
 type RequestOption func(*RequestOpts)
 
 type RequestOpts struct {
 	Timeout       time.Duration
 	SelectionOpts SelectionOpts
+	Interceptors  []any
 }
 
 type SelectionOpts struct {
@@ -42,3 +47,16 @@ func WithSelectionOpts(opts SelectionOpts) RequestOption {
 		o.SelectionOpts = opts
 	}
 }
+
+type RequestInterceptor interface {
+	ClientRPCInterceptor | ClientMultiRPCInterceptor | StreamInterceptor
+}
+
+func WithRequestInterceptors[T RequestInterceptor](interceptors ...T) RequestOption {
+	return func(o *RequestOpts) {
+		o.Interceptors = slices.Grow(o.Interceptors, len(interceptors))
+		for _, i := range interceptors {
+			o.Interceptors = append(o.Interceptors, i)
+		}
+	}
+}