Repository navigation
Expand file tree
/
Copy pathmessage.go
More file actions
222 lines (210 loc) · 5.99 KB
/
Copy pathmessage.go
File metadata and controls
222 lines (210 loc) · 5.99 KB
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
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
package wrapper
import (
"context"
"encoding/json"
"errors"
"fmt"
"log/slog"
"strconv"
)
// weakBool is a boolean value that can be unmarshalled from a JSON boolean
// or integer (where 0 = false and 1 = true).
type weakBool bool
func (bit *weakBool) UnmarshalJSON(data []byte) error {
asString := string(data)
if asString == "1" || asString == "true" {
*bit = true
} else if asString == "0" || asString == "false" {
*bit = false
} else {
return fmt.Errorf("invalid value for boolean: %s", asString)
}
return nil
}
// Message is a ws-wrapper JSON-encoded message.
// See https://github.com/bminer/ws-wrapper/blob/master/README.md#protocol
type Message struct {
Channel string `json:"c,omitempty"`
AnonymousChannel int `json:"h,omitempty"`
Arguments []json.RawMessage `json:"a,omitempty"` // Arguments[0] is the event name
RequestID *int `json:"i,omitempty"`
ResponseData any `json:"d,omitempty"`
ResponseError any `json:"e,omitempty"`
ResponseJSError weakBool `json:"_,omitempty"`
CancelReason any `json:"x,omitempty"` // Request cancellation signal
IgnoreIfFalse *weakBool `json:"ws-wrapper,omitempty"`
processed chan struct{}
}
// EventName returns the name of the event or empty string if the message is
// invalid
func (m Message) EventName() string {
// Parse first argument
if len(m.Arguments) < 1 {
return ""
}
var name string
err := json.Unmarshal(m.Arguments[0], &name)
if err != nil {
return ""
}
return name
}
// HandlerArguments returns the arguments for the event handler
func (m Message) HandlerArguments() []json.RawMessage {
if len(m.Arguments) < 2 {
return nil
}
return m.Arguments[1:] // exclude event name
}
// Response returns the response data and error for this message.
func (m Message) Response() (any, error) {
if m.RequestID == nil {
return nil, errors.New("message is not a response")
}
if m.ResponseError == nil {
// No error
return m.ResponseData, nil
}
// Handle JavaScript error
if m.ResponseJSError {
jsErr, ok := m.ResponseError.(map[string]any)
if !ok {
return nil, errors.New(
"response is a malformed JavaScript error: not an object",
)
}
errMsg, ok := jsErr["message"].(string)
if !ok {
return nil, errors.New(
"response is a malformed JavaScript error: " +
"message key is not a string",
)
}
return nil, errors.New(errMsg)
}
// Handle string error
errMsg, ok := m.ResponseError.(string)
if !ok {
return nil, errors.New("response error is not a string")
}
return nil, errors.New(errMsg)
}
// CancelCause returns the cancel/abort reason from this message as an error.
// Works for both inbound request cancellations ({i, x}) and anonymous channel
// aborts ({h, x}). Returns "message is not a cancellation" if CancelReason is
// nil; returns context.Canceled if the reason cannot be parsed.
func (m Message) CancelCause() error {
if m.CancelReason == nil {
return errors.New("message is not a cancellation")
}
// Handle JavaScript error
if m.ResponseJSError {
jsErr, ok := m.CancelReason.(map[string]any)
if !ok {
return errors.New(
"cancel reason is a malformed JavaScript error: not an object",
)
}
errMsg, ok := jsErr["message"].(string)
if !ok {
return errors.New(
"cancel reason is a malformed JavaScript error: " +
"message key is not a string",
)
}
return errors.New(errMsg)
}
// Handle string error
errMsg, ok := m.CancelReason.(string)
if !ok {
return context.Canceled
}
return errors.New(errMsg)
}
func (m Message) LogValue() slog.Value {
if m.IgnoreIfFalse != nil && !*m.IgnoreIfFalse {
return slog.GroupValue(slog.Bool("ignored", true))
}
var attrs []slog.Attr
eventName := m.EventName()
const MaxArgLength = 1024
if eventName != "" {
if m.AnonymousChannel != 0 {
attrs = []slog.Attr{
slog.Int("anonCh", m.AnonymousChannel),
slog.String("event", eventName),
}
} else {
attrs = []slog.Attr{
slog.String("ch", m.Channel),
slog.String("event", eventName),
}
}
for i, arg := range m.HandlerArguments() {
if len(arg) > MaxArgLength {
attrs = append(attrs,
slog.String(
"args["+strconv.Itoa(i)+"]",
string(arg[:MaxArgLength-14])+"...(truncated)",
),
)
} else {
attrs = append(attrs,
slog.String("args["+strconv.Itoa(i)+"]", string(arg)),
)
}
}
if m.RequestID != nil {
attrs = append(attrs, slog.Int("reqID", *m.RequestID))
}
} else if m.AnonymousChannel != 0 && m.CancelReason != nil {
// Anonymous channel abort: {h, x}
attrs = []slog.Attr{
slog.Int("anonCh", m.AnonymousChannel),
slog.Any("cancelReason", m.CancelReason),
slog.Bool("jsError", bool(m.ResponseJSError)),
}
} else if m.RequestID != nil {
if m.AnonymousChannel != 0 {
// Anonymous channel creation: {i, h} (channel ID == request ID)
attrs = []slog.Attr{
slog.Int("reqID", *m.RequestID),
slog.Bool("anonChCreated", true),
}
} else if m.CancelReason != nil {
// Request cancellation: {i, x}
attrs = []slog.Attr{
slog.Int("reqID", *m.RequestID),
slog.Any("cancelReason", m.CancelReason),
slog.Bool("jsError", bool(m.ResponseJSError)),
}
} else if m.ResponseError != nil {
// Response rejection: {i, e}
attrs = []slog.Attr{
slog.Int("reqID", *m.RequestID),
slog.Any("error", m.ResponseError),
slog.Bool("jsError", bool(m.ResponseJSError)),
}
} else {
// Response resolution: {i, d}
attrs = []slog.Attr{
slog.Int("reqID", *m.RequestID),
slog.Any("data", m.ResponseData),
}
}
} else {
attrs = []slog.Attr{slog.Bool("invalid", true)}
}
return slog.GroupValue(attrs...)
}
// Processed returns a channel that is closed when the message is done being
// processed. This can be used by a "message" handler to measure message
// processing time.
func (m Message) Processed() <-chan struct{} {
return m.processed
}
// messageResponse is a response to a message
type messageResponse struct {
Data any
Error error
}