-
Notifications
You must be signed in to change notification settings - Fork 1
/
client.go
132 lines (108 loc) · 2.4 KB
/
client.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
package etp
import (
"context"
"errors"
"fmt"
"sync"
"github.com/txix-open/etp/v3/internal"
"nhooyr.io/websocket"
)
var (
ErrClientClosed = errors.New("client closed")
)
type Client struct {
mux *mux
idGenerator *internal.IdGenerator
opts *clientOptions
conn *Conn
lock sync.Locker
}
func NewClient(opts ...ClientOption) *Client {
options := defaultClientOptions()
for _, opt := range opts {
opt(options)
}
return &Client{
mux: newMux(),
idGenerator: internal.NewIdGenerator(),
opts: options,
lock: &sync.Mutex{},
conn: nil,
}
}
func (c *Client) On(event string, handler Handler) *Client {
c.mux.On(event, handler)
return c
}
func (c *Client) OnConnect(handler ConnectHandler) *Client {
c.mux.OnConnect(handler)
return c
}
func (c *Client) OnDisconnect(handler DisconnectHandler) *Client {
c.mux.OnDisconnect(handler)
return c
}
func (c *Client) OnError(handler ErrorHandler) *Client {
c.mux.OnError(handler)
return c
}
func (c *Client) OnUnknownEvent(handler Handler) *Client {
c.mux.OnUnknownEvent(handler)
return c
}
func (c *Client) Dial(ctx context.Context, url string) error {
c.lock.Lock()
defer c.lock.Unlock()
if c.conn != nil {
return errors.New("already connected")
}
ws, resp, err := websocket.Dial(ctx, url, c.opts.dialOptions)
if err != nil {
return fmt.Errorf("websocket dial: %w", err)
}
ws.SetReadLimit(c.opts.readLimit)
id := c.idGenerator.Next()
conn := newConn(id, resp.Request, ws)
c.conn = conn
keeper := newKeeper(conn, c.mux)
go func() {
defer func() {
_ = ws.CloseNow()
}()
keeper.Serve(context.Background())
c.lock.Lock()
defer c.lock.Unlock()
if c.conn != nil && c.conn.Id() == id {
c.conn = nil
}
}()
return nil
}
func (c *Client) Emit(ctx context.Context, event string, data []byte) error {
if c.conn == nil {
return ErrClientClosed
}
return c.conn.Emit(ctx, event, data)
}
func (c *Client) EmitWithAck(ctx context.Context, event string, data []byte) ([]byte, error) {
if c.conn == nil {
return nil, ErrClientClosed
}
return c.conn.EmitWithAck(ctx, event, data)
}
func (c *Client) Ping(ctx context.Context) error {
if c.conn == nil {
return ErrClientClosed
}
return c.conn.Ping(ctx)
}
func (c *Client) Close() error {
c.lock.Lock()
defer c.lock.Unlock()
if c.conn == nil {
return ErrClientClosed
}
err := c.conn.Close()
c.conn = nil
return err
}