Move events api to its own package because its a cross-cutting concern

Author: dmashuda

internal/dev_server/api/events/events_stream.go

@@ -0,0 +1,82 @@
+package events
+
+import (
+	"context"
+	"encoding/json"
+	"log"
+	"net/http"
+
+	"github.com/google/uuid"
+	"github.com/launchdarkly/ldcli/internal/dev_server/model"
+	"github.com/pkg/errors"
+
+	"github.com/launchdarkly/ldcli/internal/dev_server/sdk"
+)
+
+type sdkEventObserver struct {
+	ctx             context.Context
+	debugSessionKey string
+	updateChan      chan<- sdk.Message
+}
+
+func newSdkEventObserver(updateChan chan<- sdk.Message, ctx context.Context) sdkEventObserver {
+	debugSessionKey := uuid.New().String()
+	db := model.EventStoreFromContext(ctx)
+	err := db.CreateDebugSession(ctx, debugSessionKey)
+	if err != nil {
+		log.Printf("sdkEventObserver: error writting debug session: %v", err)
+	}
+	return sdkEventObserver{
+		debugSessionKey: debugSessionKey,
+		ctx:             ctx,
+		updateChan:      updateChan,
+	}
+}
+
+func (o sdkEventObserver) Handle(message interface{}) {
+	str, ok := message.(json.RawMessage)
+	if !ok {
+		return
+	}
+
+	event := sdk.SDKEventBase{}
+	err := json.Unmarshal(str, &event)
+	if err != nil {
+		log.Printf("sdkEventObserver: error unmarshaling event: %v", err)
+		return
+	}
+
+	db := model.EventStoreFromContext(o.ctx)
+
+	err = db.WriteEvent(o.ctx, o.debugSessionKey, event.Kind, str)
+	if err != nil {
+		log.Printf("sdkEventObserver: error writting event: %v", err)
+		return
+	}
+
+	o.updateChan <- sdk.Message{Event: sdk.TYPE_PUT, Data: str}
+}
+
+func SdkEventsTeeHandler(writer http.ResponseWriter, request *http.Request) {
+	updateChan, errChan := sdk.OpenStream(
+		writer,
+		request.Context().Done(),
+		sdk.Message{Event: sdk.TYPE_PUT, Data: []byte{}},
+	)
+	defer close(updateChan)
+	observers := model.GetObserversFromContext(request.Context())
+
+	observerId := observers.RegisterObserver(newSdkEventObserver(updateChan, request.Context()))
+	defer func() {
+		ok := observers.DeregisterObserver(observerId)
+		if !ok {
+			log.Printf("unable to remove observer")
+		}
+	}()
+
+	err := <-errChan
+	if err != nil {
+		sdk.WriteError(request.Context(), writer, errors.Wrap(err, "stream failure"))
+		return
+	}
+}

internal/dev_server/api/events/routes.go

@@ -0,0 +1,7 @@
+package events
+
+import "github.com/gorilla/mux"
+
+func BindRoutes(router *mux.Router) {
+	router.HandleFunc("/events/tee", SdkEventsTeeHandler)
+}

internal/dev_server/dev_server.go

