observer pattern

Author: moribellamy

internal/dev_server/sdk/sdk_events.go

@@ -1,41 +1,67 @@
 package sdk
 
 import (
-	"github.com/pkg/errors"
 	"io"
 	"log"
 	"net/http"
+	"sync"
+
+	"github.com/launchdarkly/ldcli/internal/dev_server/model"
+	"github.com/pkg/errors"
 )
 
+type sdkEventObserver struct {
+	updateChan chan<- Message
+}
+
+func (o sdkEventObserver) Handle(message interface{}) {
+	str, ok := message.(string)
+	if !ok {
+		return
+	}
+	o.updateChan <- Message{Event: TYPE_PUT, Data: []byte(str)}
+}
+
+var observers *model.Observers
+var once sync.Once
+
 func SdkEventsReceiveHandler(writer http.ResponseWriter, request *http.Request) {
-	log.Println(request.URL.Path)
+	once.Do(func() {
+		observers = model.NewObservers()
+	})
+
 	bodyStr, err := io.ReadAll(request.Body)
 	if err != nil {
 		log.Printf("SdkEventsReceiveHandler: error reading request body: %v", err)
 		return
 	}
-	log.Println(string(bodyStr))
+	if observers != nil {
+		observers.Notify(string(bodyStr))
+	}
 
 	writer.Header().Set("Content-Type", "application/json")
 	writer.WriteHeader(http.StatusAccepted)
 }
 
 func SdkEventsTeeHandler(writer http.ResponseWriter, request *http.Request) {
-	// Initialize SSE
+	once.Do(func() {
+		observers = model.NewObservers()
+	})
+
 	updateChan, errChan := OpenStream(
 		writer,
 		request.Context().Done(),
-		Message{Event: TYPE_PUT, Data: []byte("start")},
+		Message{Event: TYPE_PUT, Data: []byte{}},
 	)
 	defer close(updateChan)
 
-	// Use updateChan to continually send messages back to the client. OpenStream, above,
-	// takes care of flushing the data.
-	//
-	// If the client cancels the request, OpenStream will notice via request.Context.Done().
-	// Otherwise, this connection is never explicitly closed by us.
-	updateChan <- Message{Event: TYPE_PUT, Data: []byte("data1")}
-	updateChan <- Message{Event: TYPE_PUT, Data: []byte("data2")}
+	observerId := observers.RegisterObserver(sdkEventObserver{updateChan})
+	defer func() {
+		ok := observers.DeregisterObserver(observerId)
+		if !ok {
+			log.Printf("unable to remove observer")
+		}
+	}()
 
 	err := <-errChan
 	if err != nil {