Merge pull request gagliardetto#124 from JumpCrypto/pires/fix_ws_clie…

…nt_goroutine_leak

Fix ws.ConnectWithOptions leaking goroutines

rpc/ws/client.go

@@ -37,6 +37,8 @@ type result interface{}
 type Client struct {
 	rpcURL                  string
 	conn                    *websocket.Conn
+	connCtx                 context.Context
+	connCtxCancel           context.CancelFunc
 	lock                    sync.RWMutex
 	subscriptionByRequestID map[uint64]*Subscription
 	subscriptionByWSSubID   map[uint64]*Subscription
@@ -83,12 +85,15 @@ func ConnectWithOptions(ctx context.Context, rpcEndpoint string, opt *Options) (
 		return nil, fmt.Errorf("new ws client: dial: %w", err)
 	}
 
+	c.connCtx, c.connCtxCancel = context.WithCancel(context.Background())
 	go func() {
 		c.conn.SetReadDeadline(time.Now().Add(pongWait))
 		c.conn.SetPongHandler(func(string) error { c.conn.SetReadDeadline(time.Now().Add(pongWait)); return nil })
 		ticker := time.NewTicker(pingPeriod)
 		for {
 			select {
+			case <-c.connCtx.Done():
+				return
 			case <-ticker.C:
 				c.sendPing()
 			}
@@ -111,17 +116,23 @@ func (c *Client) sendPing() {
 func (c *Client) Close() {
 	c.lock.Lock()
 	defer c.lock.Unlock()
+	c.connCtxCancel()
 	c.conn.Close()
 }
 
 func (c *Client) receiveMessages() {
 	for {
-		_, message, err := c.conn.ReadMessage()
-		if err != nil {
-			c.closeAllSubscription(err)
+		select {
+		case <-c.connCtx.Done():
 			return
+		default:
+			_, message, err := c.conn.ReadMessage()
+			if err != nil {
+				c.closeAllSubscription(err)
+				return
+			}
+			c.handleMessage(message)
 		}
-		c.handleMessage(message)
 	}
 }