chore: Hook up flag change listeners to FDv2

Author: keelerm84

internal/datasourcev2/streaming_data_source_test.go

@@ -115,8 +115,7 @@ func TestStreamingProcessorDoesNotUseConfiguredTimeoutAsReadTimeout(t *testing.T
 		}}).
 		WithPutObjects(ldservicesv2.NewServerSDKData().ToPutObjects()).
 		WithTransferred(1)
-	streamHandler, stream := ldservices.ServerSideStreamingV2ServiceHandler(protocol.Next())
-	protocol.Enqueue(stream)
+	streamHandler, _ := ldservices.ServerSideStreamingV2ServiceProtocolHandler(protocol)
 
 	dd := mocks.NewMockDataDestination(datastore.NewInMemoryDataStore(sharedtest.NewTestLoggers()))
 
@@ -154,7 +153,12 @@ func TestStreamProcessorRecoverableErrorsCauseStreamRestart(t *testing.T) {
 	expectRestart := func(t *testing.T, p streamingTestParams) {
 		// Allow time for a reconnect (which sends the initial payload (server-intent), then we can queue up the transferred event.
 		<-time.After(300 * time.Millisecond)
-		p.protocol.WithTransferred(1).Enqueue(p.stream)
+		p.protocol.WithIntent(fdv2proto.ServerIntent{
+			Payload: fdv2proto.Payload{
+				ID: "fake-id", Target: 0, Code: "none", Reason: "caughtup",
+			}}).
+			WithTransferred(1).
+			Enqueue(p.stream)
 
 		<-p.requests
 		th.RequireValue(t, p.requests, time.Millisecond*300, "expected stream restart, did not see one")
@@ -251,8 +255,7 @@ func runStreamingTest(
 		WithTransferred(1)
 
 	events := make(chan eventsource.Event, 1000)
-	streamHandler, stream := ldservices.ServerSideStreamingV2ServiceHandler(protocol.Next())
-	protocol.Enqueue(stream)
+	streamHandler, stream := ldservices.ServerSideStreamingV2ServiceProtocolHandler(protocol)
 
 	handler, requestsCh := httphelpers.RecordingHandler(streamHandler)
 
@@ -307,8 +310,7 @@ func testStreamProcessorRecoverableHTTPError(t *testing.T, statusCode int) {
 		}}).
 		WithPutObjects(data.ToPutObjects()).
 		WithTransferred(1)
-	streamHandler, streamSender := ldservices.ServerSideStreamingV2ServiceHandler(protocol.Next())
-	protocol.Enqueue(streamSender)
+	streamHandler, _ := ldservices.ServerSideStreamingV2ServiceProtocolHandler(protocol)
 
 	sequentialHandler := httphelpers.SequentialHandler(
 		httphelpers.HandlerWithStatus(statusCode), // fails the first time

internal/datasystem/data_model_dependencies.go

@@ -0,0 +1,67 @@
+package datasystem
+
+import (
+	"github.com/launchdarkly/go-server-sdk/v7/internal/toposort"
+	st "github.com/launchdarkly/go-server-sdk/v7/subsystems/ldstoretypes"
+)
+
+// Maintains a bidirectional dependency graph that can be updated whenever an item has changed.
+//
+// This type is duplicated from internal.datasource. This duplication will be
+// removed when FDv1 is removed.
+type dependencyTracker struct {
+	dependenciesFrom toposort.AdjacencyList
+	dependenciesTo   toposort.AdjacencyList
+}
+
+func newDependencyTracker() *dependencyTracker {
+	return &dependencyTracker{
+		make(toposort.AdjacencyList),
+		make(toposort.AdjacencyList),
+	}
+}
+
+// Updates the dependency graph when an item has changed.
+func (d *dependencyTracker) updateDependenciesFrom(
+	kind st.DataKind,
+	fromKey string,
+	fromItem st.ItemDescriptor,
+) {
+	fromWhat := toposort.NewVertex(kind, fromKey)
+	updatedDependencies := toposort.GetNeighbors(kind, fromItem)
+
+	oldDependencySet := d.dependenciesFrom[fromWhat]
+	for oldDep := range oldDependencySet {
+		depsToThisOldDep := d.dependenciesTo[oldDep]
+		if depsToThisOldDep != nil {
+			delete(depsToThisOldDep, fromWhat)
+		}
+	}
+
+	d.dependenciesFrom[fromWhat] = updatedDependencies
+	for newDep := range updatedDependencies {
+		depsToThisNewDep := d.dependenciesTo[newDep]
+		if depsToThisNewDep == nil {
+			depsToThisNewDep = make(toposort.Neighbors)
+			d.dependenciesTo[newDep] = depsToThisNewDep
+		}
+		depsToThisNewDep.Add(fromWhat)
+	}
+}
+
+func (d *dependencyTracker) reset() {
+	d.dependenciesFrom = make(toposort.AdjacencyList)
+	d.dependenciesTo = make(toposort.AdjacencyList)
+}
+
+// Populates the given set with the union of the initial item and all items that directly or indirectly
+// depend on it (based on the current state of the dependency graph).
+func (d *dependencyTracker) addAffectedItems(itemsOut toposort.Neighbors, initialModifiedItem toposort.Vertex) {
+	if !itemsOut.Contains(initialModifiedItem) {
+		itemsOut.Add(initialModifiedItem)
+		affectedItems := d.dependenciesTo[initialModifiedItem]
+		for affectedItem := range affectedItems {
+			d.addAffectedItems(itemsOut, affectedItem)
+		}
+	}
+}