Codebase list golang-github-nlopes-slack / 32bffe6
Started work on Disconnect method and fixed unmarshalling for channel creation. Michael Stewart authored 8 years ago Norberto Lopes committed 8 years ago
3 changed file(s) with 44 addition(s) and 18 deletion(s). Raw diff Collapse all Expand all
+9
-2
websocket.go less more
00 package slack
11
22 import (
3 "errors"
34 "log"
45 "sync"
56 "time"
2021 conn *websocket.Conn
2122 IncomingEvents chan SlackEvent
2223 outgoingMessages chan OutgoingMessage
24 keepRunning chan bool
25 isRunning bool
2326
2427 // Client is the main API, embedded
2528 Client
3740 pings: make(map[int]time.Time),
3841 IncomingEvents: make(chan SlackEvent, 50),
3942 outgoingMessages: make(chan OutgoingMessage, 20),
43 keepRunning: make(chan bool),
44 isRunning: true,
4045 }
4146 }
4247
4348 // Disconnect and wait, blocking until a successful disconnection.
4449 func (rtm *RTM) Disconnect() error {
45 log.Println("RTM::Disconnect not implemented!")
46 return nil
50 if !rtm.isRunning {
51 return errors.New("Invalid call to Disconnect - Slack API is already disconnected")
52 }
53 return rtm.killConnection(true)
4754 }
4855
4956 // Reconnect only makes sense if you've successfully disconnectd with Disconnect().
+11
-1
websocket_internals.go less more
1414 ConnectionCount int
1515 }
1616
17 type DisconnectedEvent struct{}
17 type DisconnectedEvent struct {
18 Intentional bool
19 }
1820
1921 type LatencyReport struct {
2022 Value time.Duration
2123 }
2224
2325 type InvalidAuthEvent struct{}
26
27 type UnmarshallingErrorEvent struct {
28 ErrorObj error
29 }
30
31 func (u UnmarshallingErrorEvent) Error() string {
32 return u.ErrorObj.Error()
33 }
+24
-15
websocket_managed_conn.go less more
6969 Info: info,
7070 }}
7171
72 killCh := make(chan bool, 3)
7372 connErrors := make(chan error, 10) // in case we get many such errors
7473
75 go rtm.keepalive(30*time.Second, conn, killCh, connErrors)
76 go rtm.handleIncomingEvents(conn, killCh, connErrors)
77 go rtm.handleOutgoingMessages(conn, killCh, connErrors)
74 go rtm.keepalive(30*time.Second, conn, connErrors)
75 go rtm.handleIncomingEvents(conn, connErrors)
76 go rtm.handleOutgoingMessages(conn, connErrors)
7877
7978 // Here, block and await for disconnection, if it ever happens.
8079 err = <-connErrors
8180
8281 rtm.Debugln("RTM connection error:", err)
83 rtm.IncomingEvents <- SlackEvent{"disconnected", &DisconnectedEvent{}}
84 killCh <- true // 3 child go-routines
85 killCh <- true
86 killCh <- true
87 }
82 rtm.killConnection(false)
83 }
84 }
85
86 func (rtm *RTM) killConnection(intentional bool) error {
87 rtm.mutex.Lock()
88 rtm.IncomingEvents <- SlackEvent{"disconnected", &DisconnectedEvent{}}
89 close(rtm.keepRunning)
90 err := rtm.conn.Close()
91 rtm.isRunning = false
92 rtm.mutex.Unlock()
93 return err
8894 }
8995
9096 func (rtm *RTM) startRTMAndDial() (*Info, *websocket.Conn, error) {
100106 return info, conn, err
101107 }
102108
103 func (rtm *RTM) keepalive(interval time.Duration, conn *websocket.Conn, killCh chan bool, errors chan error) {
109 func (rtm *RTM) keepalive(interval time.Duration, conn *websocket.Conn, errors chan error) {
104110 ticker := time.NewTicker(interval)
105111 defer ticker.Stop()
106112
107113 for {
108114 select {
109 case <-killCh:
115 case <-rtm.keepRunning:
110116 return
111117 case <-ticker.C:
112118 rtm.ping(conn, errors)
128134 }
129135 }
130136
131 func (rtm *RTM) handleOutgoingMessages(conn *websocket.Conn, killCh chan bool, errors chan error) {
137 func (rtm *RTM) handleOutgoingMessages(conn *websocket.Conn, errors chan error) {
132138 // we pass "conn" in case we do a reconnection, in that case we'll
133139 // have a new `conn` even though we're dealing with the same
134140 // incoming and outgoing channels for messages/events.
135141 for {
136142 select {
137 case <-killCh:
143 case <-rtm.keepRunning:
138144 return
139145
140146 case msg := <-rtm.outgoingMessages:
151157 }
152158 }
153159
154 func (rtm *RTM) handleIncomingEvents(conn *websocket.Conn, killCh chan bool, errors chan error) {
160 func (rtm *RTM) handleIncomingEvents(conn *websocket.Conn, errors chan error) {
155161 for {
156162
157163 select {
158 case <-killCh:
164 case <-rtm.keepRunning:
159165 return
160166 default:
161167 }
233239 if err != nil {
234240 rtm.Debugf("RTM Error unmarshalling %q event: %s", em.Type, err)
235241 rtm.Debugf(" -> Erroneous %q event: %s", em.Type, string(event))
242 rtm.IncomingEvents <- SlackEvent{"unmarshalling-error", &UnmarshallingErrorEvent{err}}
236243 return
237244 }
238245
239246 rtm.IncomingEvents <- SlackEvent{em.Type, recvEvent}
240247 } else {
241248 rtm.Debugf("RTM Error, received unmapped event %q: %s\n", em.Type, string(event))
249 err := fmt.Errorf("RTM Error, received unmapped event %q: %s\n", em.Type, string(event))
250 rtm.IncomingEvents <- SlackEvent{"unmarshalling-error", &UnmarshallingErrorEvent{err}}
242251 }
243252 }
244253 }