arikawa/gateway/gateway.go

382 lines
9.4 KiB
Go
Raw Normal View History

2020-01-15 04:56:50 +00:00
// Package gateway handles the Discord gateway (or Websocket) connection, its
// events, and everything related to it. This includes logging into the
// Websocket.
//
// This package does not abstract events and function handlers; instead, it
// leaves that to the session package. This package exposes only a single Events
// channel.
2020-01-15 04:43:34 +00:00
package gateway
import (
"context"
"log"
2020-01-15 04:43:34 +00:00
"net/url"
"sync"
2020-01-15 04:43:34 +00:00
"time"
"github.com/diamondburned/arikawa/api"
2020-01-15 18:32:54 +00:00
"github.com/diamondburned/arikawa/internal/httputil"
"github.com/diamondburned/arikawa/internal/json"
"github.com/diamondburned/arikawa/internal/wsutil"
2020-01-15 04:43:34 +00:00
"github.com/pkg/errors"
)
const (
EndpointGateway = api.Endpoint + "gateway"
EndpointGatewayBot = api.EndpointGateway + "/bot"
Version = "6"
Encoding = "json"
)
var (
// WSTimeout is the timeout for connecting and writing to the Websocket,
// before Gateway cancels and fails.
WSTimeout = wsutil.DefaultTimeout
// WSBuffer is the size of the Event channel. This has to be at least 1 to
// make space for the first Event: Ready or Resumed.
WSBuffer = 10
2020-01-15 07:34:18 +00:00
// WSError is the default error handler
2020-01-29 03:54:22 +00:00
WSError = func(err error) { log.Println("Gateway error:", err) }
// WSFatal is the default fatal handler, which is called when the Gateway
// can't recover.
WSFatal = func(err error) { log.Fatalln("Gateway failed:", err) }
2020-01-17 22:29:13 +00:00
// WSExtraReadTimeout is the duration to be added to Hello, as a read
// timeout for the websocket.
WSExtraReadTimeout = time.Second
// WSRetries controls the number of Reconnects before erroring out.
WSRetries = 3
WSDebug = func(v ...interface{}) {}
2020-01-15 04:43:34 +00:00
)
var (
ErrMissingForResume = errors.New(
"missing session ID or sequence for resuming")
ErrWSMaxTries = errors.New("max tries reached")
)
func GatewayURL() (string, error) {
var Gateway struct {
URL string `json:"url"`
}
return Gateway.URL, httputil.DefaultClient.RequestJSON(
&Gateway, "GET", EndpointGateway)
}
type Gateway struct {
WS *wsutil.Websocket
json.Driver
// Timeout for connecting and writing to the Websocket, uses default
// WSTimeout (global).
WSTimeout time.Duration
2020-01-15 04:56:50 +00:00
// All events sent over are pointers to Event structs (structs suffixed with
// "Event"). This shouldn't be accessed if the Gateway is created with a
// Session.
2020-01-15 04:56:50 +00:00
Events chan Event
2020-01-15 04:43:34 +00:00
2020-01-15 04:56:50 +00:00
SessionID string
2020-01-15 04:43:34 +00:00
2020-01-15 07:34:18 +00:00
Identifier *Identifier
Pacemaker *Pacemaker
Sequence *Sequence
2020-01-15 04:43:34 +00:00
ErrorLog func(err error) // default to log.Println
// FatalError is where Reconnect errors will go to. When an error is sent
// here, the Gateway is already dead. This channel is buffered once.
FatalError <-chan error
fatalError chan error
2020-01-15 04:43:34 +00:00
// Only use for debugging
// If this channel is non-nil, all incoming OP packets will also be sent
2020-01-15 04:56:50 +00:00
// here. This should be buffered, so to not block the main loop.
OP chan *OP
2020-01-15 04:43:34 +00:00
// Mutex to hold off calls when the WS is not available. Doesn't block if
// Start() is not called or Close() is called. Also doesn't block for
// Identify or Resume.
available sync.RWMutex
2020-01-15 04:43:34 +00:00
// Filled by methods, internal use
paceDeath chan error
waitGroup *sync.WaitGroup
2020-01-15 04:43:34 +00:00
}
// NewGateway starts a new Gateway with the default stdlib JSON driver. For more
// information, refer to NewGatewayWithDriver.
func NewGateway(token string) (*Gateway, error) {
return NewGatewayWithDriver(token, json.Default{})
}
// NewGatewayWithDriver connects to the Gateway and authenticates automatically.
func NewGatewayWithDriver(token string, driver json.Driver) (*Gateway, error) {
URL, err := GatewayURL()
if err != nil {
return nil, errors.Wrap(err, "Failed to get gateway endpoint")
}
g := &Gateway{
2020-01-15 07:34:18 +00:00
Driver: driver,
WSTimeout: WSTimeout,
Events: make(chan Event, WSBuffer),
Identifier: DefaultIdentifier(token),
Sequence: NewSequence(),
ErrorLog: WSError,
fatalError: make(chan error, 1),
2020-01-15 04:43:34 +00:00
}
g.FatalError = g.fatalError
2020-01-15 04:43:34 +00:00
// Parameters for the gateway
param := url.Values{}
param.Set("v", Version)
param.Set("encoding", Encoding)
// Append the form to the URL
URL += "?" + param.Encode()
// Create a new undialed Websocket.
2020-02-11 17:29:30 +00:00
ws, err := wsutil.NewCustom(wsutil.NewConn(driver), URL)
2020-01-15 04:43:34 +00:00
if err != nil {
return nil, errors.Wrap(err, "Failed to connect to Gateway "+URL)
}
g.WS = ws
// Try and dial it
2020-01-17 22:29:13 +00:00
return g, nil
2020-01-15 04:43:34 +00:00
}
// Close closes the underlying Websocket connection.
func (g *Gateway) Close() error {
WSDebug("Stopping pacemaker...")
// If the pacemaker is running:
// Stop the pacemaker and the event handler
2020-01-16 03:28:21 +00:00
g.Pacemaker.Stop()
WSDebug("Stopped pacemaker. Waiting for WaitGroup to be done.")
// This should work, since Pacemaker should signal its loop to stop, which
// would also exit our event loop. Both would be 2.
g.waitGroup.Wait()
// Mark g.waitGroup as empty:
g.waitGroup = nil
2020-01-16 03:28:21 +00:00
// Stop the Websocket
2020-01-15 04:43:34 +00:00
return g.WS.Close(nil)
}
// Reconnects and resumes.
func (g *Gateway) Reconnect() error {
WSDebug("Reconnecting...")
2020-01-29 03:54:22 +00:00
// If the event loop is not dead:
if g.paceDeath != nil {
WSDebug("Gateway is not closed, closing before reconnecting...")
2020-01-29 03:54:22 +00:00
g.Close()
WSDebug("Gateway is closed asynchronously. Goroutine may not be exited.")
2020-01-29 03:54:22 +00:00
}
for i := 0; i < WSRetries; i++ {
WSDebug("Trying to dial, attempt", i)
2020-01-17 22:29:13 +00:00
// Condition: err == ErrInvalidSession:
// If the connection is rate limited (documented behavior):
// https://discordapp.com/developers/docs/topics/gateway#rate-limiting
2020-01-19 06:34:52 +00:00
if err := g.Open(); err != nil && err != ErrInvalidSession {
g.ErrorLog(errors.Wrap(err, "Failed to open gateway"))
continue
2020-01-17 22:29:13 +00:00
}
WSDebug("Started after attempt:", i)
return nil
}
return ErrWSMaxTries
}
2020-01-17 22:29:13 +00:00
func (g *Gateway) Open() error {
ctx := context.Background()
// Reconnect to the Gateway
if err := g.WS.Dial(ctx); err != nil {
return errors.Wrap(err, "Failed to reconnect")
}
2020-01-17 22:29:13 +00:00
WSDebug("Trying to start...")
2020-01-17 22:29:13 +00:00
// Try to resume the connection
if err := g.Start(); err != nil {
return err
2020-01-17 22:29:13 +00:00
}
// Started successfully, return
return nil
2020-01-15 04:43:34 +00:00
}
// Start authenticates with the websocket, or resume from a dead Websocket
// connection. This function doesn't block.
2020-01-15 04:43:34 +00:00
func (g *Gateway) Start() error {
g.available.Lock()
defer g.available.Unlock()
if err := g.start(); err != nil {
WSDebug("Start failed:", err)
if err := g.Close(); err != nil {
WSDebug("Failed to close after start fail:", err)
}
return err
}
return nil
}
// Wait blocks until the Gateway fatally exits when it couldn't reconnect
// anymore. To use this withh other channels, check out g.FatalError.
func (g *Gateway) Wait() error {
return <-g.FatalError
}
func (g *Gateway) start() error {
2020-01-15 04:43:34 +00:00
// This is where we'll get our events
ch := g.WS.Listen()
// Wait for an OP 10 Hello
var hello HelloEvent
2020-01-15 07:34:18 +00:00
if _, err := AssertEvent(g, <-ch, HelloOP, &hello); err != nil {
2020-01-15 04:43:34 +00:00
return errors.Wrap(err, "Error at Hello")
}
// Make a new WaitGroup for use in background loops:
g.waitGroup = new(sync.WaitGroup)
// Start the pacemaker with the heartrate received from Hello:
2020-01-16 03:28:21 +00:00
g.Pacemaker = &Pacemaker{
Heartrate: hello.HeartbeatInterval.Duration(),
Pace: g.Heartbeat,
OnDead: g.Reconnect,
}
// Pacemaker dies here, only when it's fatal.
g.paceDeath = g.Pacemaker.StartAsync(g.waitGroup)
2020-01-16 03:28:21 +00:00
2020-01-15 04:43:34 +00:00
// Send Discord either the Identify packet (if it's a fresh connection), or
// a Resume packet (if it's a dead connection).
if g.SessionID == "" {
// SessionID is empty, so this is a completely new session.
if err := g.Identify(); err != nil {
return errors.Wrap(err, "Failed to identify")
}
} else {
if err := g.Resume(); err != nil {
return errors.Wrap(err, "Failed to resume")
}
}
// Expect at least one event
ev := <-ch
// Check for error
if ev.Error != nil {
return errors.Wrap(ev.Error, "First error")
}
// Handle the event
if err := HandleEvent(g, ev.Data); err != nil {
return errors.Wrap(err, "WS handler error on first event")
}
2020-01-15 04:43:34 +00:00
// Start the event handler
g.waitGroup.Add(1)
2020-01-29 00:01:39 +00:00
go g.handleWS()
2020-01-15 04:43:34 +00:00
WSDebug("Started successfully.")
2020-01-15 04:43:34 +00:00
return nil
}
// handleWS uses the Websocket and parses them into g.Events.
2020-01-29 00:01:39 +00:00
func (g *Gateway) handleWS() {
err := g.eventLoop()
g.waitGroup.Done()
if err != nil {
g.ErrorLog(err)
g.fatalError <- errors.Wrap(g.Reconnect(), "Failed to reconnect")
// Reconnect should spawn another eventLoop in its Start function.
}
}
2020-01-15 04:43:34 +00:00
func (g *Gateway) eventLoop() error {
ch := g.WS.Listen()
2020-01-15 04:43:34 +00:00
for {
select {
case err := <-g.paceDeath:
// Got a paceDeath, we're exiting from here on out.
g.paceDeath = nil // mark
if err == nil {
WSDebug("Pacemaker stopped without errors.")
// No error, just exit normally.
return nil
}
2020-03-02 00:39:40 +00:00
return errors.Wrap(err, "Pacemaker died, reconnecting")
2020-01-15 04:43:34 +00:00
case ev := <-ch:
// Check for error
if ev.Error != nil {
return ev.Error
2020-01-15 04:43:34 +00:00
}
if len(ev.Data) == 0 {
return errors.New("Event data is empty, reconnecting.")
}
2020-01-15 04:43:34 +00:00
// Handle the event
if err := HandleEvent(g, ev.Data); err != nil {
2020-01-16 03:28:21 +00:00
g.ErrorLog(errors.Wrap(err, "WS handler error"))
2020-01-15 04:43:34 +00:00
}
}
}
}
func (g *Gateway) Send(code OPCode, v interface{}) error {
return g.send(true, code, v)
}
func (g *Gateway) send(lock bool, code OPCode, v interface{}) error {
2020-01-15 04:43:34 +00:00
var op = OP{
Code: code,
}
if v != nil {
2020-01-15 07:34:18 +00:00
b, err := g.Driver.Marshal(v)
2020-01-15 04:43:34 +00:00
if err != nil {
return errors.Wrap(err, "Failed to encode v")
}
op.Data = b
}
2020-01-15 07:34:18 +00:00
b, err := g.Driver.Marshal(op)
2020-01-15 04:43:34 +00:00
if err != nil {
return errors.Wrap(err, "Failed to encode payload")
}
ctx, cancel := context.WithTimeout(context.Background(), g.WSTimeout)
defer cancel()
if lock {
g.available.RLock()
defer g.available.RUnlock()
}
2020-01-15 04:43:34 +00:00
return g.WS.Send(ctx, b)
}