refactor: cleanup store/streaming/constructor.go #14044

(cherry picked from commit c6189bb)

# Conflicts:
#	store/streaming/constructor.go
#	types/utils.go

store/streaming/constructor.go

@@ -13,6 +13,7 @@ import (
 	serverTypes "github.com/cosmos/cosmos-sdk/server/types"
 	"github.com/cosmos/cosmos-sdk/store/streaming/file"
 	"github.com/cosmos/cosmos-sdk/store/types"
+	sdk "github.com/cosmos/cosmos-sdk/types"
 
 	"github.com/spf13/cast"
 )
@@ -26,18 +27,23 @@ type ServiceType int
 const (
 	Unknown ServiceType = iota
 	File
-	// add more in the future
 )
 
 // Streaming option keys
 const (
+<<<<<<< HEAD
 	OptStreamersFilePrefix          = "streamers.file.prefix"
 	OptStreamersFileWriteDir        = "streamers.file.write_dir"
 	OptStreamersFileOutputMetadata  = "streamers.file.output-metadata"
 	OptStreamersFileStopNodeOnError = "streamers.file.stop-node-on-error"
 	OptStreamersFileFsync           = "streamers.file.fsync"
 
 	OptStoreStreamers = "store.streamers"
+=======
+	OptStreamersFilePrefix   = "streamers.file.prefix"
+	OptStreamersFileWriteDir = "streamers.file.write_dir"
+	OptStoreStreamers        = "store.streamers"
+>>>>>>> c6189bb63 (refactor: cleanup store/streaming/constructor.go #14044)
 )
 
 // ServiceTypeFromString returns the streaming.ServiceType corresponding to the
@@ -46,6 +52,7 @@ func ServiceTypeFromString(name string) ServiceType {
 	switch strings.ToLower(name) {
 	case "file", "f":
 		return File
+
 	default:
 		return Unknown
 	}
@@ -56,25 +63,30 @@ func (sst ServiceType) String() string {
 	switch sst {
 	case File:
 		return "file"
+
 	default:
 		return "unknown"
 	}
 }
 
-// ServiceConstructorLookupTable is a mapping of streaming.ServiceTypes to streaming.ServiceConstructors
+// ServiceConstructorLookupTable is a mapping of streaming.ServiceTypes to
+// streaming.ServiceConstructors types.
 var ServiceConstructorLookupTable = map[ServiceType]ServiceConstructor{
 	File: NewFileStreamingService,
 }
 
-// NewServiceConstructor returns the streaming.ServiceConstructor corresponding to the provided name
+// NewServiceConstructor returns the streaming.ServiceConstructor corresponding
+// to the provided name.
 func NewServiceConstructor(name string) (ServiceConstructor, error) {
 	ssType := ServiceTypeFromString(name)
 	if ssType == Unknown {
 		return nil, fmt.Errorf("unrecognized streaming service name %s", name)
 	}
+
 	if constructor, ok := ServiceConstructorLookupTable[ssType]; ok && constructor != nil {
 		return constructor, nil
 	}
+
 	return nil, fmt.Errorf("streaming service constructor of type %s not found", ssType.String())
 }
 
@@ -85,6 +97,7 @@ func NewFileStreamingService(
 	keys []types.StoreKey,
 	marshaller codec.BinaryCodec,
 ) (baseapp.StreamingService, error) {
+<<<<<<< HEAD
 	homePath := cast.ToString(opts.Get(flags.FlagHome))
 	filePrefix := cast.ToString(opts.Get(OptStreamersFilePrefix))
 	fileDir := cast.ToString(opts.Get(OptStreamersFileWriteDir))
@@ -105,6 +118,12 @@ func NewFileStreamingService(
 	}
 
 	return file.NewStreamingService(fileDir, filePrefix, keys, marshaller, outputMetadata, stopNodeOnErr, fsync)
+=======
+	filePrefix := cast.ToString(opts.Get(OptStreamersFilePrefix))
+	fileDir := cast.ToString(opts.Get(OptStreamersFileWriteDir))
+
+	return file.NewStreamingService(fileDir, filePrefix, keys, marshaller)
+>>>>>>> c6189bb63 (refactor: cleanup store/streaming/constructor.go #14044)
 }
 
 // LoadStreamingServices is a function for loading StreamingServices onto the
@@ -119,14 +138,19 @@ func LoadStreamingServices(
 ) ([]baseapp.StreamingService, *sync.WaitGroup, error) {
 	// waitgroup and quit channel for optional shutdown coordination of the streaming service(s)
 	wg := new(sync.WaitGroup)
+
 	// configure state listening capabilities using AppOptions
-	streamers := cast.ToStringSlice(appOpts.Get("store.streamers"))
+	streamers := cast.ToStringSlice(appOpts.Get(OptStoreStreamers))
 	activeStreamers := make([]baseapp.StreamingService, 0, len(streamers))
+
 	for _, streamerName := range streamers {
+		var exposeStoreKeys []types.StoreKey
+
 		// get the store keys allowed to be exposed for this streaming service
 		exposeKeyStrs := cast.ToStringSlice(appOpts.Get(fmt.Sprintf("streamers.%s.keys", streamerName)))
-		var exposeStoreKeys []types.StoreKey
-		if exposeAll(exposeKeyStrs) { // if list contains `*`, expose all StoreKeys
+
+		// if list contains '*', expose all store keys
+		if sdk.SliceContains(exposeKeyStrs, "*") {
 			exposeStoreKeys = make([]types.StoreKey, 0, len(keys))
 			for _, storeKey := range keys {
 				exposeStoreKeys = append(exposeStoreKeys, storeKey)
@@ -139,45 +163,46 @@ func LoadStreamingServices(
 				}
 			}
 		}
-		if len(exposeStoreKeys) == 0 { // short circuit if we are not exposing anything
+
+		if len(exposeStoreKeys) == 0 {
 			continue
 		}
-		// get the constructor for this streamer name
+
 		constructor, err := NewServiceConstructor(streamerName)
 		if err != nil {
-			// close any services we may have already spun up before hitting the error on this one
+			// Close any services we may have already spun up before hitting the error
+			// on this one.
 			for _, activeStreamer := range activeStreamers {
 				activeStreamer.Close()
 			}
+
 			return nil, nil, err
 		}
 
 		// Generate the streaming service using the constructor, appOptions, and the
 		// StoreKeys we want to expose.
 		streamingService, err := constructor(appOpts, exposeStoreKeys, appCodec)
 		if err != nil {
-			// close any services we may have already spun up before hitting the error on this one
+			// Close any services we may have already spun up before hitting the error
+			// on this one.
 			for _, activeStreamer := range activeStreamers {
 				activeStreamer.Close()
 			}
+
 			return nil, nil, err
 		}
+
 		// register the streaming service with the BaseApp
 		bApp.SetStreamingService(streamingService)
+
 		// kick off the background streaming service loop
 		streamingService.Stream(wg)
+
 		// add to the list of active streamers
 		activeStreamers = append(activeStreamers, streamingService)
 	}
-	// if there are no active streamers, activeStreamers is empty (len == 0) and the waitGroup is not waiting on anything
-	return activeStreamers, wg, nil
-}
 
-func exposeAll(list []string) bool {
-	for _, ele := range list {
-		if ele == "*" {
-			return true
-		}
-	}
-	return false
+	// If there are no active streamers, activeStreamers is empty (len == 0) and
+	// the waitGroup is not waiting on anything.
+	return activeStreamers, wg, nil
 }

types/utils.go

@@ -6,7 +6,13 @@ import (
 	"fmt"
 	"time"
 
+<<<<<<< HEAD
 	dbm "github.com/tendermint/tm-db"
+=======
+	log "github.com/tendermint/tendermint/libs/log"
+
+	"github.com/cosmos/cosmos-sdk/types/kv"
+>>>>>>> c6189bb63 (refactor: cleanup store/streaming/constructor.go #14044)
 )
 
 var (
@@ -104,3 +110,52 @@ func CopyBytes(bz []byte) (ret []byte) {
 	copy(ret, bz)
 	return ret
 }
+<<<<<<< HEAD
+=======
+
+// AppendLengthPrefixedBytes combines the slices of bytes to one slice of bytes.
+func AppendLengthPrefixedBytes(args ...[]byte) []byte {
+	length := 0
+	for _, v := range args {
+		length += len(v)
+	}
+	res := make([]byte, length)
+
+	length = 0
+	for _, v := range args {
+		copy(res[length:length+len(v)], v)
+		length += len(v)
+	}
+
+	return res
+}
+
+// ParseLengthPrefixedBytes panics when store key length is not equal to the given length.
+func ParseLengthPrefixedBytes(key []byte, startIndex int, sliceLength int) ([]byte, int) {
+	neededLength := startIndex + sliceLength
+	endIndex := neededLength - 1
+	kv.AssertKeyAtLeastLength(key, neededLength)
+	byteSlice := key[startIndex:neededLength]
+
+	return byteSlice, endIndex
+}
+
+// LogDeferred logs an error in a deferred function call if the returned error is non-nil.
+func LogDeferred(logger log.Logger, f func() error) {
+	if err := f(); err != nil {
+		logger.Error(err.Error())
+	}
+}
+
+// SliceContains implements a generic function for checking if a slice contains
+// a certain value.
+func SliceContains[T comparable](elements []T, v T) bool {
+	for _, s := range elements {
+		if v == s {
+			return true
+		}
+	}
+
+	return false
+}
+>>>>>>> c6189bb63 (refactor: cleanup store/streaming/constructor.go #14044)