u

diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index b5908e736b916e8f45c195d1d9fcdb8c9f4399b5..b6b3c31c6e12558395dd5e794fe3be920e64f3a0 100644 (file)
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -33,4 +33,5 @@ jobs:
         go test -count 1 -timeout 30s -v -race github.com/qydysky/part/msgq
         go test -count 1 -timeout 5s -v -race github.com/qydysky/part/sync
         go test -count 1 -timeout 10s -v -race github.com/qydysky/part/web
+        go test -count 1 -timeout 10s -v -run "Test_Client" -race github.com/qydysky/part/websocket
         CC=gcc;CXX=g++;CGO_ENABLED=1;GOOS=linux go test -count 1 -timeout 10s -v -race -ldflags '-extldflags=-static -extldflags=-lm' github.com/qydysky/part/sql

diff --git a/websocket/Client.go b/websocket/Client.go
index 37f43a5d48cf834197a21be87779477fca333296..4d3d646fa814b1cf0efc44d031cee7ac05e1e29e 100644 (file)
--- a/websocket/Client.go
+++ b/websocket/Client.go
@@ -5,7 +5,7 @@ import (
        "io"
        "net/http"
        "net/url"
-       "sync/atomic"
+       "sync"
        "time"
 
        "github.com/gorilla/websocket"
@@ -29,9 +29,10 @@ type Client struct {
        Func_normal_close func()
        Func_abort_close  func()
 
-       close atomic.Bool
+       close bool
+       err   error
 
-       err error
+       l sync.RWMutex
 }
 
 type WsMsg struct {
@@ -45,7 +46,7 @@ type Ping struct {
        had_pong bool
 }
 
-func New_client(config Client) (*Client, error) {
+func New_client(config *Client) (*Client, error) {
        tmp := Client{
                TO:                300 * 1000,
                Func_normal_close: func() {},
@@ -100,7 +101,9 @@ func (o *Client) Handle() (*msgq.MsgType[*WsMsg], error) {
                dial.Proxy = proxy
        }
        c, response, err := dial.Dial(o.Url, tmp_Header)
-
+       if err := c.SetWriteDeadline(time.Now().Add(time.Duration(o.TO * int(time.Millisecond)))); err != nil {
+               o.error(err)
+       }
        if err != nil {
                e := err.Error()
                if response != nil {
@@ -121,10 +124,14 @@ func (o *Client) Handle() (*msgq.MsgType[*WsMsg], error) {
 
        // rec
        go func() {
-               defer o.msg.ClearAll()
-               defer o.close.Store(true)
+               defer func() {
+                       o.msg.ClearAll()
+                       o.l.Lock()
+                       o.close = true
+                       o.l.Unlock()
+               }()
+
                for {
-                       c.SetReadDeadline(time.Now().Add(time.Duration(o.TO * int(time.Millisecond))))
                        msg_type, message, err := c.ReadMessage()
                        if err != nil {
                                if e, ok := err.(*websocket.CloseError); ok {
@@ -135,7 +142,7 @@ func (o *Client) Handle() (*msgq.MsgType[*WsMsg], error) {
                                                o.Func_abort_close()
                                        default:
                                        }
-                                       o.err = errors.Join(o.err, err)
+                                       o.error(err)
                                }
                                return
                        }
@@ -170,26 +177,25 @@ func (o *Client) Handle() (*msgq.MsgType[*WsMsg], error) {
                        wm.Type = websocket.TextMessage
                }
                if err := c.WriteMessage(wm.Type, wm.Msg); err != nil {
-                       o.err = errors.Join(o.err, err)
+                       o.error(err)
                        o.msg.ClearAll()
                        return true
                }
                if wm.Type == websocket.PingMessage {
                        time.AfterFunc(time.Duration(o.TO*int(time.Millisecond)), func() {
                                if time.Now().UnixMilli() > o.pingT+int64(o.TO) {
-                                       o.err = errors.Join(o.err, errors.New("PongFail"))
+                                       o.error(errors.New("PongFail"))
                                        o.Close()
                                }
                        })
                }
-               o.err = errors.Join(o.err, c.SetWriteDeadline(time.Now().Add(time.Duration(o.TO*int(time.Millisecond)))))
                return false
        })
 
        // close
        o.msg.Pull_tag_only(`close`, func(_ *WsMsg) (disable bool) {
                if err := c.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, o.Msg_normal_close)); err != nil {
-                       o.err = errors.Join(o.err, err)
+                       o.error(err)
                        o.msg.ClearAll()
                }
                return true
@@ -211,10 +217,22 @@ func (o *Client) Close() {
        o.msg.PushLock_tag(`close`, nil)
 }
 
-func (o *Client) Isclose() bool {
-       return o.close.Load()
+func (o *Client) Isclose() (isclose bool) {
+       o.l.RLock()
+       isclose = o.close
+       o.l.RUnlock()
+       return
+}
+
+func (o *Client) Error() (e error) {
+       o.l.RLock()
+       e = o.err
+       o.l.RUnlock()
+       return
 }
 
-func (o *Client) Error() error {
-       return o.err
+func (o *Client) error(e error) {
+       o.l.Lock()
+       o.err = errors.Join(o.err, e)
+       o.l.Unlock()
 }

diff --git a/websocket/Client_test.go b/websocket/Client_test.go
index c7376b1dd2b816a1036d3d07ba37e1833cfa0fd4..d2fe5a50fa7f4d48909167e940ca9f81515fa60a 100644 (file)
--- a/websocket/Client_test.go
+++ b/websocket/Client_test.go
@@ -47,7 +47,7 @@ func Test_Client(t *testing.T) {
 
        time.Sleep(time.Second)
 
-       c, e := New_client(Client{
+       c, e := New_client(&Client{
                Url: "ws://127.0.0.1:10888/ws",
                Func_normal_close: func() {
                        t.Log("close")