@@ -3,7 +3,6 @@ package dev_server
 import (
 	"context"
 	"fmt"
-	"github.com/launchdarkly/ldcli/internal/dev_server/events_db"
 	"log"
 	"net/http"
 	"os"
@@ -15,7 +14,9 @@ import (
 	"github.com/launchdarkly/ldcli/internal/client"
 	"github.com/launchdarkly/ldcli/internal/dev_server/adapters"
 	"github.com/launchdarkly/ldcli/internal/dev_server/api"
+	"github.com/launchdarkly/ldcli/internal/dev_server/api/events"
 	"github.com/launchdarkly/ldcli/internal/dev_server/db"
+	"github.com/launchdarkly/ldcli/internal/dev_server/events_db"
 	"github.com/launchdarkly/ldcli/internal/dev_server/model"
 	"github.com/launchdarkly/ldcli/internal/dev_server/sdk"
 	"github.com/launchdarkly/ldcli/internal/dev_server/ui"
@@ -77,6 +78,7 @@ func (c LDClient) RunServer(ctx context.Context, serverParams ServerParams) {
 		ui.AssetHandler.ServeHTTP(w, r)
 	})
 	sdk.BindRoutes(r)
+	events.BindRoutes(r)
 	handler := api.HandlerFromMux(apiServer, r)
 	handler = api.CorsHeadersWithConfig(serverParams.CorsEnabled, serverParams.CorsOrigin)(handler)
 	handler = handlers.CombinedLoggingHandler(os.Stdout, handler)

internal/dev_server/sdk/routes.go

@@ -12,7 +12,6 @@ func BindRoutes(router *mux.Router) {
 	// events
 	router.HandleFunc("/bulk", SdkEventsReceiveHandler)
 	router.HandleFunc("/diagnostic", DevNull)
-	router.HandleFunc("/events/tee", SdkEventsTeeHandler)
 	router.Handle("/events/bulk/{envId}", EventsCorsHeaders(DevNull))
 	router.Handle("/events/diagnostic/{envId}", EventsCorsHeaders(DevNull))
 	router.HandleFunc("/mobile", DevNull)

internal/dev_server/sdk/sdk_events.go

@@ -1,73 +1,25 @@
 package sdk
 
 import (
-	"context"
 	"encoding/json"
 	"io"
 	"log"
 	"net/http"
 
-	"github.com/google/uuid"
 	"github.com/launchdarkly/ldcli/internal/dev_server/model"
-	"github.com/pkg/errors"
 )
 
-func newSdkEventObserver(updateChan chan<- Message, ctx context.Context) sdkEventObserver {
-	debugSessionKey := uuid.New().String()
-	db := model.EventStoreFromContext(ctx)
-	err := db.CreateDebugSession(ctx, debugSessionKey)
-	if err != nil {
-		log.Printf("sdkEventObserver: error writting debug session: %v", err)
-	}
-	return sdkEventObserver{
-		debugSessionKey: debugSessionKey,
-		ctx:             ctx,
-		updateChan:      updateChan,
-	}
-}
-
 type SDKEventBase struct {
 	Kind string `json:"kind"`
 }
 
-type sdkEventObserver struct {
-	ctx             context.Context
-	debugSessionKey string
-	updateChan      chan<- Message
-}
-
-func (o sdkEventObserver) Handle(message interface{}) {
-	str, ok := message.(json.RawMessage)
-	if !ok {
-		return
-	}
-
-	event := SDKEventBase{}
-	err := json.Unmarshal(str, &event)
-	if err != nil {
-		log.Printf("sdkEventObserver: error unmarshaling event: %v", err)
-		return
-	}
-
-	db := model.EventStoreFromContext(o.ctx)
-
-	err = db.WriteEvent(o.ctx, o.debugSessionKey, event.Kind, str)
-	if err != nil {
-		log.Printf("sdkEventObserver: error writting event: %v", err)
-		return
-	}
-
-	o.updateChan <- Message{Event: TYPE_PUT, Data: str}
-}
-
-var observers *model.Observers = model.NewObservers()
-
 func SdkEventsReceiveHandler(writer http.ResponseWriter, request *http.Request) {
 	bodyStr, err := io.ReadAll(request.Body)
 	if err != nil {
 		log.Printf("SdkEventsReceiveHandler: error reading request body: %v", err)
 		return
 	}
+	observers := model.GetObserversFromContext(request.Context())
 
 	var arr []json.RawMessage
 	err = json.Unmarshal(bodyStr, &arr)
@@ -83,26 +35,3 @@ func SdkEventsReceiveHandler(writer http.ResponseWriter, request *http.Request)
 	writer.Header().Set("Content-Type", "application/json")
 	writer.WriteHeader(http.StatusAccepted)
 }
-
-func SdkEventsTeeHandler(writer http.ResponseWriter, request *http.Request) {
-	updateChan, errChan := OpenStream(
-		writer,
-		request.Context().Done(),
-		Message{Event: TYPE_PUT, Data: []byte{}},
-	)
-	defer close(updateChan)
-
-	observerId := observers.RegisterObserver(newSdkEventObserver(updateChan, request.Context()))
-	defer func() {
-		ok := observers.DeregisterObserver(observerId)
-		if !ok {
-			log.Printf("unable to remove observer")
-		}
-	}()
-
-	err := <-errChan
-	if err != nil {
-		WriteError(request.Context(), writer, errors.Wrap(err, "stream failure"))
-		return
-	}
-}