Initial commit - FR: realtime event listener for RTDB (#229)

Author: johndelavega

cmd/listen/main.go

@@ -0,0 +1,107 @@
+// 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 main
+
+import (
+	"fmt"
+	"log"
+	"os"
+
+	"golang.org/x/net/context"
+	"google.golang.org/api/option"
+
+	firebase "firebase.google.com/go"
+	"firebase.google.com/go/db"
+)
+
+func main() {
+
+	// opt := option.WithCredentialsFile("c:/users/username/.firebase/firebase.json") // Windows
+	//
+	opt := option.WithCredentialsFile("/home/username/.firebase/firebase.json") // Linux, edit 1.
+
+	config := &firebase.Config{
+		DatabaseURL: "https://mydb.firebaseio.com", // edit 2.
+	}
+
+	app, err := firebase.NewApp(context.Background(), config, opt)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	ctx := context.Background()
+
+	// DatabaseWithURL
+	client, err := app.Database(ctx)
+
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	testpath := "user1/path1"
+	ref := client.NewRef(testpath)
+
+	args := os.Args
+	if len(args) > 1 {
+		triggerEvent(ctx, client, testpath, args[1])
+		return // exit app
+	}
+
+	// SnapshotIterator
+	iter, err := ref.Listen(ctx)
+	if err != nil {
+		fmt.Printf(" Error: failed to create Listener %v\n", err)
+		return
+	}
+
+	fmt.Printf("client app | Ref Path: %s | iter.Snapshot = %v\n", ref.Path, iter.Snapshot)
+	fmt.Printf("           | Ref Key: %s \n", ref.Key)
+
+	defer iter.Stop()
+
+	go func() {
+		for {
+			event, err := iter.Next()
+
+			if err != nil {
+				break
+			}
+
+			fmt.Printf("client app | Ref Path: %s | event.Path %s | event.Snapshot() = %v\n", ref.Path, event.Path, event.Snapshot())
+			fmt.Printf("\n")
+		}
+	}()
+
+	fmt.Printf("\n >>> edit value of any key from %s in firebase console to trigger event\n\n", testpath)
+	fmt.Printf("\n >>> press <enter> to close http connection\n\n")
+	fmt.Printf("Waiting for events...\n\n")
+
+	fmt.Scanln()
+	iter.Stop()
+
+	fmt.Printf("\n >>> press <enter> to exit app\n\n\n")
+	fmt.Scanln()
+}
+
+func triggerEvent(ctx context.Context, client *db.Client, testpath string, val string) {
+
+	ref := client.NewRef(testpath + "/key1")
+
+	if err := ref.Set(ctx, val); err != nil {
+		log.Fatal(err)
+	} else {
+		fmt.Printf("OK - Set %s to %v\n", testpath+"/key1", val)
+	}
+}

db/event.go

@@ -0,0 +1,91 @@
+// 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 (
+	"encoding/json"
+	"errors"
+	"net/http"
+)
+
+// SSE = Sever-Sent Events = ssevent
+
+// EventType ...
+type EventType uint
+
+// EventType ...
+const (
+	ChildChanged EventType = iota
+	ChildAdded             // to be implemented
+	ChildRemoved           // to be implemented
+	ValueChanged           // to be implemented
+)
+
+// Event Sever-Sent Event object
+type Event struct {
+	EventType EventType // ChildChanged, ValueChanged, ChildAdded, ChildRemoved
+
+	Data string // JSON-encoded snapshot
+	Path string // snapshot path
+}
+
+// SnapshotIterator iterator for continuous event
+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
+	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
+}
+
+// Unmarshal current snapshot Event.Data
+func (e *Event) Unmarshal(v interface{}) error {
+
+	return json.Unmarshal([]byte(e.Data), v)
+}
+
+// Next ...
+func (i *SnapshotIterator) Next() (*Event, error) {
+
+	// prevent channel block
+	if i.active == false {
+		return nil, errors.New("SnapshotIterator is not active")
+	}
+
+	sseDataString := <-i.SSEDataChan
+
+	// todo: determine EventType
+
+	path, ss, err := splitSSEData(sseDataString)
+
+	return &Event{
+		EventType: ChildChanged,
+		Data:      ss, // snapshot
+		Path:      path,
+	}, err
+
+} // Next()
+
+// Stop ...
+func (i *SnapshotIterator) Stop() {
+	i.active = false
+	if i.resp != nil {
+		i.resp.Body.Close()
+	}
+}