-
Notifications
You must be signed in to change notification settings - Fork 11
/
send.go
78 lines (64 loc) · 1.37 KB
/
send.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
package shadiaosocketio
import (
"errors"
"github.com/Baiguoshuai1/shadiaosocketio/protocol"
"log"
"time"
)
var (
ErrorSendTimeout = errors.New("timeout")
ErrorSocketOverflood = errors.New("socket overflood")
)
/*
*
Send message packet to socket
*/
func send(c *Channel, msg *protocol.Message) error {
defer func() {
if r := recover(); r != nil {
log.Println("socket.io send panic: ", r)
}
}()
if !c.IsAlive() {
return nil
}
out := protocol.GetMsgPacket(msg)
if len(c.out) == queueBufferSize {
return ErrorSocketOverflood
}
c.out <- out
return nil
}
func (c *Channel) Emit(method string, args ...interface{}) error {
msg := &protocol.Message{
Type: protocol.EVENT,
AckId: -1,
Method: method,
Nsp: protocol.DefaultNsp,
Args: args,
}
return send(c, msg)
}
func (c *Channel) Ack(method string, timeout time.Duration, args ...interface{}) (interface{}, error) {
msg := &protocol.Message{
Type: protocol.EVENT,
AckId: c.ack.getNextId(),
Method: method,
Nsp: protocol.DefaultNsp,
Args: args,
}
waiter := make(chan interface{})
c.ack.addWaiter(msg.AckId, waiter)
err := send(c, msg)
if err != nil {
c.ack.removeWaiter(msg.AckId)
}
select {
case result := <-waiter:
c.ack.removeWaiter(msg.AckId)
return result, nil
case <-time.After(timeout):
c.ack.removeWaiter(msg.AckId)
return nil, ErrorSendTimeout
}
}