Merge pull request #18 from Fallenstedt/0.3.1

0.3.1

Author: Fallenstedt

README.md

@@ -41,10 +41,33 @@ Create a twitterstream instance with your access token from above.
 	api := twitterstream.NewTwitterStream(tok.AccessToken)
 ```
 
+##### Set your unmarshal hook
+It is encouraged you set an unmarshal hook for thread-safety. Go's `bytes.Buffer` is not thread safe. Sharing a `bytes.Buffer`
+across multiple goroutines introduces risk of panics when decoding json [source](https://github.com/Fallenstedt/twitter-stream/issues/13).
+To avoid panics, it's encouraged to unmarshal json in the same goroutine where the `bytes.Buffer` exists. Use `SetUnmarshalHook` to set a function that unmarshals json.
+
+By default, twitterstream's unmarshal hook will return `[]byte` if you want to live dangerously.
+
+```go
+    api.Stream.SetUnmarshalHook(func(bytes []byte) (interface{}, error) {
+        // StreemData is a struct that represents your returned json
+    	// This is a quick resource to generate a struct from your json
+        // https://mholt.github.io/json-to-go/
+        data := StreamData{}
+        if err := json.Unmarshal(bytes, &data); err != nil {
+            log.Printf("Failed to unmarshal bytes: %v", err)
+        }
+        return data, err
+    })
+```  
+
+
+
+
 ##### Start Stream
 Start your stream. This is a long-running HTTP GET request. 
 You can get specific data you want by adding [query params](https://developer.twitter.com/en/docs/twitter-api/tweets/filtered-stream/api-reference/get-tweets-search-stream).
-Additionally, [view an example of query params here](https://developer.twitter.com/en/docs/twitter-api/expansions).
+Additionally, [view an example of query params here](https://developer.twitter.com/en/docs/twitter-api/expansions), or in the [examples](https://github.com/fallenstedt/twitter-stream/tree/master/example)
 
 ```go
     err := api.Stream.StartStream("")
