magic

contrib/codecs/http/client.go

@@ -66,6 +66,10 @@ func (c *Client) SetHeader(key string, value string) {
 	c.headers.Set(key, value)
 }
 
+func (c *Client) Closed() <-chan struct{} {
+	return make(chan struct{})
+}
+
 func (c *Client) Do(ctx context.Context, result any, method string, params any) error {
 	req, err := codec.NewRequest(ctx, codec.NewId(c.id.Add(1)), method, params)
 	if err != nil {

contrib/codecs/inproc/client.go

@@ -14,6 +14,9 @@ type Client struct {
 	p *clientutil.IdReply
 	c *Codec
 
+	ctx context.Context
+	cn  context.CancelFunc
+
 	m       codec.Middlewares
 	handler codec.Handler
 	mu      sync.Mutex
@@ -34,12 +37,19 @@ func NewClient(c *Codec, handler codec.Handler) *Client {
 		c:       c,
 		handler: handler,
 	}
+	cl.ctx, cl.cn = context.WithCancel(context.Background())
 	go cl.listen()
+
 	return cl
 }
 
+func (c *Client) Closed() <-chan struct{} {
+	return c.ctx.Done()
+}
+
 func (c *Client) listen() error {
 	var msg json.RawMessage
+	defer c.cn()
 	for {
 		err := json.NewDecoder(c.c.rd).Decode(&msg)
 		if err != nil {

contrib/codecs/rdwr/client.go

@@ -47,6 +47,10 @@ func (c *Client) SetHandlerPeer(pi codec.PeerInfo) {
 	c.handlerPeer = pi
 }
 
+func (c *Client) Closed() <-chan struct{} {
+	return c.ctx.Done()
+}
+
 func (c *Client) Mount(h codec.Middleware) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
@@ -58,6 +62,7 @@ func (c *Client) Mount(h codec.Middleware) {
 
 func (c *Client) listen() error {
 	var msg json.RawMessage
+	defer c.cn()
 	for {
 		err := json.NewDecoder(c.rd).Decode(&msg)
 		if err != nil {

contrib/codecs/redis/client.go

@@ -54,6 +54,10 @@ func NewClient(c redis.UniversalClient, domain string) *Client {
 	return cl
 }
 
+func (c *Client) Closed() <-chan struct{} {
+	return c.ctx.Done()
+}
+
 func (c *Client) SetHandlerPeer(pi codec.PeerInfo) {
 	c.handlerPeer = pi
 }

pkg/codec/jrpc.go

@@ -33,12 +33,18 @@ type Mounter interface {
 	Mount(Middleware)
 }
 
+type Closeder interface {
+	Closed() <-chan struct{}
+}
+
 type Conn interface {
 	Doer
 	BatchCaller
-	io.Closer
 
 	Mounter
+
+	io.Closer
+	Closeder
 }
 
 type StreamingConn interface {