commit
28598eb3ab
|
@ -64,13 +64,13 @@ func cmdStream(c *cli.Context) error {
|
||||||
|
|
||||||
t := c.String("type")
|
t := c.String("type")
|
||||||
if t == "public" {
|
if t == "public" {
|
||||||
q, err = client.StreamingPublic(ctx)
|
q, err = client.StreamingPublic(ctx, false)
|
||||||
} else if t == "" || t == "public/local" {
|
} else if t == "" || t == "public/local" {
|
||||||
q, err = client.StreamingPublicLocal(ctx)
|
q, err = client.StreamingPublic(ctx, true)
|
||||||
} else if strings.HasPrefix(t, "user:") {
|
} else if strings.HasPrefix(t, "user:") {
|
||||||
q, err = client.StreamingUser(ctx, t[5:])
|
q, err = client.StreamingUser(ctx)
|
||||||
} else if strings.HasPrefix(t, "hashtag:") {
|
} else if strings.HasPrefix(t, "hashtag:") {
|
||||||
q, err = client.StreamingHashtag(ctx, t[8:])
|
q, err = client.StreamingHashtag(ctx, t[8:], false)
|
||||||
} else {
|
} else {
|
||||||
return errors.New("invalid type")
|
return errors.New("invalid type")
|
||||||
}
|
}
|
||||||
|
|
|
@ -173,9 +173,14 @@ func (c *Client) GetTimelinePublic(ctx context.Context, isLocal bool) ([]*Status
|
||||||
}
|
}
|
||||||
|
|
||||||
// GetTimelineHashtag return statuses from tagged timeline.
|
// GetTimelineHashtag return statuses from tagged timeline.
|
||||||
func (c *Client) GetTimelineHashtag(ctx context.Context, tag string) ([]*Status, error) {
|
func (c *Client) GetTimelineHashtag(ctx context.Context, tag string, isLocal bool) ([]*Status, error) {
|
||||||
|
params := url.Values{}
|
||||||
|
if isLocal {
|
||||||
|
params.Set("local", "t")
|
||||||
|
}
|
||||||
|
|
||||||
var statuses []*Status
|
var statuses []*Status
|
||||||
err := c.doAPI(ctx, http.MethodGet, fmt.Sprintf("/api/v1/timelines/tag/%s", (&url.URL{Path: tag}).EscapedPath()), nil, &statuses, nil)
|
err := c.doAPI(ctx, http.MethodGet, fmt.Sprintf("/api/v1/timelines/tag/%s", url.PathEscape(tag)), params, &statuses, nil)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
|
@ -374,11 +374,11 @@ func TestGetTimelineHashtag(t *testing.T) {
|
||||||
ClientSecret: "bar",
|
ClientSecret: "bar",
|
||||||
AccessToken: "zoo",
|
AccessToken: "zoo",
|
||||||
})
|
})
|
||||||
_, err := client.GetTimelineHashtag(context.Background(), "notfound")
|
_, err := client.GetTimelineHashtag(context.Background(), "notfound", false)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
t.Fatalf("should be fail: %v", err)
|
t.Fatalf("should be fail: %v", err)
|
||||||
}
|
}
|
||||||
tags, err := client.GetTimelineHashtag(context.Background(), "zzz")
|
tags, err := client.GetTimelineHashtag(context.Background(), "zzz", true)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("should not be fail: %v", err)
|
t.Fatalf("should not be fail: %v", err)
|
||||||
}
|
}
|
||||||
|
|
57
streaming.go
57
streaming.go
|
@ -86,7 +86,15 @@ func (c *Client) streaming(ctx context.Context, p string, params url.Values) (ch
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
u.Path = path.Join(u.Path, "/api/v1/streaming/"+p)
|
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)
|
||||||
|
|
||||||
var resp *http.Response
|
var resp *http.Response
|
||||||
|
|
||||||
|
@ -95,19 +103,10 @@ func (c *Client) streaming(ctx context.Context, p string, params url.Values) (ch
|
||||||
defer ctx.Done()
|
defer ctx.Done()
|
||||||
|
|
||||||
for {
|
for {
|
||||||
var in io.Reader
|
|
||||||
if params != nil {
|
|
||||||
in = strings.NewReader(params.Encode())
|
|
||||||
}
|
|
||||||
req, err := http.NewRequest(http.MethodGet, u.String(), in)
|
|
||||||
if err == nil {
|
|
||||||
req = req.WithContext(ctx)
|
|
||||||
req.Header.Set("Authorization", "Bearer "+c.config.AccessToken)
|
|
||||||
resp, err = c.Do(req)
|
resp, err = c.Do(req)
|
||||||
if resp != nil && resp.StatusCode != http.StatusOK {
|
if resp != nil && resp.StatusCode != http.StatusOK {
|
||||||
err = parseAPIError("bad request", resp)
|
err = parseAPIError("bad request", resp)
|
||||||
}
|
}
|
||||||
}
|
|
||||||
if err == nil {
|
if err == nil {
|
||||||
err = handleReader(ctx, q, resp.Body)
|
err = handleReader(ctx, q, resp.Body)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
|
@ -129,28 +128,30 @@ func (c *Client) streaming(ctx context.Context, p string, params url.Values) (ch
|
||||||
return q, nil
|
return q, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
// StreamingPublic return channel to read events on public.
|
|
||||||
func (c *Client) StreamingPublic(ctx context.Context) (chan Event, error) {
|
|
||||||
params := url.Values{}
|
|
||||||
return c.streaming(ctx, "public", params)
|
|
||||||
}
|
|
||||||
|
|
||||||
// StreamingPublicLocal return channel to read events on public.
|
|
||||||
func (c *Client) StreamingPublicLocal(ctx context.Context) (chan Event, error) {
|
|
||||||
params := url.Values{}
|
|
||||||
return c.streaming(ctx, "public/local", params)
|
|
||||||
}
|
|
||||||
|
|
||||||
// StreamingUser return channel to read events on home.
|
// StreamingUser return channel to read events on home.
|
||||||
func (c *Client) StreamingUser(ctx context.Context, user string) (chan Event, error) {
|
func (c *Client) StreamingUser(ctx context.Context) (chan Event, error) {
|
||||||
params := url.Values{}
|
return c.streaming(ctx, "user", nil)
|
||||||
params.Set("user", user)
|
}
|
||||||
return c.streaming(ctx, "user", params)
|
|
||||||
|
// 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.
|
// StreamingHashtag return channel to read events on tagged timeline.
|
||||||
func (c *Client) StreamingHashtag(ctx context.Context, tag string) (chan Event, error) {
|
func (c *Client) StreamingHashtag(ctx context.Context, tag string, isLocal bool) (chan Event, error) {
|
||||||
params := url.Values{}
|
params := url.Values{}
|
||||||
params.Set("tag", tag)
|
params.Set("tag", tag)
|
||||||
return c.streaming(ctx, "hashtag", params)
|
|
||||||
|
p := "hashtag"
|
||||||
|
if isLocal {
|
||||||
|
p = path.Join(p, "local")
|
||||||
|
}
|
||||||
|
|
||||||
|
return c.streaming(ctx, p, params)
|
||||||
}
|
}
|
||||||
|
|
|
@ -38,7 +38,7 @@ data: {"content": "bar"}
|
||||||
AccessToken: "zoo",
|
AccessToken: "zoo",
|
||||||
})
|
})
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
q, err := client.StreamingPublic(ctx)
|
q, err := client.StreamingPublic(ctx, false)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("should not be fail: %v", err)
|
t.Fatalf("should not be fail: %v", err)
|
||||||
}
|
}
|
||||||
|
|
|
@ -24,29 +24,29 @@ type Stream struct {
|
||||||
Payload interface{} `json:"payload"`
|
Payload interface{} `json:"payload"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// StreamingWSPublic return channel to read events on public using WebSocket.
|
|
||||||
func (c *WSClient) StreamingWSPublic(ctx context.Context) (chan Event, error) {
|
|
||||||
return c.streamingWS(ctx, "public", "")
|
|
||||||
}
|
|
||||||
|
|
||||||
// StreamingWSPublicLocal return channel to read events on public local using WebSocket.
|
|
||||||
func (c *WSClient) StreamingWSPublicLocal(ctx context.Context) (chan Event, error) {
|
|
||||||
return c.streamingWS(ctx, "public:local", "")
|
|
||||||
}
|
|
||||||
|
|
||||||
// StreamingWSUser return channel to read events on home using WebSocket.
|
// StreamingWSUser return channel to read events on home using WebSocket.
|
||||||
func (c *WSClient) StreamingWSUser(ctx context.Context) (chan Event, error) {
|
func (c *WSClient) StreamingWSUser(ctx context.Context) (chan Event, error) {
|
||||||
return c.streamingWS(ctx, "user", "")
|
return c.streamingWS(ctx, "user", "")
|
||||||
}
|
}
|
||||||
|
|
||||||
// StreamingWSHashtag return channel to read events on tagged timeline using WebSocket.
|
// StreamingWSPublic return channel to read events on public using WebSocket.
|
||||||
func (c *WSClient) StreamingWSHashtag(ctx context.Context, tag string) (chan Event, error) {
|
func (c *WSClient) StreamingWSPublic(ctx context.Context, isLocal bool) (chan Event, error) {
|
||||||
return c.streamingWS(ctx, "hashtag", tag)
|
s := "public"
|
||||||
|
if isLocal {
|
||||||
|
s += ":local"
|
||||||
|
}
|
||||||
|
|
||||||
|
return c.streamingWS(ctx, s, "")
|
||||||
}
|
}
|
||||||
|
|
||||||
// StreamingWSHashtagLocal return channel to read events on tagged local timeline using WebSocket.
|
// StreamingWSHashtag return channel to read events on tagged timeline using WebSocket.
|
||||||
func (c *WSClient) StreamingWSHashtagLocal(ctx context.Context, tag string) (chan Event, error) {
|
func (c *WSClient) StreamingWSHashtag(ctx context.Context, tag string, isLocal bool) (chan Event, error) {
|
||||||
return c.streamingWS(ctx, "hashtag:local", tag)
|
s := "hashtag"
|
||||||
|
if isLocal {
|
||||||
|
s += ":local"
|
||||||
|
}
|
||||||
|
|
||||||
|
return c.streamingWS(ctx, s, tag)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (c *WSClient) streamingWS(ctx context.Context, stream, tag string) (chan Event, error) {
|
func (c *WSClient) streamingWS(ctx context.Context, stream, tag string) (chan Event, error) {
|
||||||
|
|
|
@ -16,21 +16,7 @@ func TestStreamingWSPublic(t *testing.T) {
|
||||||
|
|
||||||
client := NewClient(&Config{Server: ts.URL}).NewWSClient()
|
client := NewClient(&Config{Server: ts.URL}).NewWSClient()
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
q, err := client.StreamingWSPublic(ctx)
|
q, err := client.StreamingWSPublic(ctx, false)
|
||||||
if err != nil {
|
|
||||||
t.Fatalf("should not be fail: %v", err)
|
|
||||||
}
|
|
||||||
|
|
||||||
wsTest(t, q, cancel)
|
|
||||||
}
|
|
||||||
|
|
||||||
func TestStreamingWSPublicLocal(t *testing.T) {
|
|
||||||
ts := httptest.NewServer(http.HandlerFunc(wsMock))
|
|
||||||
defer ts.Close()
|
|
||||||
|
|
||||||
client := NewClient(&Config{Server: ts.URL}).NewWSClient()
|
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
|
||||||
q, err := client.StreamingWSPublicLocal(ctx)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("should not be fail: %v", err)
|
t.Fatalf("should not be fail: %v", err)
|
||||||
}
|
}
|
||||||
|
@ -58,25 +44,17 @@ func TestStreamingWSHashtag(t *testing.T) {
|
||||||
|
|
||||||
client := NewClient(&Config{Server: ts.URL}).NewWSClient()
|
client := NewClient(&Config{Server: ts.URL}).NewWSClient()
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
q, err := client.StreamingWSHashtag(ctx, "zzz")
|
q, err := client.StreamingWSHashtag(ctx, "zzz", true)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("should not be fail: %v", err)
|
t.Fatalf("should not be fail: %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
wsTest(t, q, cancel)
|
wsTest(t, q, cancel)
|
||||||
}
|
|
||||||
|
|
||||||
func TestStreamingWSHashtagLocal(t *testing.T) {
|
ctx, cancel = context.WithCancel(context.Background())
|
||||||
ts := httptest.NewServer(http.HandlerFunc(wsMock))
|
q, err = client.StreamingWSHashtag(ctx, "zzz", false)
|
||||||
defer ts.Close()
|
|
||||||
|
|
||||||
client := NewClient(&Config{Server: ts.URL}).NewWSClient()
|
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
|
||||||
q, err := client.StreamingWSHashtagLocal(ctx, "zzz")
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("should not be fail: %v", err)
|
t.Fatalf("should not be fail: %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
wsTest(t, q, cancel)
|
wsTest(t, q, cancel)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@ -135,7 +113,7 @@ func wsTest(t *testing.T, q chan Event, cancel func()) {
|
||||||
events = append(events, e)
|
events = append(events, e)
|
||||||
}
|
}
|
||||||
if len(events) != 4 {
|
if len(events) != 4 {
|
||||||
t.Fatalf("result should be two: %d", len(events))
|
t.Fatalf("result should be four: %d", len(events))
|
||||||
}
|
}
|
||||||
if events[0].(*UpdateEvent).Status.Content != "foo" {
|
if events[0].(*UpdateEvent).Status.Content != "foo" {
|
||||||
t.Fatalf("want %q but %q", "foo", events[0].(*UpdateEvent).Status.Content)
|
t.Fatalf("want %q but %q", "foo", events[0].(*UpdateEvent).Status.Content)
|
||||||
|
@ -156,7 +134,7 @@ func TestStreamingWS(t *testing.T) {
|
||||||
defer ts.Close()
|
defer ts.Close()
|
||||||
|
|
||||||
client := NewClient(&Config{Server: ":"}).NewWSClient()
|
client := NewClient(&Config{Server: ":"}).NewWSClient()
|
||||||
_, err := client.StreamingWSPublicLocal(context.Background())
|
_, err := client.StreamingWSPublic(context.Background(), true)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
t.Fatalf("should be fail: %v", err)
|
t.Fatalf("should be fail: %v", err)
|
||||||
}
|
}
|
||||||
|
@ -164,7 +142,7 @@ func TestStreamingWS(t *testing.T) {
|
||||||
client = NewClient(&Config{Server: ts.URL}).NewWSClient()
|
client = NewClient(&Config{Server: ts.URL}).NewWSClient()
|
||||||
ctx, cancel := context.WithCancel(context.Background())
|
ctx, cancel := context.WithCancel(context.Background())
|
||||||
cancel()
|
cancel()
|
||||||
q, err := client.StreamingWSPublicLocal(ctx)
|
q, err := client.StreamingWSPublic(ctx, true)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Fatalf("should not be fail: %v", err)
|
t.Fatalf("should not be fail: %v", err)
|
||||||
}
|
}
|
||||||
|
|
Loading…
Reference in New Issue