Repository: sivchari/gotwtr
Pull request #63 (2021-11-13 13:43:47)

fix sampled stream implementation

mmmommm: # Todo

- [ ] Add abnormal test case

Patch:
diff --git a/client.go b/client.go
index e965e96..5fdf750 100644
--- a/client.go
+++ b/client.go
@@ -33,7 +33,7 @@ type Client interface {
 	// PostFollowing(ctx context.Context, id string, tuid string) (*PostFollowingResponse, error)
 	// PostRetweet(ctx context.Context, uid string, tid string) (*PostRetweetResponse, error)
 	RetweetsLookup(ctx context.Context, id string, opt ...*RetweetsLookupOpts) (*RetweetsLookupResponse, error)
-	SampledStream(ctx context.Context, opt ...*SampledStreamOpts) (*SampledStreamResponse, error)
+	SampledStream(ctx context.Context, ch chan<- SampledStreamResponse, errCh chan<- error, opt ...*SampledStreamOpts)
 	// SearchFullArchiveTweets(ctx context.Context, query string, opt ...*TweetSearchOption) (*TweetSearchResponse, error)
 	SearchRecentTweets(ctx context.Context, query string, opt ...*TweetSearchOption) (*TweetSearchResponse, error)
 	SearchSpaces(ctx context.Context, query string, opt ...*SearchSpacesOption) (*SearchSpacesResponse, error)
@@ -126,8 +126,8 @@ func (c *client) RetweetsLookup(ctx context.Context, id string, opt ...*Retweets
 	return retweetsLookup(ctx, c, id, opt...)
 }
 
-func (c *client) SampledStream(ctx context.Context, opt ...*SampledStreamOpts) (*SampledStreamResponse, error) {
-	return sampledStream(ctx, c, opt...)
+func (c *client) SampledStream(ctx context.Context, ch chan<- SampledStreamResponse, errCh chan<- error, opt ...*SampledStreamOpts) {
+	sampledStream(ctx, c, ch, errCh, opt...)
 }
 
 // func (c *client) SearchFullArchiveTweets(ctx context.Context, query string, opt ...*TweetSearchOption) (*TweetSearchResponse, error) {
diff --git a/example/tweet/stream/main.go b/example/tweet/stream/main.go
index a545983..1e26e68 100644
--- a/example/tweet/stream/main.go
+++ b/example/tweet/stream/main.go
@@ -10,11 +10,15 @@ import (
 func main() {
 	client := gotwtr.New("key")
 	// sampled stream
-	tsr, err := client.SampledStream(context.Background())
-	if err != nil {
-		panic(err)
-	}
-	for _, t := range tsr.Tweets {
-		fmt.Println(t)
+	ch := make(chan gotwtr.SampledStreamResponse)
+	errCh := make(chan error)
+	client.SampledStream(context.Background(), ch, errCh)
+	select {
+	case data := <-ch:
+		fmt.Println(data.Tweet)
+	case err := <-errCh:
+		if err != nil {
+			panic(err)
+		}
 	}
 }
diff --git a/sampled_stream.go b/sampled_stream.go
index 040f8c8..f3ec00a 100644
--- a/sampled_stream.go
+++ b/sampled_stream.go
@@ -6,12 +6,52 @@ import (
 	"errors"
 	"fmt"
 	"net/http"
+	"sync"
 )
 
-func sampledStream(ctx context.Context, c *client, opt ...*SampledStreamOpts) (*SampledStreamResponse, error) {
+func stopped(done <-chan struct{}) bool {
+	select {
+	case <-done:
+		return true
+	default:
+		return false
+	}
+}
+
+func (s *StreamResponse) retry(req *http.Request) {
+	defer close(s.ch)
+	defer s.wg.Done()
+	for !stopped(s.done) {
+		resp, err := s.client.Do(req)
+		if err != nil {
+			s.errCh <- err
+			return
+		}
+		defer resp.Body.Close()
+		if resp.StatusCode != http.StatusOK {
+			s.errCh <- &HTTPError{
+				APIName: "sampled stream",
+				Status:  resp.Status,
+				URL:     req.URL.String(),
+			}
+		}
+		var res SampledStreamResponse
+		if err := json.NewDecoder(resp.Body).Decode(&res); err != nil {
+			s.errCh <- err
+		}
+		select {
+		case s.ch <- res:
+			continue
+		case <-s.done:
+			return
+		}
+	}
+}
+
+func sampledStream(ctx context.Context, c *client, ch chan<- SampledStreamResponse, errCh chan<- error, opt ...*SampledStreamOpts) {
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, sampleStream, nil)
 	if err != nil {
-		return nil, fmt.Errorf("sampled stream new request with ctx: %w", err)
+		errCh <- fmt.Errorf("sampled stream new request with ctx: %w", err)
 	}
 	req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", c.bearerToken))
 
@@ -22,27 +62,17 @@ func sampledStream(ctx context.Context, c *client, opt ...*SampledStreamOpts) (*
 	case 1:
 		sopt = *opt[0]
 	default:
-		return nil, errors.New("sampled stream: only one option is allowed")
+		errCh <- errors.New("sampled stream: only one option is allowed")
 	}
 	sopt.addQuery(req)
 
-	resp, err := c.client.Do(req)
-	if err != nil {
-		return nil, fmt.Errorf("sampled stream response: %w", err)
-	}
-	defer resp.Body.Close()
-
-	var sampledStream SampledStreamResponse
-	if err := json.NewDecoder(resp.Body).Decode(&sampledStream); err != nil {
-		return nil, fmt.Errorf("sampled stream decode: %w", err)
-	}
-	if resp.StatusCode != http.StatusOK {
-		return &sampledStream, &HTTPError{
-			APIName: "sampled stream",
-			Status:  resp.Status,
-			URL:     req.URL.String(),
-		}
+	s := &StreamResponse{
+		client: c.client,
+		errCh:  errCh,
+		ch:     ch,
+		done:   make(chan struct{}),
+		wg:     &sync.WaitGroup{},
 	}
-
-	return &sampledStream, nil
+	s.wg.Add(1)
+	go s.retry(req)
 }
diff --git a/tweet.go b/tweet.go
index a08dc50..217c73d 100644
--- a/tweet.go
+++ b/tweet.go
@@ -209,12 +209,6 @@ type TweetSearchMeta struct {
 	NextToken   string `json:"next_token,omitempty"`
 }
 
-type SampledStreamResponse struct {
-	Tweets   []*Tweet            `json:"data"`
-	Includes *TweetIncludes      `json:"includes,omitempty"`
-	Errors   []*APIResponseError `json:"errors,omitempty"`
-}
-
 type TimeseriesCount struct {
 	Start      string `json:"start"`
 	End        string `json:"end"`
diff --git a/tweet_stream.go b/tweet_stream.go
new file mode 100644
index 0000000..7ab14f3
--- /dev/null
+++ b/tweet_stream.go
@@ -0,0 +1,20 @@
+package gotwtr
+
+import (
+	"net/http"
+	"sync"
+)
+
+type SampledStreamResponse struct {
+	Tweet    *Tweet              `json:"data"`
+	Includes *TweetIncludes      `json:"includes,omitempty"`
+	Errors   []*APIResponseError `json:"errors,omitempty"`
+}
+
+type StreamResponse struct {
+	client *http.Client
+	errCh  chan<- error
+	ch     chan<- SampledStreamResponse
+	done   chan struct{}
+	wg     *sync.WaitGroup
+}


Test patch:
diff --git a/sampled_stream_test.go b/sampled_stream_test.go
index d1e468a..c2a18b2 100644
--- a/sampled_stream_test.go
+++ b/sampled_stream_test.go
@@ -29,35 +29,9 @@ func Test_sampledStream(t *testing.T) {
 				ctx: context.Background(),
 				client: mockHTTPClient(func(request *http.Request) *http.Response {
 					body := `{
-						"data": [
-							{
-								"author_id": "2244994945",
-								"created_at": "2020-02-14T19:00:55.000Z",
-								"id": "1228393702244134912",
-								"text": "What did the developer write in their Valentine’s card?\n  \nwhile(true) {\n    I = Love(You);  \n}"
-							},
-							{
-								"author_id": "2244994945",
-								"created_at": "2020-02-12T17:09:56.000Z",
-								"id": "1227640996038684673",
-								"text": "Doctors: Googling stuff online does not make you a doctor\n\nDevelopers: https://t.co/mrju5ypPkb"
-							},
-							{
-								"author_id": "2244994945",
-								"created_at": "2019-11-27T20:26:41.000Z",
-								"id": "1199786642791452673",
-								"text": "C#"
-							}
-						],
-						"includes": {
-							"users": [
-								{
-									"created_at": "2013-12-14T04:35:55.000Z",
-									"id": "2244994945",
-									"name": "Twitter Dev",
-									"username": "TwitterDev"
-								}
-							]
+						"data": {
+							"id": "1067094924124872705",
+							"text": "Just getting started with Twitter APIs? Find out what you need in order to build an app. Watch this video! https://t.co/Hg8nkfoizN"
 						}
 					}`
 					return &http.Response{
@@ -68,35 +42,9 @@ func Test_sampledStream(t *testing.T) {
 				opt: []*gotwtr.SampledStreamOpts{},
 			},
 			want: &gotwtr.SampledStreamResponse{
-				Tweets: []*gotwtr.Tweet{
-					{
-						AuthorID:  "2244994945",
-						CreatedAt: "2020-02-14T19:00:55.000Z",
-						ID:        "1228393702244134912",
-						Text:      "What did the developer write in their Valentine’s card?\n  \nwhile(true) {\n    I = Love(You);  \n}",
-					},
-					{
-						AuthorID:  "2244994945",
-						CreatedAt: "2020-02-12T17:09:56.000Z",
-						ID:        "1227640996038684673",
-						Text:      "Doctors: Googling stuff online does not make you a doctor\n\nDevelopers: https://t.co/mrju5ypPkb",
-					},
-					{
-						AuthorID:  "2244994945",
-						CreatedAt: "2019-11-27T20:26:41.000Z",
-						ID:        "1199786642791452673",
-						Text:      "C#",
-					},
-				},
-				Includes: &gotwtr.TweetIncludes{
-					Users: []*gotwtr.User{
-						{
-							CreatedAt: "2013-12-14T04:35:55.000Z",
-							ID:        "2244994945",
-							Name:      "Twitter Dev",
-							UserName:  "TwitterDev",
-						},
-					},
+				Tweet: &gotwtr.Tweet{
+					ID:   "1067094924124872705",
+					Text: "Just getting started with Twitter APIs? Find out what you need in order to build an app. Watch this video! https://t.co/Hg8nkfoizN",
 				},
 			},
 			wantErr: false,
@@ -106,15 +54,21 @@ func Test_sampledStream(t *testing.T) {
 		tt := tt
 		t.Run(tt.name, func(t *testing.T) {
 			t.Parallel()
+			ch := make(chan gotwtr.SampledStreamResponse)
+			errCh := make(chan error)
 			c := gotwtr.New("test-key", gotwtr.WithHTTPClient(tt.args.client))
-			got, err := c.SampledStream(tt.args.ctx, tt.args.opt...)
-			if (err != nil) != tt.wantErr {
-				t.Errorf("SampledStream() error = %v, wantErr %v", err, tt.wantErr)
-				return
-			}
-			if diff := cmp.Diff(got, tt.want); diff != "" {
-				t.Errorf("SampledStream() mismatch (-want +got):\n%s", diff)
-				return
+			c.SampledStream(tt.args.ctx, ch, errCh, tt.args.opt...)
+			select {
+			case got := <-ch:
+				if diff := cmp.Diff(&got, tt.want); diff != "" {
+					t.Errorf("SampledStream() mismatch (-want +got):\n%s", diff)
+					return
+				}
+			case err := <-errCh:
+				if (err != nil) != tt.wantErr {
+					t.Errorf("SampledStream() error = %v, wantErr %v", err, tt.wantErr)
+					return
+				}
 			}
 		})
 	}