-
Notifications
You must be signed in to change notification settings - Fork 0
/
listener.go
136 lines (124 loc) · 2.85 KB
/
listener.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
package eventbus
import (
"bytes"
"encoding/json"
"errors"
"log"
"net"
)
//--------------------------------------------------
// Listener
//--------------------------------------------------
type Callback func(event *Event, err error)
type Listener interface {
On(t EventType, callback Callback)
}
type listenerImpl struct {
UUID string `json:"uuid"`
Type EventType `json:"type"`
Conn net.Conn `json:"-"`
}
//--------------------------------------------------
// Methods
//--------------------------------------------------
func (l *listenerImpl) On(t EventType, callback Callback) {
//convert in bytes
b, err := fromListener(l, t)
if err != nil {
callback(nil, err)
}
//log
log.Printf("Registering listener %s on %s", l.UUID, l.Type.Name)
//create frame
frameOut := NewFrame(RegReqFrameType, b)
//write frame in conn (registering)
err = WriteFrame(l.Conn, frameOut)
if err != nil {
callback(nil, err)
}
//
go (func() {
//loop
for {
//log
log.Printf("Waiting event: %s", l.Type.Name)
//read conn
frameIn, err := ReadFrame(l.Conn)
if err != nil {
callback(nil, err)
}
//switch action by frame type
switch frameIn.Type {
case RegRespFrameType:
//convert data in frame to event
e, err := toEvent(frameIn.Data)
if err != nil {
callback(nil, err)
}
//log
log.Printf("Execute callback")
//exec callback
callback(e, nil)
case PngReqFrameType:
//create frame
frameResp := NewFrame(PngRespFrameType, []byte{})
err := WriteFrame(l.Conn, frameResp)
if err != nil {
callback(nil, err)
}
case ErrRespFrameType:
callback(nil, errors.New(string(frameIn.Data)))
default:
callback(nil, errors.New("Fame type unknown"))
}
}
})()
}
//--------------------------------------------------
// Private Operations
//--------------------------------------------------
func fromListener(l *listenerImpl, etyp EventType) ([]byte, error) {
//
l.Type = etyp
//
buf := bytes.NewBuffer([]byte{})
enc := json.NewEncoder(buf)
err := enc.Encode(l)
if err != nil {
return nil, err
}
//
return buf.Bytes(), nil
}
func toListener(conn net.Conn, b []byte) (Listener, error) {
//create new listener
l := new(listenerImpl)
//convert bytes to listener
buf := bytes.NewBuffer(b)
dec := json.NewDecoder(buf)
err := dec.Decode(l)
if err != nil {
return nil, err
}
//set connection
l.Conn = conn
//result
return l, nil
}
//--------------------------------------------------
// Public Operations
//--------------------------------------------------
func NewListener(host string, port int, uuid string) (Listener, error) {
//setup conn
addr := net.TCPAddr{Port: port, IP: net.ParseIP(host)}
conn, err := net.Dial("tcp", addr.String())
if err != nil {
return nil, err
}
//instance listener
ln := new(listenerImpl)
ln.UUID = uuid
ln.Conn = conn
//result
return ln, nil
}