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
|
package mastodon
import (
"bufio"
"context"
"encoding/json"
"io"
"net/http"
"net/url"
"path"
"strings"
)
// UpdateEvent is struct for passing status event to app.
type UpdateEvent struct {
Status *Status `json:"status"`
}
func (e *UpdateEvent) event() {}
// NotificationEvent is struct for passing notification event to app.
type NotificationEvent struct {
Notification *Notification `json:"notification"`
}
func (e *NotificationEvent) event() {}
// DeleteEvent is struct for passing deletion event to app.
type DeleteEvent struct{ ID string }
func (e *DeleteEvent) event() {}
// ErrorEvent is struct for passing errors to app.
type ErrorEvent struct{ err error }
func (e *ErrorEvent) event() {}
func (e *ErrorEvent) Error() string { return e.err.Error() }
// Event is interface passing events to app.
type Event interface {
event()
}
func handleReader(q chan Event, r io.Reader) error {
var name string
s := bufio.NewScanner(r)
for s.Scan() {
line := s.Text()
token := strings.SplitN(line, ":", 2)
if len(token) != 2 {
continue
}
switch strings.TrimSpace(token[0]) {
case "event":
name = strings.TrimSpace(token[1])
case "data":
var err error
switch name {
case "update":
var status Status
err = json.Unmarshal([]byte(token[1]), &status)
if err == nil {
q <- &UpdateEvent{&status}
}
case "notification":
var notification Notification
err = json.Unmarshal([]byte(token[1]), ¬ification)
if err == nil {
q <- &NotificationEvent{¬ification}
}
case "delete":
q <- &DeleteEvent{ID: string(strings.TrimSpace(token[1]))}
}
if err != nil {
q <- &ErrorEvent{err}
}
}
}
return s.Err()
}
func (c *Client) streaming(ctx context.Context, p string, params url.Values) (chan Event, error) {
u, err := url.Parse(c.config.Server)
if err != nil {
return nil, err
}
u.Path = path.Join(u.Path, "/api/v1/streaming", p)
u.RawQuery = params.Encode()
req, err := http.NewRequest(http.MethodGet, u.String(), nil)
if err != nil {
return nil, err
}
req = req.WithContext(ctx)
req.Header.Set("Authorization", "Bearer "+c.config.AccessToken)
q := make(chan Event)
go func() {
defer close(q)
for {
select {
case <-ctx.Done():
return
default:
}
c.doStreaming(req, q)
}
}()
return q, nil
}
func (c *Client) doStreaming(req *http.Request, q chan Event) {
resp, err := c.Do(req)
if err != nil {
q <- &ErrorEvent{err}
return
}
defer resp.Body.Close()
if resp.StatusCode != http.StatusOK {
q <- &ErrorEvent{parseAPIError("bad request", resp)}
return
}
err = handleReader(q, resp.Body)
if err != nil {
q <- &ErrorEvent{err}
}
}
// StreamingUser return channel to read events on home.
func (c *Client) StreamingUser(ctx context.Context) (chan Event, error) {
return c.streaming(ctx, "user", nil)
}
// StreamingPublic return channel to read events on public.
func (c *Client) StreamingPublic(ctx context.Context, isLocal bool) (chan Event, error) {
p := "public"
if isLocal {
p = path.Join(p, "local")
}
return c.streaming(ctx, p, nil)
}
// StreamingHashtag return channel to read events on tagged timeline.
func (c *Client) StreamingHashtag(ctx context.Context, tag string, isLocal bool) (chan Event, error) {
params := url.Values{}
params.Set("tag", tag)
p := "hashtag"
if isLocal {
p = path.Join(p, "local")
}
return c.streaming(ctx, p, params)
}
// StreamingList return channel to read events on a list.
func (c *Client) StreamingList(ctx context.Context, id string) (chan Event, error) {
params := url.Values{}
params.Set("list", string(id))
return c.streaming(ctx, "list", params)
}
|
