Function concurrentBatchWriteEntity

lib/gcpspanner/client.go:1679–1738

(
	ctx context.Context, c *Client, wg *sync.WaitGroup, batchSize int,
	entityChan <-chan SpannerStruct, toMutationFn func(SpannerStruct) (*spanner.Mutation, error),
	table string, errChan chan error, workerID int)

	ExternalStruct any,
	SpannerStruct any,
	Key comparable](c *Client) *entityWriter[M, ExternalStruct, SpannerStruct, Key] {
	return &entityWriter[M, ExternalStruct, SpannerStruct, Key]{c}
}

func newEntityReader[
	M readableEntityMapper[ExternalStruct, SpannerStruct, Key],
	SpannerStruct any,
	ExternalStruct any,
	Key comparable](c *Client) *entityReader[M, ExternalStruct, SpannerStruct, Key] {
	return &entityReader[M, ExternalStruct, SpannerStruct, Key]{c}
}

func newEntityRemover[
	M removableEntityMapper[ExternalStruct, SpannerStruct, Key],
	SpannerStruct any,
	ExternalStruct any,
	Key comparable](c *Client) *entityRemover[M, ExternalStruct, SpannerStruct, Key] {
	return &entityRemover[M, ExternalStruct, SpannerStruct, Key]{c}
}

func concurrentBatchWriteEntity[SpannerStruct any](
	ctx context.Context, c *Client, wg *sync.WaitGroup, batchSize int,
	entityChan <-chan SpannerStruct, toMutationFn func(SpannerStruct) (*spanner.Mutation, error),
	table string, errChan chan error, workerID int) {
	var totalBatches, entityCount uint
	success := true
	defer func() {
		wg.Done()
		slog.InfoContext(ctx, "batch writer worker finishing", "id", workerID,
			"totalBatches", totalBatches, "entityCount", entityCount,
			"success", success, "table", table)
	}()
	slog.InfoContext(ctx, "batch writer worker starting", "id", workerID, "table", table)
	for {
		batch := make([]*spanner.Mutation, 0, batchSize)
		for range batchSize {
			select {
			case entity, isChannelStillOpen := <-entityChan:
				// If the channel is closed, go ahead and apply what we have and return.
				if !isChannelStillOpen {
					if len(batch) > 0 {
						slog.InfoContext(ctx, "sending final batch", "size", len(batch), "id", workerID, "table", table)
						totalBatches++
						entityCount += uint(len(batch))
						err := c.BatchWriteMutations(ctx, c.Client, batch)
						if err != nil {
							success = false
							errChan <- err
						}
					}

					return
				}
				// Else, the channel is still open and it has received a value.
				// Create a mutation and append it to the upcoming batch
				m, err := toMutationFn(entity)
				if err != nil {
					success = false