@@ -54,7 +77,7 @@ Additionally, [view an example of query params here](https://developer.twitter.c
 	}
 ```
 
-4. Consume Messages from the Stream
+Consume Messages from the Stream
 Handle any `io.EOF` and other errors that arise first, then unmarshal your bytes into your favorite struct. Below is an example with strings 
 ```go
 	go func() {

VERSION

@@ -1 +1 @@
-0.3.0
+0.3.1

example/main.go

@@ -1,8 +1,10 @@
 package main
 
 import (
+	"encoding/json"
 	"fmt"
 	twitterstream "github.com/fallenstedt/twitter-stream"
+	"log"
 	"time"
 )
 
@@ -13,12 +15,33 @@ const secret = "YOUR_SECRET"
 func main() {
 	// Use your favorite function from below here
  	startStream()
- 	//addRules()
- 	//getRules()
-	//deleteRules()
+ 	addRules()
+ 	getRules()
+	deleteRules()
+}
+
+type StreamData struct {
+	Data struct {
+		Text      string    `json:"text"`
+		ID        string    `json:"id"`
+		CreatedAt time.Time `json:"created_at"`
+		AuthorID  string    `json:"author_id"`
+	} `json:"data"`
+	Includes struct {
+		Users []struct {
+			ID       string `json:"id"`
+			Name     string `json:"name"`
+			Username string `json:"username"`
+		} `json:"users"`
+	} `json:"includes"`
+	MatchingRules []struct {
+		ID  int64  `json:"id"`
+		Tag string `json:"tag"`
+	} `json:"matching_rules"`
 }
 
 
+
 func startStream() {
 	tok, err := twitterstream.NewTokenGenerator().SetApiKeyAndSecret(key, secret).RequestBearerToken()
 
@@ -27,8 +50,18 @@ func startStream() {
 	}
 
 	api := twitterstream.NewTwitterStream(tok.AccessToken)
+	// It is encouraged you unmarashal json with twitterstream's unmarshal hook. This is a thread-safe
+	// way to unmarshal json
+	api.Stream.SetUnmarshalHook(func(bytes []byte) (interface{}, error) {
+		data := StreamData{}
+		if err := json.Unmarshal(bytes, &data); err != nil {
+			log.Printf("Failed to unmarshal bytes: %v", err)
+		}
+		return data, err
+	})
+	defer  api.Stream.StopStream()
 
-	err = api.Stream.StartStream("")
+	err = api.Stream.StartStream("?expansions=author_id&tweet.fields=created_at")
 
 	if err != nil {
 		panic(err)
@@ -39,7 +72,14 @@ func startStream() {
 			if message.Err != nil {
 				panic(message.Err)
 			}
-			fmt.Println(string(message.Data))
+
+			// Type assertion
+			tweet, ok := message.Data.(StreamData)
+			if !ok {
+				continue
+			}
+
+			fmt.Println(tweet.Data.Text)
 		}
 	}()
 

stream.go

@@ -2,78 +2,92 @@ package twitterstream
 
 import (
 	"net/http"
-	"sync"
 )
 
+
 type (
+	// UnmarshalHook is a function that will unmarshal json.
+	UnmarshalHook func([]byte) (interface{}, error)
+
 	// IStream is the interface that the stream struct implements.
 	IStream interface {
 		StartStream(queryParams string) error
 		StopStream()
 		GetMessages() <-chan StreamMessage
+		SetUnmarshalHook(hook UnmarshalHook)
 	}
 
 	// StreamMessage is the message that is sent from the messages channel.
 	StreamMessage struct {
-		Data []byte
+		Data interface{}
 		Err  error
 	}
 
-
-	stream struct {
+	// Stream is the struct that manages a long running TCP connection with Twitter.
+	// It accepts an 'unamarshalHook' which allows you to unmarshal json in a thread-safe manner.
+	// It is highly encouraged to set a unmarshal hook before starting a stream. Unmarshaling json
+	// in a separate goroutine is not recommended because the Go bytes.Buffer is not thread safe.
+	Stream struct {
+		unmarshalHook UnmarshalHook
 		messages   chan StreamMessage
 		httpClient IHttpClient
 		done       chan struct{}
-		group      *sync.WaitGroup
 		reader     IStreamResponseBodyReader
 	}
 )
 
-func newStream(httpClient IHttpClient, reader IStreamResponseBodyReader) *stream {
-	return &stream{
+func newStream(httpClient IHttpClient, reader IStreamResponseBodyReader) *Stream {
+	return &Stream{
+		unmarshalHook: func(bytes []byte) (interface{}, error) {
+			return bytes, nil
+		},
 		messages:   make(chan StreamMessage),
 		done:       make(chan struct{}),
-		group:      new(sync.WaitGroup),
 		reader:     reader,
 		httpClient: httpClient,
 	}
 }
 
+// SetUnmarshalHook sets the function that unmarshals json. It is highly encouraged
+// that you unmarshal json with this hook to promote thread-safety. Go's bytes.Buffer is not
+// thread safe and can result in panics when a bytes.Buffer is shared across goroutines.
+func (s *Stream) SetUnmarshalHook(hook UnmarshalHook) {
+	s.unmarshalHook = hook
+}
+
 // GetMessages returns the read-only messages channel
-func (s *stream) GetMessages() <-chan StreamMessage {
+func (s *Stream) GetMessages() <-chan StreamMessage {
 	return s.messages
 }
 
 // StopStream sends a close signal to stop the stream of tweets.
-func (s *stream) StopStream() {
+func (s *Stream) StopStream() {
 	close(s.done)
 }
 
-// StartStream makes an HTTP GET request to twitter and starts streaming tweets to the Messages channel using Server Sent Events.
+// StartStream makes an HTTP GET request to twitter and starts streaming tweets to the Messages channel.
 // Accepts query params described in GET /2/tweets/search/stream to expand the payload that is returned. Query params string must begin with a ?.
 // See available query params here https://developer.twitter.com/en/docs/twitter-api/tweets/filtered-stream/api-reference/get-tweets-search-stream.
 // See an example here: https://developer.twitter.com/en/docs/twitter-api/expansions.
-func (s *stream) StartStream(optionalQueryParams string) error {
+func (s *Stream) StartStream(optionalQueryParams string) error {
 	res, err := s.httpClient.getSearchStream(optionalQueryParams)
 
 	if err != nil {
 		return err
 	}
 
 	s.reader.setStreamResponseBody(res.Body)
-	s.group.Add(1)
 
 	go s.streamMessages(res)
 
 	return nil
 }
 
-func (s *stream) streamMessages(res *http.Response) {
+func (s *Stream) streamMessages(res *http.Response) {
 	defer res.Body.Close()
-	defer s.group.Done()
 
 	for !stopped(s.done) {
-		data, err := s.reader.readNext()
+		b, err := s.reader.readNext()
 		if err != nil {
 			s.messages <- StreamMessage{
 				Data: nil,
@@ -82,14 +96,16 @@ func (s *stream) streamMessages(res *http.Response) {
 			s.StopStream()
 			break
 		}
-		if len(data) == 0 {
+		if len(b) == 0 {
 			// empty keep-alive
 			continue
 		}
 
+		data, err := s.unmarshalHook(b)
+
 		s.messages <- StreamMessage{
 			Data: data,
-			Err:  nil,
+			Err:  err,
 		}
 	}
 }

stream_test.go

@@ -87,7 +87,12 @@ func TestStartStream(t *testing.T) {
 				}
 			}()
 			r := <-result
-			if string(tt.result.Data) != string(r.Data) {
+
+			expected, _ := tt.result.Data.([]byte)
+			res, _ := r.Data.([]byte)
+
+
+			if string(expected) != string(res) {
 				t.Errorf("got %v, want %s", result, tt.result)
 			}
 		})