From c7255e69e858573604b449f65c205a430fb05e1a Mon Sep 17 00:00:00 2001 From: 178inaba <178inaba@users.noreply.github.com> Date: Sun, 23 Apr 2017 15:42:57 +0900 Subject: [PATCH 1/9] Add WebSocket implementation of streaming API --- mastodon.go | 3 + streaming_ws.go | 171 +++++++++++++++++++++++++++++++++++++++++++ streaming_ws_test.go | 1 + 3 files changed, 175 insertions(+) create mode 100644 streaming_ws.go create mode 100644 streaming_ws_test.go diff --git a/mastodon.go b/mastodon.go index 16ccb08..594d16a 100644 --- a/mastodon.go +++ b/mastodon.go @@ -13,6 +13,8 @@ import ( "path/filepath" "strings" "time" + + "github.com/gorilla/websocket" ) // Config is a setting for access mastodon APIs. @@ -26,6 +28,7 @@ type Config struct { // Client is a API client for mastodon. type Client struct { http.Client + websocket.Dialer config *Config interval time.Duration } diff --git a/streaming_ws.go b/streaming_ws.go new file mode 100644 index 0000000..20dd0a1 --- /dev/null +++ b/streaming_ws.go @@ -0,0 +1,171 @@ +package mastodon + +import ( + "context" + "encoding/json" + "net/url" + "path" + + "github.com/gorilla/websocket" +) + +// Stream is a struct of data that flows in streaming. +type Stream struct { + Event string `json:"event"` + Payload interface{} `json:"payload"` +} + +// StreamingWSPublic return channel to read events on public using WebSocket. +func (c *Client) 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 *Client) StreamingWSPublicLocal(ctx context.Context) (chan Event, error) { + return c.streamingWS(ctx, "public:local", "") +} + +// StreamingWSUser return channel to read events on home using WebSocket. +func (c *Client) StreamingWSUser(ctx context.Context, user string) (chan Event, error) { + return c.streamingWS(ctx, "user", "") +} + +// StreamingWSHashtag return channel to read events on tagged timeline using WebSocket. +func (c *Client) StreamingWSHashtag(ctx context.Context, tag string) (chan Event, error) { + return c.streamingWS(ctx, "hashtag", tag) +} + +// StreamingWSHashtagLocal return channel to read events on tagged local timeline using WebSocket. +func (c *Client) StreamingWSHashtagLocal(ctx context.Context, tag string) (chan Event, error) { + return c.streamingWS(ctx, "hashtag:local", tag) +} + +func (c *Client) streamingWS(ctx context.Context, stream, tag string) (chan Event, error) { + params := url.Values{} + params.Set("access_token", c.config.AccessToken) + params.Set("stream", stream) + if tag != "" { + params.Set("tag", tag) + } + + u, err := changeWebSocketScheme(c.config.Server) + if err != nil { + return nil, err + } + u.Path = path.Join(u.Path, "/api/v1/streaming") + u.RawQuery = params.Encode() + + q := make(chan Event) + go func() { + for { + err := c.handleWS(ctx, u.String(), q) + if err != nil { + return + } + } + }() + + return q, nil +} + +func (c *Client) handleWS(ctx context.Context, rawurl string, q chan Event) error { + conn, err := c.dialRedirect(rawurl) + if err != nil { + q <- &ErrorEvent{err: err} + + // End. + return err + } + defer conn.Close() + + for { + select { + case <-ctx.Done(): + q <- &ErrorEvent{err: ctx.Err()} + + // End. + return ctx.Err() + default: + } + + var s Stream + err := conn.ReadJSON(&s) + if err != nil { + q <- &ErrorEvent{err: err} + + // Reconnect. + break + } + + err = nil + switch s.Event { + case "update": + var status Status + err = json.Unmarshal([]byte(s.Payload.(string)), &status) + if err == nil { + q <- &UpdateEvent{Status: &status} + } + case "notification": + var notification Notification + err = json.Unmarshal([]byte(s.Payload.(string)), ¬ification) + if err == nil { + q <- &NotificationEvent{Notification: ¬ification} + } + case "delete": + if err == nil { + q <- &DeleteEvent{ID: int64(s.Payload.(float64))} + } + } + if err != nil { + q <- &ErrorEvent{err} + } + } + + return nil +} + +func (c *Client) dialRedirect(rawurl string) (conn *websocket.Conn, err error) { + for { + conn, rawurl, err = c.dial(rawurl) + if err != nil { + return nil, err + } else if conn != nil { + return conn, nil + } + } +} + +func (c *Client) dial(rawurl string) (*websocket.Conn, string, error) { + conn, resp, err := c.Dial(rawurl, nil) + if err != nil && err != websocket.ErrBadHandshake { + return nil, "", err + } + defer resp.Body.Close() + + if loc := resp.Header.Get("Location"); loc != "" { + u, err := changeWebSocketScheme(loc) + if err != nil { + return nil, "", err + } + + return nil, u.String(), nil + } + + return conn, "", err +} + +func changeWebSocketScheme(rawurl string) (*url.URL, error) { + u, err := url.Parse(rawurl) + if err != nil { + return nil, err + } + + switch u.Scheme { + case "http": + u.Scheme = "ws" + case "https": + u.Scheme = "wss" + } + + return u, nil +} diff --git a/streaming_ws_test.go b/streaming_ws_test.go new file mode 100644 index 0000000..6cd653a --- /dev/null +++ b/streaming_ws_test.go @@ -0,0 +1 @@ +package mastodon From 1d2b6deeef67ef6559370b636831e7f2548b7de7 Mon Sep 17 00:00:00 2001 From: 178inaba <178inaba@users.noreply.github.com> Date: Sun, 23 Apr 2017 17:28:31 +0900 Subject: [PATCH 2/9] Add streaming API WebSocket test --- streaming_ws.go | 2 +- streaming_ws_test.go | 177 +++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 178 insertions(+), 1 deletion(-) diff --git a/streaming_ws.go b/streaming_ws.go index 20dd0a1..896d8e2 100644 --- a/streaming_ws.go +++ b/streaming_ws.go @@ -26,7 +26,7 @@ func (c *Client) StreamingWSPublicLocal(ctx context.Context) (chan Event, error) } // StreamingWSUser return channel to read events on home using WebSocket. -func (c *Client) StreamingWSUser(ctx context.Context, user string) (chan Event, error) { +func (c *Client) StreamingWSUser(ctx context.Context) (chan Event, error) { return c.streamingWS(ctx, "user", "") } diff --git a/streaming_ws_test.go b/streaming_ws_test.go index 6cd653a..1e0af77 100644 --- a/streaming_ws_test.go +++ b/streaming_ws_test.go @@ -1 +1,178 @@ package mastodon + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" + "time" + + "github.com/gorilla/websocket" +) + +func TestStreamingWSPublic(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(wsMock)) + defer ts.Close() + + client := NewClient(&Config{ + Server: ts.URL, + ClientID: "foo", + ClientSecret: "bar", + AccessToken: "zoo", + }) + ctx, cancel := context.WithCancel(context.Background()) + q, err := client.StreamingWSPublic(ctx) + 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, + ClientID: "foo", + ClientSecret: "bar", + AccessToken: "zoo", + }) + ctx, cancel := context.WithCancel(context.Background()) + q, err := client.StreamingWSPublicLocal(ctx) + if err != nil { + t.Fatalf("should not be fail: %v", err) + } + + wsTest(t, q, cancel) +} + +func TestStreamingWSUser(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(wsMock)) + defer ts.Close() + + client := NewClient(&Config{ + Server: ts.URL, + ClientID: "foo", + ClientSecret: "bar", + AccessToken: "zoo", + }) + ctx, cancel := context.WithCancel(context.Background()) + q, err := client.StreamingWSUser(ctx) + if err != nil { + t.Fatalf("should not be fail: %v", err) + } + + wsTest(t, q, cancel) +} +func TestStreamingWSHashtag(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(wsMock)) + defer ts.Close() + + client := NewClient(&Config{ + Server: ts.URL, + ClientID: "foo", + ClientSecret: "bar", + AccessToken: "zoo", + }) + ctx, cancel := context.WithCancel(context.Background()) + q, err := client.StreamingWSHashtag(ctx, "zzz") + if err != nil { + t.Fatalf("should not be fail: %v", err) + } + + wsTest(t, q, cancel) +} +func TestStreamingWSHashtagLocal(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(wsMock)) + defer ts.Close() + + client := NewClient(&Config{ + Server: ts.URL, + ClientID: "foo", + ClientSecret: "bar", + AccessToken: "zoo", + }) + ctx, cancel := context.WithCancel(context.Background()) + q, err := client.StreamingWSHashtagLocal(ctx, "zzz") + if err != nil { + t.Fatalf("should not be fail: %v", err) + } + + wsTest(t, q, cancel) +} + +func wsMock(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/api/v1/streaming" { + http.Error(w, http.StatusText(http.StatusNotFound), http.StatusNotFound) + return + } + + u := websocket.Upgrader{} + conn, err := u.Upgrade(w, r, nil) + if err != nil { + http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError) + return + } + defer conn.Close() + + err = conn.WriteMessage(websocket.TextMessage, + []byte(`{"event":"update","payload":"{\"content\":\"foo\"}"}`)) + if err != nil { + http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError) + return + } + + err = conn.WriteMessage(websocket.TextMessage, + []byte(`{"event":"update","payload":"{\"content\":\"bar\"}"}`)) + if err != nil { + http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError) + return + } + + time.Sleep(10 * time.Second) +} + +func wsTest(t *testing.T, q chan Event, cancel func()) { + time.AfterFunc(time.Second, func() { + cancel() + close(q) + }) + events := []Event{} + for e := range q { + events = append(events, e) + } + if len(events) != 2 { + t.Fatalf("result should be two: %d", len(events)) + } + if events[0].(*UpdateEvent).Status.Content != "foo" { + t.Fatalf("want %q but %q", "foo", events[0].(*UpdateEvent).Status.Content) + } + if events[1].(*UpdateEvent).Status.Content != "bar" { + t.Fatalf("want %q but %q", "bar", events[1].(*UpdateEvent).Status.Content) + } +} + +func TestChangeWebSocketScheme(t *testing.T) { + _, err := changeWebSocketScheme(":") + if err == nil { + t.Fatalf("should be fail: %v", err) + } + + u, err := changeWebSocketScheme("http://example.com/") + if err != nil { + t.Fatalf("should not be fail: %v", err) + } + if u.Scheme != "ws" { + t.Fatalf("want %q but %q", "ws", u.Scheme) + } + + u, err = changeWebSocketScheme("https://example.com/") + if err != nil { + t.Fatalf("should not be fail: %v", err) + } + if u.Scheme != "wss" { + t.Fatalf("want %q but %q", "wss", u.Scheme) + } +} From 2379baeb2440392836a57be3a429ed0234eabc44 Mon Sep 17 00:00:00 2001 From: 178inaba <178inaba@users.noreply.github.com> Date: Sun, 23 Apr 2017 18:42:13 +0900 Subject: [PATCH 3/9] Add TestDial --- streaming_ws_test.go | 35 +++++++++++++++++++++++++++++++++++ 1 file changed, 35 insertions(+) diff --git a/streaming_ws_test.go b/streaming_ws_test.go index 1e0af77..001e511 100644 --- a/streaming_ws_test.go +++ b/streaming_ws_test.go @@ -66,6 +66,7 @@ func TestStreamingWSUser(t *testing.T) { wsTest(t, q, cancel) } + func TestStreamingWSHashtag(t *testing.T) { ts := httptest.NewServer(http.HandlerFunc(wsMock)) defer ts.Close() @@ -84,6 +85,7 @@ func TestStreamingWSHashtag(t *testing.T) { wsTest(t, q, cancel) } + func TestStreamingWSHashtagLocal(t *testing.T) { ts := httptest.NewServer(http.HandlerFunc(wsMock)) defer ts.Close() @@ -154,6 +156,39 @@ func wsTest(t *testing.T, q chan Event, cancel func()) { } } +func TestDial(t *testing.T) { + canErr := true + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if canErr { + canErr = false + http.Redirect(w, r, ":", http.StatusMovedPermanently) + return + } + + http.Redirect(w, r, "http://www.example.com/", http.StatusMovedPermanently) + })) + defer ts.Close() + + client := NewClient(&Config{}) + _, _, err := client.dial(":") + if err == nil { + t.Fatalf("should be fail: %v", err) + } + + _, rawurl, err := client.dial("ws://" + ts.Listener.Addr().String()) + if err == nil { + t.Fatalf("should not be fail: %v", err) + } + + _, rawurl, err = client.dial("ws://" + ts.Listener.Addr().String()) + if err != nil { + t.Fatalf("should not be fail: %v", err) + } + if rawurl != "ws://www.example.com/" { + t.Fatalf("want %q but %q", "ws://www.example.com/", rawurl) + } +} + func TestChangeWebSocketScheme(t *testing.T) { _, err := changeWebSocketScheme(":") if err == nil { From f0857503d1b1f6166ab20371372b76d6a1a7f804 Mon Sep 17 00:00:00 2001 From: 178inaba <178inaba@users.noreply.github.com> Date: Sun, 23 Apr 2017 18:46:41 +0900 Subject: [PATCH 4/9] Add TestDialRedirect --- streaming_ws_test.go | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/streaming_ws_test.go b/streaming_ws_test.go index 001e511..86ef764 100644 --- a/streaming_ws_test.go +++ b/streaming_ws_test.go @@ -156,6 +156,14 @@ func wsTest(t *testing.T, q chan Event, cancel func()) { } } +func TestDialRedirect(t *testing.T) { + client := NewClient(&Config{}) + _, err := client.dialRedirect(":") + if err == nil { + t.Fatalf("should be fail: %v", err) + } +} + func TestDial(t *testing.T) { canErr := true ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { From df768f24da37b138ca18c95acaadc1fe2e429e50 Mon Sep 17 00:00:00 2001 From: 178inaba <178inaba@users.noreply.github.com> Date: Sun, 23 Apr 2017 19:47:03 +0900 Subject: [PATCH 5/9] Add TestHandleWS --- streaming_ws.go | 4 +- streaming_ws_test.go | 120 +++++++++++++++++++++++++++++++------------ 2 files changed, 87 insertions(+), 37 deletions(-) diff --git a/streaming_ws.go b/streaming_ws.go index 896d8e2..e607be1 100644 --- a/streaming_ws.go +++ b/streaming_ws.go @@ -112,9 +112,7 @@ func (c *Client) handleWS(ctx context.Context, rawurl string, q chan Event) erro q <- &NotificationEvent{Notification: ¬ification} } case "delete": - if err == nil { - q <- &DeleteEvent{ID: int64(s.Payload.(float64))} - } + q <- &DeleteEvent{ID: int64(s.Payload.(float64))} } if err != nil { q <- &ErrorEvent{err} diff --git a/streaming_ws_test.go b/streaming_ws_test.go index 86ef764..304e6bc 100644 --- a/streaming_ws_test.go +++ b/streaming_ws_test.go @@ -14,12 +14,7 @@ func TestStreamingWSPublic(t *testing.T) { ts := httptest.NewServer(http.HandlerFunc(wsMock)) defer ts.Close() - client := NewClient(&Config{ - Server: ts.URL, - ClientID: "foo", - ClientSecret: "bar", - AccessToken: "zoo", - }) + client := NewClient(&Config{Server: ts.URL}) ctx, cancel := context.WithCancel(context.Background()) q, err := client.StreamingWSPublic(ctx) if err != nil { @@ -33,12 +28,7 @@ func TestStreamingWSPublicLocal(t *testing.T) { ts := httptest.NewServer(http.HandlerFunc(wsMock)) defer ts.Close() - client := NewClient(&Config{ - Server: ts.URL, - ClientID: "foo", - ClientSecret: "bar", - AccessToken: "zoo", - }) + client := NewClient(&Config{Server: ts.URL}) ctx, cancel := context.WithCancel(context.Background()) q, err := client.StreamingWSPublicLocal(ctx) if err != nil { @@ -52,12 +42,7 @@ func TestStreamingWSUser(t *testing.T) { ts := httptest.NewServer(http.HandlerFunc(wsMock)) defer ts.Close() - client := NewClient(&Config{ - Server: ts.URL, - ClientID: "foo", - ClientSecret: "bar", - AccessToken: "zoo", - }) + client := NewClient(&Config{Server: ts.URL}) ctx, cancel := context.WithCancel(context.Background()) q, err := client.StreamingWSUser(ctx) if err != nil { @@ -71,12 +56,7 @@ func TestStreamingWSHashtag(t *testing.T) { ts := httptest.NewServer(http.HandlerFunc(wsMock)) defer ts.Close() - client := NewClient(&Config{ - Server: ts.URL, - ClientID: "foo", - ClientSecret: "bar", - AccessToken: "zoo", - }) + client := NewClient(&Config{Server: ts.URL}) ctx, cancel := context.WithCancel(context.Background()) q, err := client.StreamingWSHashtag(ctx, "zzz") if err != nil { @@ -90,12 +70,7 @@ func TestStreamingWSHashtagLocal(t *testing.T) { ts := httptest.NewServer(http.HandlerFunc(wsMock)) defer ts.Close() - client := NewClient(&Config{ - Server: ts.URL, - ClientID: "foo", - ClientSecret: "bar", - AccessToken: "zoo", - }) + client := NewClient(&Config{Server: ts.URL}) ctx, cancel := context.WithCancel(context.Background()) q, err := client.StreamingWSHashtagLocal(ctx, "zzz") if err != nil { @@ -127,7 +102,21 @@ func wsMock(w http.ResponseWriter, r *http.Request) { } err = conn.WriteMessage(websocket.TextMessage, - []byte(`{"event":"update","payload":"{\"content\":\"bar\"}"}`)) + []byte(`{"event":"notification","payload":"{\"id\":123}"}`)) + if err != nil { + http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError) + return + } + + err = conn.WriteMessage(websocket.TextMessage, + []byte(`{"event":"delete","payload":1234567}`)) + if err != nil { + http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError) + return + } + + err = conn.WriteMessage(websocket.TextMessage, + []byte(`{"event":"update","payload":""}`)) if err != nil { http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError) return @@ -145,15 +134,78 @@ func wsTest(t *testing.T, q chan Event, cancel func()) { for e := range q { events = append(events, e) } - if len(events) != 2 { + if len(events) != 4 { t.Fatalf("result should be two: %d", len(events)) } if events[0].(*UpdateEvent).Status.Content != "foo" { t.Fatalf("want %q but %q", "foo", events[0].(*UpdateEvent).Status.Content) } - if events[1].(*UpdateEvent).Status.Content != "bar" { - t.Fatalf("want %q but %q", "bar", events[1].(*UpdateEvent).Status.Content) + if events[1].(*NotificationEvent).Notification.ID != 123 { + t.Fatalf("want %d but %d", 123, events[1].(*NotificationEvent).Notification.ID) } + if events[2].(*DeleteEvent).ID != 1234567 { + t.Fatalf("want %d but %d", 1234567, events[2].(*DeleteEvent).ID) + } + if errorEvent, ok := events[3].(*ErrorEvent); !ok { + t.Fatalf("should be fail: %v", errorEvent.err) + } +} + +func TestHandleWS(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + u := websocket.Upgrader{} + conn, err := u.Upgrade(w, r, nil) + if err != nil { + http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError) + return + } + defer conn.Close() + + err = conn.WriteMessage(websocket.TextMessage, + []byte(``)) + if err != nil { + http.Error(w, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError) + return + } + + time.Sleep(10 * time.Second) + })) + defer ts.Close() + + q := make(chan Event) + client := NewClient(&Config{}) + + go func() { + e := <-q + if errorEvent, ok := e.(*ErrorEvent); !ok { + t.Fatalf("should be fail: %v", errorEvent.err) + } + }() + err := client.handleWS(context.Background(), ":", q) + if err == nil { + t.Fatalf("should be fail: %v", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + cancel() + go func() { + e := <-q + if errorEvent, ok := e.(*ErrorEvent); !ok { + t.Fatalf("should be fail: %v", errorEvent.err) + } + }() + err = client.handleWS(ctx, "ws://"+ts.Listener.Addr().String(), q) + if err == nil { + t.Fatalf("should be fail: %v", err) + } + + go func() { + e := <-q + if errorEvent, ok := e.(*ErrorEvent); !ok { + t.Fatalf("should be fail: %v", errorEvent.err) + } + }() + client.handleWS(context.Background(), "ws://"+ts.Listener.Addr().String(), q) } func TestDialRedirect(t *testing.T) { From ee6a6dbcc97838b9dbba51cb57eec969b004a239 Mon Sep 17 00:00:00 2001 From: 178inaba <178inaba@users.noreply.github.com> Date: Sun, 23 Apr 2017 20:04:06 +0900 Subject: [PATCH 6/9] Add TestStreamingWS --- streaming_ws_test.go | 25 +++++++++++++++++++++++++ 1 file changed, 25 insertions(+) diff --git a/streaming_ws_test.go b/streaming_ws_test.go index 304e6bc..f041984 100644 --- a/streaming_ws_test.go +++ b/streaming_ws_test.go @@ -151,6 +151,31 @@ func wsTest(t *testing.T, q chan Event, cancel func()) { } } +func TestStreamingWS(t *testing.T) { + ts := httptest.NewServer(http.HandlerFunc(wsMock)) + defer ts.Close() + + client := NewClient(&Config{Server: ":"}) + _, err := client.StreamingWSPublicLocal(context.Background()) + if err == nil { + t.Fatalf("should be fail: %v", err) + } + + client = NewClient(&Config{Server: ts.URL}) + ctx, cancel := context.WithCancel(context.Background()) + cancel() + q, err := client.StreamingWSPublicLocal(ctx) + if err != nil { + t.Fatalf("should not be fail: %v", err) + } + go func() { + e := <-q + if errorEvent, ok := e.(*ErrorEvent); !ok { + t.Fatalf("should be fail: %v", errorEvent.err) + } + }() +} + func TestHandleWS(t *testing.T) { ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { u := websocket.Upgrader{} From 010ec2eaf92da466640e155dfac5a9611b6a0f33 Mon Sep 17 00:00:00 2001 From: 178inaba <178inaba@users.noreply.github.com> Date: Mon, 24 Apr 2017 13:55:07 +0900 Subject: [PATCH 7/9] Add WSClient --- streaming_ws.go | 31 ++++++++++++++++++++----------- streaming_ws_test.go | 20 ++++++++++---------- 2 files changed, 30 insertions(+), 21 deletions(-) diff --git a/streaming_ws.go b/streaming_ws.go index e607be1..915465f 100644 --- a/streaming_ws.go +++ b/streaming_ws.go @@ -9,6 +9,15 @@ import ( "github.com/gorilla/websocket" ) +// WSClient is a WebSocket client. +type WSClient struct { + websocket.Dialer + client *Client +} + +// NewWSClient return WebSocket client. +func (c *Client) NewWSClient() *WSClient { return &WSClient{client: c} } + // Stream is a struct of data that flows in streaming. type Stream struct { Event string `json:"event"` @@ -16,39 +25,39 @@ type Stream struct { } // StreamingWSPublic return channel to read events on public using WebSocket. -func (c *Client) StreamingWSPublic(ctx context.Context) (chan Event, error) { +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 *Client) StreamingWSPublicLocal(ctx context.Context) (chan Event, error) { +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. -func (c *Client) StreamingWSUser(ctx context.Context) (chan Event, error) { +func (c *WSClient) StreamingWSUser(ctx context.Context) (chan Event, error) { return c.streamingWS(ctx, "user", "") } // StreamingWSHashtag return channel to read events on tagged timeline using WebSocket. -func (c *Client) StreamingWSHashtag(ctx context.Context, tag string) (chan Event, error) { +func (c *WSClient) StreamingWSHashtag(ctx context.Context, tag string) (chan Event, error) { return c.streamingWS(ctx, "hashtag", tag) } // StreamingWSHashtagLocal return channel to read events on tagged local timeline using WebSocket. -func (c *Client) StreamingWSHashtagLocal(ctx context.Context, tag string) (chan Event, error) { +func (c *WSClient) StreamingWSHashtagLocal(ctx context.Context, tag string) (chan Event, error) { return c.streamingWS(ctx, "hashtag:local", tag) } -func (c *Client) streamingWS(ctx context.Context, stream, tag string) (chan Event, error) { +func (c *WSClient) streamingWS(ctx context.Context, stream, tag string) (chan Event, error) { params := url.Values{} - params.Set("access_token", c.config.AccessToken) + params.Set("access_token", c.client.config.AccessToken) params.Set("stream", stream) if tag != "" { params.Set("tag", tag) } - u, err := changeWebSocketScheme(c.config.Server) + u, err := changeWebSocketScheme(c.client.config.Server) if err != nil { return nil, err } @@ -68,7 +77,7 @@ func (c *Client) streamingWS(ctx context.Context, stream, tag string) (chan Even return q, nil } -func (c *Client) handleWS(ctx context.Context, rawurl string, q chan Event) error { +func (c *WSClient) handleWS(ctx context.Context, rawurl string, q chan Event) error { conn, err := c.dialRedirect(rawurl) if err != nil { q <- &ErrorEvent{err: err} @@ -122,7 +131,7 @@ func (c *Client) handleWS(ctx context.Context, rawurl string, q chan Event) erro return nil } -func (c *Client) dialRedirect(rawurl string) (conn *websocket.Conn, err error) { +func (c *WSClient) dialRedirect(rawurl string) (conn *websocket.Conn, err error) { for { conn, rawurl, err = c.dial(rawurl) if err != nil { @@ -133,7 +142,7 @@ func (c *Client) dialRedirect(rawurl string) (conn *websocket.Conn, err error) { } } -func (c *Client) dial(rawurl string) (*websocket.Conn, string, error) { +func (c *WSClient) dial(rawurl string) (*websocket.Conn, string, error) { conn, resp, err := c.Dial(rawurl, nil) if err != nil && err != websocket.ErrBadHandshake { return nil, "", err diff --git a/streaming_ws_test.go b/streaming_ws_test.go index f041984..9d5386b 100644 --- a/streaming_ws_test.go +++ b/streaming_ws_test.go @@ -14,7 +14,7 @@ func TestStreamingWSPublic(t *testing.T) { ts := httptest.NewServer(http.HandlerFunc(wsMock)) defer ts.Close() - client := NewClient(&Config{Server: ts.URL}) + client := NewClient(&Config{Server: ts.URL}).NewWSClient() ctx, cancel := context.WithCancel(context.Background()) q, err := client.StreamingWSPublic(ctx) if err != nil { @@ -28,7 +28,7 @@ func TestStreamingWSPublicLocal(t *testing.T) { ts := httptest.NewServer(http.HandlerFunc(wsMock)) defer ts.Close() - client := NewClient(&Config{Server: ts.URL}) + client := NewClient(&Config{Server: ts.URL}).NewWSClient() ctx, cancel := context.WithCancel(context.Background()) q, err := client.StreamingWSPublicLocal(ctx) if err != nil { @@ -42,7 +42,7 @@ func TestStreamingWSUser(t *testing.T) { ts := httptest.NewServer(http.HandlerFunc(wsMock)) defer ts.Close() - client := NewClient(&Config{Server: ts.URL}) + client := NewClient(&Config{Server: ts.URL}).NewWSClient() ctx, cancel := context.WithCancel(context.Background()) q, err := client.StreamingWSUser(ctx) if err != nil { @@ -56,7 +56,7 @@ func TestStreamingWSHashtag(t *testing.T) { ts := httptest.NewServer(http.HandlerFunc(wsMock)) defer ts.Close() - client := NewClient(&Config{Server: ts.URL}) + client := NewClient(&Config{Server: ts.URL}).NewWSClient() ctx, cancel := context.WithCancel(context.Background()) q, err := client.StreamingWSHashtag(ctx, "zzz") if err != nil { @@ -70,7 +70,7 @@ func TestStreamingWSHashtagLocal(t *testing.T) { ts := httptest.NewServer(http.HandlerFunc(wsMock)) defer ts.Close() - client := NewClient(&Config{Server: ts.URL}) + client := NewClient(&Config{Server: ts.URL}).NewWSClient() ctx, cancel := context.WithCancel(context.Background()) q, err := client.StreamingWSHashtagLocal(ctx, "zzz") if err != nil { @@ -155,13 +155,13 @@ func TestStreamingWS(t *testing.T) { ts := httptest.NewServer(http.HandlerFunc(wsMock)) defer ts.Close() - client := NewClient(&Config{Server: ":"}) + client := NewClient(&Config{Server: ":"}).NewWSClient() _, err := client.StreamingWSPublicLocal(context.Background()) if err == nil { t.Fatalf("should be fail: %v", err) } - client = NewClient(&Config{Server: ts.URL}) + client = NewClient(&Config{Server: ts.URL}).NewWSClient() ctx, cancel := context.WithCancel(context.Background()) cancel() q, err := client.StreamingWSPublicLocal(ctx) @@ -198,7 +198,7 @@ func TestHandleWS(t *testing.T) { defer ts.Close() q := make(chan Event) - client := NewClient(&Config{}) + client := NewClient(&Config{}).NewWSClient() go func() { e := <-q @@ -234,7 +234,7 @@ func TestHandleWS(t *testing.T) { } func TestDialRedirect(t *testing.T) { - client := NewClient(&Config{}) + client := NewClient(&Config{}).NewWSClient() _, err := client.dialRedirect(":") if err == nil { t.Fatalf("should be fail: %v", err) @@ -254,7 +254,7 @@ func TestDial(t *testing.T) { })) defer ts.Close() - client := NewClient(&Config{}) + client := NewClient(&Config{}).NewWSClient() _, _, err := client.dial(":") if err == nil { t.Fatalf("should be fail: %v", err) From 34cce1bb0dcfc0df36ec96933892a167e0266903 Mon Sep 17 00:00:00 2001 From: 178inaba <178inaba@users.noreply.github.com> Date: Mon, 24 Apr 2017 13:57:12 +0900 Subject: [PATCH 8/9] Remove websocket.Dialer from Client --- mastodon.go | 3 --- 1 file changed, 3 deletions(-) diff --git a/mastodon.go b/mastodon.go index 594d16a..16ccb08 100644 --- a/mastodon.go +++ b/mastodon.go @@ -13,8 +13,6 @@ import ( "path/filepath" "strings" "time" - - "github.com/gorilla/websocket" ) // Config is a setting for access mastodon APIs. @@ -28,7 +26,6 @@ type Config struct { // Client is a API client for mastodon. type Client struct { http.Client - websocket.Dialer config *Config interval time.Duration } From ed709d74a4b09ca5ba22d402ab22fc9b7b5bb4df Mon Sep 17 00:00:00 2001 From: 178inaba <178inaba@users.noreply.github.com> Date: Mon, 24 Apr 2017 17:25:56 +0900 Subject: [PATCH 9/9] Fix error message --- streaming_ws_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/streaming_ws_test.go b/streaming_ws_test.go index 9d5386b..5456a5d 100644 --- a/streaming_ws_test.go +++ b/streaming_ws_test.go @@ -262,7 +262,7 @@ func TestDial(t *testing.T) { _, rawurl, err := client.dial("ws://" + ts.Listener.Addr().String()) if err == nil { - t.Fatalf("should not be fail: %v", err) + t.Fatalf("should be fail: %v", err) } _, rawurl, err = client.dial("ws://" + ts.Listener.Addr().String())