Fixed bug, SnapshotIterator.Stop() not working from reconnect code fix (#229)

added SnapshotIterator.Done()
updated test apps
fixed variable scope for resp *http.Response

Author: johndelavega

cmd/listen/main.go

@@ -73,6 +73,11 @@ func main() {
 
 	go func() {
 		for {
+
+			if iter.Done() {
+				break
+			}
+
 			event, err := iter.Next()
 
 			if err != nil {

cmd/listen2/main.go

@@ -86,6 +86,11 @@ func main() {
 
 	go func() {
 		for {
+
+			if iter.Done() {
+				break
+			}
+
 			event, err := iter.Next()
 
 			if err != nil {
@@ -123,6 +128,11 @@ func main() {
 
 	go func() {
 		for {
+
+			if iter2.Done() {
+				break
+			}
+
 			event, err := iter2.Next()
 
 			if err != nil {
@@ -136,12 +146,16 @@ func main() {
 
 	fmt.Printf("\n >>> open a new separate command line terminal, to trigger events, run: go run . anyvalue\n")
 	fmt.Printf("\n >>> OR edit value of any key from %s in firebase console to trigger events\n\n", testpath)
-	fmt.Printf("\n >>> press <enter> to close http connection\n\n")
+	fmt.Printf("\n >>> press <enter> to stop 1st Listener and close http connection\n\n")
 	fmt.Printf("Waiting for events...\n\n")
 
 	fmt.Scanln()
 	iter.Stop()
 
+	fmt.Printf("\n >>> press <enter> to stop 2nd Listener and close http connection\n\n")
+	fmt.Scanln()
+	iter2.Stop()
+
 	fmt.Printf("\n >>> press <enter> to exit app\n\n\n")
 	fmt.Scanln()
 }

db/event.go

@@ -22,7 +22,7 @@ import (
 
 // SSE = Sever-Sent Events = ssevent
 
-// EventType ...
+// EventType specific event type changes
 type EventType uint
 
 // EventType ...
@@ -41,16 +41,17 @@ type Event struct {
 	Path string // snapshot path
 }
 
-// SnapshotIterator iterator for continuous event
+// SnapshotIterator iterator for continuous events
 type SnapshotIterator struct {
 	Snapshot    string         // initial snapshot, JSON-encoded, returned from http Respoonse, server sent event, data part
 	SSEDataChan <-chan string  // continuous event snapshot, channel receive only, directional channel
+	done        *bool          // Done listening to events, also used to prevent channel block
 	resp        *http.Response // http connection keep alive
-	active      bool           // false when resp is closed, used to prevent channel block, defaults to false
 }
 
 // Snapshot ssevent data, data part
 func (e *Event) Snapshot() string {
+
 	return e.Data // ssevent data (snapshot), snapshot only, data part of ssevent data
 }
 
@@ -60,32 +61,38 @@ func (e *Event) Unmarshal(v interface{}) error {
 	return json.Unmarshal([]byte(e.Data), v)
 }
 
-// Next ...
-func (i *SnapshotIterator) Next() (*Event, error) {
+// Next realtime event
+func (it *SnapshotIterator) Next() (*Event, error) {
 
 	// prevent channel block
-	if i.active == false {
-		return nil, errors.New("SnapshotIterator is not active")
+	if *it.done == true {
+		return nil, errors.New("SnapshotIterator is done or no longer active")
 	}
 
-	sseDataString := <-i.SSEDataChan
+	sseDataString := <-it.SSEDataChan
 
 	// todo: determine EventType
 
-	path, ss, err := splitSSEData(sseDataString)
+	path, snapshot, err := splitSSEData(sseDataString)
 
 	return &Event{
 		EventType: ChildChanged,
-		Data:      ss, // snapshot
+		Data:      snapshot,
 		Path:      path,
 	}, err
 
 } // Next()
 
-// Stop ...
-func (i *SnapshotIterator) Stop() {
-	i.active = false
-	if i.resp != nil {
-		i.resp.Body.Close()
+// Stop listening for realtime events
+// close http connection
+func (it *SnapshotIterator) Stop() {
+	*it.done = true
+	if it.resp != nil {
+		it.resp.Body.Close()
 	}
 }
+
+// Done can be used to check if Stop() have been called
+func (it *SnapshotIterator) Done() bool {
+	return *it.done
+}

db/event_ref.go

@@ -26,7 +26,7 @@ import (
 	"firebase.google.com/go/internal"
 )
 
-// Listen ...
+// Listen returns an iterator that listens to realtime events
 func (r *Ref) Listen(ctx context.Context) (*SnapshotIterator, error) {
 
 	sseDataChan := make(chan string) // server-sent event data channel
@@ -38,23 +38,26 @@ func (r *Ref) Listen(ctx context.Context) (*SnapshotIterator, error) {
 
 	resp, err := r.sendListen(ctx, "GET", opts...)
 
+	done := true
+
 	if err != nil {
-		return &SnapshotIterator{active: false}, err
+		return &SnapshotIterator{done: &done}, err
 	}
 
 	snapshot, err := getInitialNodeSnapshot(resp)
 
 	if err != nil {
-		return &SnapshotIterator{active: false}, err
+		return &SnapshotIterator{done: &done}, err
 	}
 
-	go r.startListeningWithReconnect(ctx, opts, resp, sseDataChan)
+	done = false
+	go r.startListeningWithReconnect(ctx, opts, resp, &done, sseDataChan)
 
 	return &SnapshotIterator{
 		Snapshot:    snapshot,
 		SSEDataChan: sseDataChan,
-		resp:        resp, //*http.Response
-		active:      true,
+		done:        &done,
+		resp:        resp, // *http.Response
 	}, err
 
 } // Listen()
@@ -76,19 +79,24 @@ func getInitialNodeSnapshot(resp *http.Response) (string, error) {
 				b = scanner.Bytes()
 				s := string(b)
 
+				// sample sse data
+				// s = 'data: {"path":"/","data":{"test3":{"test4":4}}}'
+
 				var snapshotMap map[string]interface{}
 
+				// We only want the well formed json payload
+				// exclude or trim the first 6 chars 'data: '
 				err := json.Unmarshal([]byte(s[6:]), &snapshotMap)
 				if err != nil {
 					return "", err
 				}
+
 				snapshotByte, err := json.Marshal(snapshotMap["data"])
 				if err != nil {
 					return "", err
 				}
 
 				return string(snapshotByte), nil
-
 			}
 		}
 
@@ -98,13 +106,23 @@ func getInitialNodeSnapshot(resp *http.Response) (string, error) {
 }
 
 // called with goroutine
-func (r *Ref) startListeningWithReconnect(ctx context.Context, opts []internal.HTTPOption, resp *http.Response, sseDataChan chan<- string) {
+func (r *Ref) startListeningWithReconnect(
+	ctx context.Context,
+	opts []internal.HTTPOption,
+	resp *http.Response,
+	done *bool,
+	sseDataChan chan<- string) {
 
 	scanner := bufio.NewScanner(resp.Body)
 
 	var b []byte
 
 	for {
+
+		if *done {
+			break
+		}
+
 		if scanner.Scan() == true {
 
 			b = scanner.Bytes()
@@ -127,7 +145,8 @@ func (r *Ref) startListeningWithReconnect(ctx context.Context, opts []internal.H
 			} else if "event: auth_revoked" == string(b) {
 
 				// reconnect to re-establish authentication every hour
-				resp, err := r.sendListen(ctx, "GET", opts...)
+				var err error
+				resp, err = r.sendListen(ctx, "GET", opts...)
 
 				if err == nil {
 					// not part of existing continuing listening events, so we don't send to the listening channel
@@ -139,8 +158,10 @@ func (r *Ref) startListeningWithReconnect(ctx context.Context, opts []internal.H
 				scanner = bufio.NewScanner(resp.Body)
 			}
 		} else {
+
 			// attemp to reconnect for other connection problems
-			resp, err := r.sendListen(ctx, "GET", opts...)
+			var err error
+			resp, err = r.sendListen(ctx, "GET", opts...)
 
 			if err == nil {
 				// not part of existing continuing listening events, so we don't send to the listening channel

db/event_test.go

@@ -1,43 +1,50 @@
-// Copyright 2019 Google Inc. All Rights Reserved.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//      http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-package db
-
-import (
-	"context"
-	"testing"
-)
-
-func TestListen(t *testing.T) {
-	mock := &mockServer{}
-	srv := mock.Start(client)
-	defer srv.Close()
-
-	// mostly just a place holder, currenty test for crash only
-
-	iter, err := testref.Listen(context.Background())
-	_ = iter
-	_ = err
-
-	defer iter.Stop()
-
-	event, err := iter.Next()
-	_ = event
-	_ = err
-
-	if err == nil {
-		iter.Stop()
-	}
-
-}
+// Copyright 2019 Google Inc. All Rights Reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package db
+
+import (
+	"context"
+	"testing"
+)
+
+func TestListen(t *testing.T) {
+	mock := &mockServer{}
+	srv := mock.Start(client)
+	defer srv.Close()
+
+	// mostly just a place holder, currenty test for crash only
+
+	iter, err := testref.Listen(context.Background())
+	_ = iter
+	_ = err
+
+	defer iter.Stop()
+
+	event, err := iter.Next()
+	_ = event
+	_ = err
+
+	if iter.Done() {
+		// break
+	}
+
+	if err == nil {
+		iter.Stop()
+	}
+
+	b := iter.Done()
+	_ = b
+
+}