-
Notifications
You must be signed in to change notification settings - Fork 260
/
handler.go
141 lines (118 loc) · 2.66 KB
/
handler.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
133
134
135
136
137
138
139
140
141
package gosocketio
import (
"encoding/json"
"github.com/graarh/golang-socketio/protocol"
"sync"
"reflect"
)
const (
OnConnection = "connection"
OnDisconnection = "disconnection"
OnError = "error"
)
/**
System handler function for internal event processing
*/
type systemHandler func(c *Channel)
/**
Contains maps of message processing functions
*/
type methods struct {
messageHandlers map[string]*caller
messageHandlersLock sync.RWMutex
onConnection systemHandler
onDisconnection systemHandler
}
/**
create messageHandlers map
*/
func (m *methods) initMethods() {
m.messageHandlers = make(map[string]*caller)
}
/**
Add message processing function, and bind it to given method
*/
func (m *methods) On(method string, f interface{}) error {
c, err := newCaller(f)
if err != nil {
return err
}
m.messageHandlersLock.Lock()
defer m.messageHandlersLock.Unlock()
m.messageHandlers[method] = c
return nil
}
/**
Find message processing function associated with given method
*/
func (m *methods) findMethod(method string) (*caller, bool) {
m.messageHandlersLock.RLock()
defer m.messageHandlersLock.RUnlock()
f, ok := m.messageHandlers[method]
return f, ok
}
func (m *methods) callLoopEvent(c *Channel, event string) {
if m.onConnection != nil && event == OnConnection {
m.onConnection(c)
}
if m.onDisconnection != nil && event == OnDisconnection {
m.onDisconnection(c)
}
f, ok := m.findMethod(event)
if !ok {
return
}
f.callFunc(c, &struct{}{})
}
/**
Check incoming message
On ack_resp - look for waiter
On ack_req - look for processing function and send ack_resp
On emit - look for processing function
*/
func (m *methods) processIncomingMessage(c *Channel, msg *protocol.Message) {
switch msg.Type {
case protocol.MessageTypeEmit:
f, ok := m.findMethod(msg.Method)
if !ok {
return
}
if !f.ArgsPresent {
f.callFunc(c, &struct{}{})
return
}
data := f.getArgs()
err := json.Unmarshal([]byte(msg.Args), &data)
if err != nil {
return
}
f.callFunc(c, data)
case protocol.MessageTypeAckRequest:
f, ok := m.findMethod(msg.Method)
if !ok || !f.Out {
return
}
var result []reflect.Value
if f.ArgsPresent {
//data type should be defined for unmarshall
data := f.getArgs()
err := json.Unmarshal([]byte(msg.Args), &data)
if err != nil {
return
}
result = f.callFunc(c, data)
} else {
result = f.callFunc(c, &struct{}{})
}
ack := &protocol.Message{
Type: protocol.MessageTypeAckResponse,
AckId: msg.AckId,
}
send(ack, c, result[0].Interface())
case protocol.MessageTypeAckResponse:
waiter, err := c.ack.getWaiter(msg.AckId)
if err == nil {
waiter <- msg.Args
}
}
}