lib9p

commit 4bcdaf78db4a1251b26c1b1b7780c85d381a7de2
parent fd70e85401106c582dc0fec8f46fab95929b5cd9
Author: Matsuda Kenji <info@mtkn.jp>
Date:   Sun,  8 Oct 2023 06:57:00 +0900

add ctxDone channel to clientReq

Diffstat:
M
client.go
 | 
6
+++---
M
req.go
 | 
14
++++++++------

2 files changed, 11 insertions(+), 9 deletions(-)
diff --git a/client.go b/client.go
@@ -65,10 +65,11 @@ func (c *Client) runListener(ctx context.Context, r io.Reader) <-chan error {
 				}
 				go func() {
 					defer close(req.rmsgc)
+					defer c.rPool.delete(msg.Tag())
 					select {
 					case req.rmsgc <- msg:
-						c.rPool.delete(msg.Tag())
 					case <-ctx.Done():
+					case <-req.ctxDone:
 					}
 				}()
 			}
@@ -101,7 +102,7 @@ func (c *Client) runSpeaker(ctx context.Context, w io.Writer) chan<- Msg {
 
 // tag of tmsg is managed by the library.
 func (c *Client) transact(ctx context.Context, tmsg Msg) (<-chan Msg, <-chan error) {
-	req, err := c.rPool.add(tmsg)
+	req, err := c.rPool.add(ctx, tmsg)
 	if err != nil {
 		errc := make(chan error, 1)
 		defer close(errc)
@@ -136,7 +137,6 @@ func (c *Client) Version(ctx context.Context, mSize uint32, version string) (*RV
 	case err := <-errc:
 		return nil, err
 	case <-ctx.Done():
-		// TODO: destroy clientReq.
 		return nil, fmt.Errorf("wait transaction: %w", ctx.Err())
 	}
 }
diff --git a/req.go b/req.go
@@ -1,6 +1,7 @@
 package lib9p
 
 import (
+	"context"
 	"fmt"
 	"sync"
 )
@@ -66,6 +67,7 @@ type clientReq struct {
 	pool  *clientReqPool
 	rmsgc chan Msg
 	errc  chan error
+	ctxDone <-chan struct{}
 }
 
 type clientReqPool struct {
@@ -92,7 +94,7 @@ func (rp *clientReqPool) nextTag() (uint16, error) {
 	return 0, fmt.Errorf("run out of tag")
 }
 
-func (rp *clientReqPool) add(msg Msg) (*clientReq, error) {
+func (rp *clientReqPool) add(ctx context.Context, msg Msg) (*clientReq, error) {
 	var tag uint16
 	if _, ok := msg.(*TVersion); ok {
 		tag = NOTAG
@@ -108,10 +110,11 @@ func (rp *clientReqPool) add(msg Msg) (*clientReq, error) {
 
 	msg.SetTag(tag)
 	req := &clientReq{
-		tag: tag,
-		pool: rp,
+		tag:   tag,
+		pool:  rp,
 		rmsgc: make(chan Msg),
-		errc: make(chan error),
+		errc:  make(chan error),
+		ctxDone: ctx.Done(),
 	}
 	rp.m[tag] = req
 	return req, nil
@@ -141,4 +144,4 @@ func (rp *clientReqPool) String() string {
 		s = "[]"
 	}
 	return s
-}
-\ No newline at end of file
+}