mirror of
https://github.com/diamondburned/arikawa.git
synced 2024-10-03 08:08:47 +00:00
ws: Add RawEvent for debugging
This commit is contained in:
parent
fd59b91de1
commit
aee547fa1f
|
@ -39,20 +39,28 @@ func New() *Handler {
|
||||||
// Call calls all handlers with the given event. This is an internal method; use
|
// Call calls all handlers with the given event. This is an internal method; use
|
||||||
// with care.
|
// with care.
|
||||||
func (h *Handler) Call(ev interface{}) {
|
func (h *Handler) Call(ev interface{}) {
|
||||||
v := reflect.ValueOf(ev)
|
t := reflect.TypeOf(ev)
|
||||||
t := v.Type()
|
|
||||||
|
|
||||||
h.mutex.RLock()
|
h.mutex.RLock()
|
||||||
defer h.mutex.RUnlock()
|
defer h.mutex.RUnlock()
|
||||||
|
|
||||||
for _, entry := range h.events[t].Entries {
|
typedHandlers := h.events[t].Entries
|
||||||
|
anyHandlers := h.events[nil].Entries
|
||||||
|
|
||||||
|
if len(typedHandlers) == 0 && len(anyHandlers) == 0 {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
v := reflect.ValueOf(ev)
|
||||||
|
|
||||||
|
for _, entry := range typedHandlers {
|
||||||
if entry.isInvalid() {
|
if entry.isInvalid() {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
entry.Call(v)
|
entry.Call(v)
|
||||||
}
|
}
|
||||||
|
|
||||||
for _, entry := range h.events[nil].Entries {
|
for _, entry := range anyHandlers {
|
||||||
if entry.isInvalid() || entry.not(t) {
|
if entry.isInvalid() || entry.not(t) {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
|
@ -1,6 +1,7 @@
|
||||||
package ws
|
package ws
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"context"
|
||||||
"io"
|
"io"
|
||||||
"net/http"
|
"net/http"
|
||||||
|
|
||||||
|
@ -50,15 +51,26 @@ func NewDecodeBuffer(cap int) DecodeBuffer {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// DecodeFrom reads the given reader and decodes it into an Op.
|
// DecodeInto reads the given reader and decodes it into the Op out channel.
|
||||||
//
|
//
|
||||||
// buf is optional.
|
// buf is optional.
|
||||||
func (c Codec) DecodeFrom(r io.Reader, buf *DecodeBuffer) Op {
|
func (c Codec) DecodeInto(ctx context.Context, r io.Reader, buf *DecodeBuffer, out chan<- Op) error {
|
||||||
var op codecOp
|
var op codecOp
|
||||||
op.Data = json.Raw(buf.buf)
|
op.Data = json.Raw(buf.buf)
|
||||||
|
|
||||||
if err := json.DecodeStream(r, &op); err != nil {
|
if err := json.DecodeStream(r, &op); err != nil {
|
||||||
return newErrOp(err, "cannot read JSON stream")
|
return c.send(ctx, out, newErrOp(err, "cannot read JSON stream"))
|
||||||
|
}
|
||||||
|
|
||||||
|
if EnableRawEvents {
|
||||||
|
dt := op.Data
|
||||||
|
op := op.Op
|
||||||
|
op.Data = &RawEvent{
|
||||||
|
Raw: dt,
|
||||||
|
OriginalCode: op.Code,
|
||||||
|
OriginalType: op.Type,
|
||||||
|
}
|
||||||
|
c.send(ctx, out, op)
|
||||||
}
|
}
|
||||||
|
|
||||||
// buf isn't grown from here out. Set it back right now. If Data hasn't been
|
// buf isn't grown from here out. Set it back right now. If Data hasn't been
|
||||||
|
@ -73,15 +85,24 @@ func (c Codec) DecodeFrom(r io.Reader, buf *DecodeBuffer) Op {
|
||||||
Op: op.Code,
|
Op: op.Code,
|
||||||
Type: op.Type,
|
Type: op.Type,
|
||||||
}
|
}
|
||||||
return newErrOp(err, "")
|
return c.send(ctx, out, newErrOp(err, ""))
|
||||||
}
|
}
|
||||||
|
|
||||||
op.Op.Data = fn()
|
op.Op.Data = fn()
|
||||||
if err := op.Data.UnmarshalTo(op.Op.Data); err != nil {
|
if err := op.Data.UnmarshalTo(op.Op.Data); err != nil {
|
||||||
return newErrOp(err, "cannot unmarshal JSON data from gateway")
|
return c.send(ctx, out, newErrOp(err, "cannot unmarshal JSON data from gateway"))
|
||||||
}
|
}
|
||||||
|
|
||||||
return op.Op
|
return c.send(ctx, out, op.Op)
|
||||||
|
}
|
||||||
|
|
||||||
|
func (c *Codec) send(ctx context.Context, ch chan<- Op, op Op) error {
|
||||||
|
select {
|
||||||
|
case ch <- op:
|
||||||
|
return nil
|
||||||
|
case <-ctx.Done():
|
||||||
|
return ctx.Err()
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func newErrOp(err error, wrap string) Op {
|
func newErrOp(err error, wrap string) Op {
|
||||||
|
|
|
@ -55,8 +55,9 @@ type Conn struct {
|
||||||
}
|
}
|
||||||
|
|
||||||
type connMutex struct {
|
type connMutex struct {
|
||||||
wrmut chan struct{}
|
|
||||||
*websocket.Conn
|
*websocket.Conn
|
||||||
|
wrmut chan struct{}
|
||||||
|
cancel context.CancelFunc
|
||||||
}
|
}
|
||||||
|
|
||||||
var _ Connection = (*Conn)(nil)
|
var _ Connection = (*Conn)(nil)
|
||||||
|
@ -101,12 +102,15 @@ func (c *Conn) Dial(ctx context.Context, addr string) (<-chan Op, error) {
|
||||||
return nil, errors.Wrap(err, "failed to dial WS")
|
return nil, errors.Wrap(err, "failed to dial WS")
|
||||||
}
|
}
|
||||||
|
|
||||||
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
|
|
||||||
events := make(chan Op, 1)
|
events := make(chan Op, 1)
|
||||||
go readLoop(conn, c.codec, events)
|
go readLoop(ctx, conn, c.codec, events)
|
||||||
|
|
||||||
c.conn = &connMutex{
|
c.conn = &connMutex{
|
||||||
wrmut: make(chan struct{}, 1),
|
wrmut: make(chan struct{}, 1),
|
||||||
Conn: conn,
|
Conn: conn,
|
||||||
|
cancel: cancel,
|
||||||
}
|
}
|
||||||
|
|
||||||
return events, err
|
return events, err
|
||||||
|
@ -168,6 +172,10 @@ func (c *connMutex) close(timeout time.Duration, gracefully bool) error {
|
||||||
}
|
}
|
||||||
|
|
||||||
c.Conn = nil
|
c.Conn = nil
|
||||||
|
|
||||||
|
c.cancel()
|
||||||
|
c.cancel = nil
|
||||||
|
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -212,7 +220,7 @@ type loopState struct {
|
||||||
buf DecodeBuffer
|
buf DecodeBuffer
|
||||||
}
|
}
|
||||||
|
|
||||||
func readLoop(conn *websocket.Conn, codec Codec, opCh chan<- Op) {
|
func readLoop(ctx context.Context, conn *websocket.Conn, codec Codec, opCh chan<- Op) {
|
||||||
// Clean up the events channel in the end.
|
// Clean up the events channel in the end.
|
||||||
defer close(opCh)
|
defer close(opCh)
|
||||||
|
|
||||||
|
@ -224,8 +232,7 @@ func readLoop(conn *websocket.Conn, codec Codec, opCh chan<- Op) {
|
||||||
}
|
}
|
||||||
|
|
||||||
for {
|
for {
|
||||||
b, err := state.handle()
|
if err := state.handle(ctx, opCh); err != nil {
|
||||||
if err != nil {
|
|
||||||
WSDebug("Conn: fatal Conn error:", err)
|
WSDebug("Conn: fatal Conn error:", err)
|
||||||
|
|
||||||
closeEv := &CloseEvent{
|
closeEv := &CloseEvent{
|
||||||
|
@ -247,16 +254,14 @@ func readLoop(conn *websocket.Conn, codec Codec, opCh chan<- Op) {
|
||||||
|
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
opCh <- b
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
func (state *loopState) handle() (Op, error) {
|
func (state *loopState) handle(ctx context.Context, opCh chan<- Op) error {
|
||||||
// skip message type
|
// skip message type
|
||||||
t, r, err := state.conn.NextReader()
|
t, r, err := state.conn.NextReader()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return Op{}, err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
if t == websocket.BinaryMessage {
|
if t == websocket.BinaryMessage {
|
||||||
|
@ -265,12 +270,12 @@ func (state *loopState) handle() (Op, error) {
|
||||||
if state.zlib == nil {
|
if state.zlib == nil {
|
||||||
z, err := zlib.NewReader(r)
|
z, err := zlib.NewReader(r)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return Op{}, errors.Wrap(err, "failed to create a zlib reader")
|
return errors.Wrap(err, "failed to create a zlib reader")
|
||||||
}
|
}
|
||||||
state.zlib = z
|
state.zlib = z
|
||||||
} else {
|
} else {
|
||||||
if err := state.zlib.(zlib.Resetter).Reset(r, nil); err != nil {
|
if err := state.zlib.(zlib.Resetter).Reset(r, nil); err != nil {
|
||||||
return Op{}, errors.Wrap(err, "failed to reset zlib reader")
|
return errors.Wrap(err, "failed to reset zlib reader")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -278,5 +283,9 @@ func (state *loopState) handle() (Op, error) {
|
||||||
r = state.zlib
|
r = state.zlib
|
||||||
}
|
}
|
||||||
|
|
||||||
return state.codec.DecodeFrom(r, &state.buf), nil
|
if err := state.codec.DecodeInto(ctx, r, &state.buf, opCh); err != nil {
|
||||||
|
return errors.Wrap(err, "error distributing event")
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil
|
||||||
}
|
}
|
||||||
|
|
|
@ -5,6 +5,7 @@ import (
|
||||||
"fmt"
|
"fmt"
|
||||||
"sync"
|
"sync"
|
||||||
|
|
||||||
|
"github.com/diamondburned/arikawa/v3/utils/json"
|
||||||
"github.com/pkg/errors"
|
"github.com/pkg/errors"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
@ -35,6 +36,23 @@ func (e *CloseEvent) Op() OpCode { return -1 }
|
||||||
// EventType implements Event. It returns an emty string.
|
// EventType implements Event. It returns an emty string.
|
||||||
func (e *CloseEvent) EventType() EventType { return "__ws.CloseEvent" }
|
func (e *CloseEvent) EventType() EventType { return "__ws.CloseEvent" }
|
||||||
|
|
||||||
|
// EnableRawEvents, if true, will cause ws to generate a RawEvent for each
|
||||||
|
// regular Event. It should only be used for debugging.
|
||||||
|
var EnableRawEvents = false
|
||||||
|
|
||||||
|
// RawEvent is used if EnableRawEvents is true.
|
||||||
|
type RawEvent struct {
|
||||||
|
json.Raw
|
||||||
|
OriginalCode OpCode `json:"-"`
|
||||||
|
OriginalType EventType `json:"-"`
|
||||||
|
}
|
||||||
|
|
||||||
|
// Op implements Event. It returns -1.
|
||||||
|
func (e *RawEvent) Op() OpCode { return -1 }
|
||||||
|
|
||||||
|
// EventType implements Event. It returns an emty string.
|
||||||
|
func (e *RawEvent) EventType() EventType { return "__ws.RawEvent" }
|
||||||
|
|
||||||
// EventType is a type for event types, which is the "t" field in the payload.
|
// EventType is a type for event types, which is the "t" field in the payload.
|
||||||
type EventType string
|
type EventType string
|
||||||
|
|
||||||
|
|
Loading…
Reference in a new issue