fix: Update changeset builder to support chaining calls (#333)

Author: keelerm84

internal/datasourcev2/fdv1_polling_data_source.go

@@ -34,17 +34,14 @@ func (r *fdv1ToFDv2Requester) Request(
 	}
 
 	changeSetBuilder := subsystems.NewChangeSetBuilder()
-	err = changeSetBuilder.Start(subsystems.ServerIntent{
+	changeSetBuilder.Start(subsystems.ServerIntent{
 		Payload: subsystems.Payload{
 			ID:     "arbitrary-id",
 			Target: 0,
 			Code:   code,
 			Reason: reason,
 		},
 	})
-	if err != nil {
-		return nil, headers, err
-	}
 
 	for _, item := range data {
 		kind := subsystems.FlagKind

internal/datasourcev2/polling_http_request.go

@@ -5,11 +5,10 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"maps"
 	"net/http"
 	"net/url"
 
-	"maps"
-
 	"github.com/gregjones/httpcache"
 
 	"github.com/launchdarkly/go-sdk-common/v3/ldlog"
@@ -106,9 +105,7 @@ func (r *pollingRequester) Request(
 				if serverIntent.Payload.Code == subsystems.IntentNone {
 					return changeSet.NoChanges(), headers, nil
 				}
-				if err := changeSet.Start(serverIntent); err != nil {
-					return nil, headers, err
-				}
+				changeSet.Start(serverIntent)
 			case subsystems.EventPutObject:
 				var put subsystems.PutObject
 				if err := json.Unmarshal(event.Data, &put); err != nil {

internal/datasourcev2/streaming_data_source.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"encoding/json"
 	"errors"
+	"maps"
 	"net/http"
 	"net/url"
 	"sync"
@@ -21,8 +22,6 @@ import (
 	"github.com/launchdarkly/go-server-sdk/v7/internal/datasource"
 	"github.com/launchdarkly/go-server-sdk/v7/internal/endpoints"
 	"github.com/launchdarkly/go-server-sdk/v7/subsystems"
-
-	"maps"
 )
 
 const (
@@ -205,10 +204,7 @@ func (sp *StreamProcessor) consumeStream(stream *es.Stream, resultChan chan<- su
 					break
 				}
 
-				if err := changeSetBuilder.Start(serverIntent); err != nil {
-					gotMalformedEvent(event, err)
-					break
-				}
+				changeSetBuilder.Start(serverIntent)
 
 				// IntentNone is a special case where we won't receive a payload-transferred event, so we will need
 				// to instead immediately notify the client that we are initialized.

ldfiledatav2/file_data_source_impl.go

@@ -305,17 +305,14 @@ func mergeFileData(
 	allFileData ...fileData,
 ) (*subsystems.ChangeSet, error) {
 	builder := subsystems.NewChangeSetBuilder()
-	err := builder.Start(subsystems.ServerIntent{
+	builder.Start(subsystems.ServerIntent{
 		Payload: subsystems.Payload{
 			ID:     "",
 			Target: version,
 			Code:   subsystems.IntentTransferFull,
 			Reason: "payload-missing",
 		},
 	})
-	if err != nil {
-		return nil, fmt.Errorf("error starting change set: %w", err)
-	}
 
 	seenKeys := map[subsystems.ObjectKind]map[string]bool{
 		subsystems.FlagKind:    {},

subsystems/changeset.go

@@ -183,10 +183,10 @@ func (c *ChangeSetBuilder) Empty(selector Selector) *ChangeSet {
 }
 
 // Start begins a new change set with a given intent.
-func (c *ChangeSetBuilder) Start(intent ServerIntent) error {
+func (c *ChangeSetBuilder) Start(intent ServerIntent) *ChangeSetBuilder {
 	c.intent = &intent
 	c.changes = nil
-	return nil
+	return c
 }
 
 // ExpectChanges ensures that the current ChangeSetBuilder is prepared to handle changes.
@@ -239,22 +239,26 @@ func (c *ChangeSetBuilder) Finish(selector Selector) (*ChangeSet, error) {
 }
 
 // AddPut adds a new object to the changeset.
-func (c *ChangeSetBuilder) AddPut(kind ObjectKind, key string, version int, object json.RawMessage) {
+func (c *ChangeSetBuilder) AddPut(kind ObjectKind, key string, version int, object json.RawMessage) *ChangeSetBuilder {
 	c.changes = append(c.changes, Change{
 		Action:  ChangeTypePut,
 		Kind:    kind,
 		Key:     key,
 		Version: version,
 		Object:  object,
 	})
+
+	return c
 }
 
 // AddDelete adds a deletion to the changeset.
-func (c *ChangeSetBuilder) AddDelete(kind ObjectKind, key string, version int) {
+func (c *ChangeSetBuilder) AddDelete(kind ObjectKind, key string, version int) *ChangeSetBuilder {
 	c.changes = append(c.changes, Change{
 		Action:  ChangeTypeDelete,
 		Kind:    kind,
 		Key:     key,
 		Version: version,
 	})
+
+	return c
 }

subsystems/changeset_test.go

@@ -18,16 +18,15 @@ func TestChangeSetBuilder_MustStartToFinish(t *testing.T) {
 	_, err := builder.Finish(selector)
 	assert.Error(t, err)
 
-	assert.NoError(t, builder.Start(ServerIntent{Payload: Payload{Code: IntentNone}}))
+	builder.Start(ServerIntent{Payload: Payload{Code: IntentNone}})
 
 	_, err = builder.Finish(selector)
 	assert.NoError(t, err)
 }
 
 func TestChangeSetBuilder_Changes(t *testing.T) {
 	builder := NewChangeSetBuilder()
-	err := builder.Start(ServerIntent{Payload: Payload{Code: IntentTransferChanges}})
-	assert.NoError(t, err)
+	builder.Start(ServerIntent{Payload: Payload{Code: IntentTransferChanges}})
 
 	builder.AddPut("foo", "bar", 1, []byte("baz"))
 	builder.AddDelete("foo", "bar", 1)
@@ -44,7 +43,6 @@ func TestChangeSetBuilder_Changes(t *testing.T) {
 
 	assert.Equal(t, IntentTransferChanges, changeSet.IntentCode())
 	assert.Equal(t, selector, changeSet.Selector())
-
 }
 
 // After receiving an intent, the SDK may receive 1 or more objects before receiving a payload-transferred.
@@ -54,8 +52,7 @@ func TestChangeSetBuilder_Changes(t *testing.T) {
 // send one.
 func TestChangeSetBuilder_ImplicitIntentXferChanges(t *testing.T) {
 	builder := NewChangeSetBuilder()
-	err := builder.Start(ServerIntent{Payload: Payload{Code: IntentTransferFull}})
-	assert.NoError(t, err)
+	builder.Start(ServerIntent{Payload: Payload{Code: IntentTransferFull}})
 
 	changes1, err := builder.Finish(NewSelector("foo", 1))
 	assert.NoError(t, err)

testhelpers/ldtestdatav2/test_data_source.go

@@ -205,18 +205,15 @@ func (t *TestDataSynchronizer) makeFullTransferChangeset() (*subsystems.ChangeSe
 	version := t.version
 	t.version++
 
-	builder := subsystems.NewChangeSetBuilder()
-	err := builder.Start(subsystems.ServerIntent{
-		Payload: subsystems.Payload{
-			ID:     "",
-			Target: version,
-			Code:   subsystems.IntentTransferFull,
-			Reason: "payload-missing",
-		},
-	})
-	if err != nil {
-		return nil, err
-	}
+	builder := subsystems.NewChangeSetBuilder().
+		Start(subsystems.ServerIntent{
+			Payload: subsystems.Payload{
+				ID:     "",
+				Target: version,
+				Code:   subsystems.IntentTransferFull,
+				Reason: "payload-missing",
+			},
+		})
 
 	for key, item := range t.currentFlags {
 		json, err := json.Marshal(item.Item)
@@ -246,18 +243,15 @@ func (t *TestDataSynchronizer) makeTransferChangesForObject(
 	version := t.version
 	t.version++
 
-	builder := subsystems.NewChangeSetBuilder()
-	err := builder.Start(subsystems.ServerIntent{
-		Payload: subsystems.Payload{
-			ID:     "",
-			Target: version,
-			Code:   subsystems.IntentTransferChanges,
-			Reason: "changes",
-		},
-	})
-	if err != nil {
-		return nil, err
-	}
+	builder := subsystems.NewChangeSetBuilder().
+		Start(subsystems.ServerIntent{
+			Payload: subsystems.Payload{
+				ID:     "",
+				Target: version,
+				Code:   subsystems.IntentTransferChanges,
+				Reason: "changes",
+			},
+		})
 
 	json, err := json.Marshal(item.Item)
 	if err != nil {