diff options
author | Wim <wim@42.be> | 2019-09-07 22:46:58 +0200 |
---|---|---|
committer | GitHub <noreply@github.com> | 2019-09-07 22:46:58 +0200 |
commit | a3bee01e0af3394c19360b98fd2db1b647f49299 (patch) | |
tree | ffc5778361d55d592a718354a37c9251e75fc7f6 /vendor/github.com/nlopes/slack/websocket_managed_conn.go | |
parent | 1dc93ec4f001edd01daccbe408767d4878be25a3 (diff) | |
download | matterbridge-msglm-a3bee01e0af3394c19360b98fd2db1b647f49299.tar.gz matterbridge-msglm-a3bee01e0af3394c19360b98fd2db1b647f49299.tar.bz2 matterbridge-msglm-a3bee01e0af3394c19360b98fd2db1b647f49299.zip |
Update dependencies (#886)
Diffstat (limited to 'vendor/github.com/nlopes/slack/websocket_managed_conn.go')
-rw-r--r-- | vendor/github.com/nlopes/slack/websocket_managed_conn.go | 142 |
1 files changed, 84 insertions, 58 deletions
diff --git a/vendor/github.com/nlopes/slack/websocket_managed_conn.go b/vendor/github.com/nlopes/slack/websocket_managed_conn.go index 62157910..8b3b3833 100644 --- a/vendor/github.com/nlopes/slack/websocket_managed_conn.go +++ b/vendor/github.com/nlopes/slack/websocket_managed_conn.go @@ -10,6 +10,8 @@ import ( "time" "github.com/gorilla/websocket" + "github.com/nlopes/slack/internal/errorsx" + "github.com/nlopes/slack/internal/timex" ) // ManageConnection can be called on a Slack RTM instance returned by the @@ -38,6 +40,7 @@ func (rtm *RTM) ManageConnection() { if info, conn, err = rtm.connect(connectionCount, rtm.useRTMStart); err != nil { // when the connection is unsuccessful its fatal, and we need to bail out. rtm.Debugf("Failed to connect with RTM on try %d: %s", connectionCount, err) + rtm.disconnect() return } @@ -45,7 +48,6 @@ func (rtm *RTM) ManageConnection() { // and conn. rtm.mu.Lock() rtm.conn = conn - rtm.isConnected = true rtm.info = info rtm.mu.Unlock() @@ -56,20 +58,19 @@ func (rtm *RTM) ManageConnection() { rtm.Debugf("RTM connection succeeded on try %d", connectionCount) - keepRunning := make(chan bool) - // we're now connected (or have failed fatally) so we can set up - // listeners - go rtm.handleIncomingEvents(keepRunning) + // we're now connected so we can set up listeners + go rtm.handleIncomingEvents() // this should be a blocking call until the connection has ended - rtm.handleEvents(keepRunning) + rtm.handleEvents() - // after being disconnected we need to check if it was intentional - // if not then we should try to reconnect - if rtm.wasIntentional { + select { + case <-rtm.disconnected: + // after handle events returns we need to check if we're disconnected return + default: + // otherwise continue and run the loop again to reconnect } - // else continue and run the loop again to connect } } @@ -88,18 +89,20 @@ func (rtm *RTM) connect(connectionCount int, useRTMStart bool) (*Info, *websocke // used to provide exponential backoff wait time with jitter before trying // to connect to slack again boff := &backoff{ - Min: 100 * time.Millisecond, - Max: 5 * time.Minute, - Factor: 2, - Jitter: true, + Max: 5 * time.Minute, } for { + var ( + backoff time.Duration + ) + // send connecting event rtm.IncomingEvents <- RTMEvent{"connecting", &ConnectingEvent{ Attempt: boff.attempts + 1, ConnectionCount: connectionCount, }} + // attempt to start the connection info, conn, err := rtm.startRTMAndDial(useRTMStart) if err == nil { @@ -109,32 +112,49 @@ func (rtm *RTM) connect(connectionCount int, useRTMStart bool) (*Info, *websocke // check for fatal errors switch err.Error() { case errInvalidAuth, errInactiveAccount, errMissingAuthToken: - rtm.Debugf("Invalid auth when connecting with RTM: %s", err) + rtm.Debugf("invalid auth when connecting with RTM: %s", err) rtm.IncomingEvents <- RTMEvent{"invalid_auth", &InvalidAuthEvent{}} return nil, nil, err default: } + switch actual := err.(type) { + case statusCodeError: + if actual.Code == http.StatusNotFound { + rtm.Debugf("invalid auth when connecting with RTM: %s", err) + rtm.IncomingEvents <- RTMEvent{"invalid_auth", &InvalidAuthEvent{}} + return nil, nil, err + } + case *RateLimitedError: + backoff = actual.RetryAfter + default: + } + + backoff = timex.Max(backoff, boff.Duration()) // any other errors are treated as recoverable and we try again after // sending the event along the IncomingEvents channel rtm.IncomingEvents <- RTMEvent{"connection_error", &ConnectionErrorEvent{ Attempt: boff.attempts, + Backoff: backoff, ErrorObj: err, }} - // check if Disconnect() has been invoked. + // get time we should wait before attempting to connect again + rtm.Debugf("reconnection %d failed: %s reconnecting in %v\n", boff.attempts, err, backoff) + + // wait for one of the following to occur, + // backoff duration has elapsed, killChannel is signalled, or + // the rtm finishes disconnecting. select { + case <-time.After(backoff): // retry after the backoff. + case intentional := <-rtm.killChannel: + if intentional { + rtm.killConnection(intentional, ErrRTMDisconnected) + return nil, nil, ErrRTMDisconnected + } case <-rtm.disconnected: - rtm.IncomingEvents <- RTMEvent{"disconnected", &DisconnectedEvent{Intentional: true}} - return nil, nil, fmt.Errorf("disconnect received while trying to connect") - default: + return nil, nil, ErrRTMDisconnected } - - // get time we should wait before attempting to connect again - dur := boff.Duration() - rtm.Debugf("reconnection %d failed: %s", boff.attempts+1, err) - rtm.Debugln(" -> reconnecting in", dur) - time.Sleep(dur) } } @@ -187,15 +207,19 @@ func (rtm *RTM) startRTMAndDial(useRTMStart bool) (info *Info, _ *websocket.Conn // // This should not be called directly! Instead a boolean value (true for // intentional, false otherwise) should be sent to the killChannel on the RTM. -func (rtm *RTM) killConnection(keepRunning chan bool, intentional bool) error { +func (rtm *RTM) killConnection(intentional bool, cause error) (err error) { rtm.Debugln("killing connection") - if rtm.isConnected { - close(keepRunning) + + if rtm.conn != nil { + err = rtm.conn.Close() + } + + rtm.IncomingEvents <- RTMEvent{"disconnected", &DisconnectedEvent{Intentional: intentional, Cause: cause}} + + if intentional { + rtm.disconnect() } - rtm.isConnected = false - rtm.wasIntentional = intentional - err := rtm.conn.Close() - rtm.IncomingEvents <- RTMEvent{"disconnected", &DisconnectedEvent{intentional}} + return err } @@ -204,31 +228,28 @@ func (rtm *RTM) killConnection(keepRunning chan bool, intentional bool) error { // interval. This also sends outgoing messages that are received from the RTM's // outgoingMessages channel. This also handles incoming raw events from the RTM // rawEvents channel. -func (rtm *RTM) handleEvents(keepRunning chan bool) { +func (rtm *RTM) handleEvents() { ticker := time.NewTicker(rtm.pingInterval) defer ticker.Stop() for { select { // catch "stop" signal on channel close case intentional := <-rtm.killChannel: - _ = rtm.killConnection(keepRunning, intentional) + _ = rtm.killConnection(intentional, errorsx.String("signaled")) return - // detect when the connection is dead. case <-rtm.pingDeadman.C: - rtm.Debugln("deadman switch trigger disconnecting") - _ = rtm.killConnection(keepRunning, false) + _ = rtm.killConnection(false, errorsx.String("deadman switch triggered")) + return // send pings on ticker interval case <-ticker.C: - err := rtm.ping() - if err != nil { - _ = rtm.killConnection(keepRunning, false) + if err := rtm.ping(); err != nil { + _ = rtm.killConnection(false, err) return } case <-rtm.forcePing: - err := rtm.ping() - if err != nil { - _ = rtm.killConnection(keepRunning, false) + if err := rtm.ping(); err != nil { + _ = rtm.killConnection(false, err) return } // listen for messages that need to be sent @@ -238,7 +259,8 @@ func (rtm *RTM) handleEvents(keepRunning chan bool) { case rawEvent := <-rtm.rawEvents: switch rtm.handleRawEvent(rawEvent) { case rtmEventTypeGoodbye: - _ = rtm.killConnection(keepRunning, false) + _ = rtm.killConnection(false, errorsx.String("goodbye detected")) + return default: } } @@ -250,17 +272,10 @@ func (rtm *RTM) handleEvents(keepRunning chan bool) { // // This will stop executing once the RTM's keepRunning channel has been closed // or has anything sent to it. -func (rtm *RTM) handleIncomingEvents(keepRunning <-chan bool) { +func (rtm *RTM) handleIncomingEvents() { for { - // non-blocking listen to see if channel is closed - select { - // catch "stop" signal on channel close - case <-keepRunning: + if err := rtm.receiveIncomingEvent(); err != nil { return - default: - if err := rtm.receiveIncomingEvent(); err != nil { - return - } } } } @@ -296,7 +311,6 @@ func (rtm *RTM) sendOutgoingMessage(msg OutgoingMessage) { Message: msg, ErrorObj: err, }} - // TODO force ping? } } @@ -332,20 +346,32 @@ func (rtm *RTM) receiveIncomingEvent() error { // 'PING' message // trigger a 'PING' to detect potential websocket disconnect - rtm.forcePing <- true + select { + case rtm.forcePing <- true: + case <-rtm.disconnected: + } case err != nil: // All other errors from ReadJSON come from NextReader, and should // kill the read loop and force a reconnect. rtm.IncomingEvents <- RTMEvent{"incoming_error", &IncomingEventError{ ErrorObj: err, }} - rtm.killChannel <- false + + select { + case rtm.killChannel <- false: + case <-rtm.disconnected: + } + return err case len(event) == 0: rtm.Debugln("Received empty event") default: - rtm.Debugln("Incoming Event:", string(event[:])) - rtm.rawEvents <- event + rtm.Debugln("Incoming Event:", string(event)) + select { + case rtm.rawEvents <- event: + case <-rtm.disconnected: + rtm.Debugln("disonnected while attempting to send raw event") + } } return nil } |