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

(cherry picked from commit c6189bb)

# Conflicts:
#	store/streaming/constructor.go

store/streaming/constructor.go

@@ -13,32 +13,45 @@ import (
 	serverTypes "github.com/cosmos/cosmos-sdk/server/types"
 	"github.com/cosmos/cosmos-sdk/store/streaming/file"
 	"github.com/cosmos/cosmos-sdk/store/types"
+<<<<<<< HEAD
 	"github.com/tendermint/tendermint/libs/log"
+=======
+	sdk "github.com/cosmos/cosmos-sdk/types"
+>>>>>>> c6189bb63 (refactor: cleanup store/streaming/constructor.go #14044)
 
 	"github.com/spf13/cast"
 )
 
 // ServiceConstructor is used to construct a streaming service
+<<<<<<< HEAD
 type ServiceConstructor func(serverTypes.AppOptions, []types.StoreKey, codec.BinaryCodec, log.Logger) (baseapp.StreamingService, error)
+=======
+type ServiceConstructor func(serverTypes.AppOptions, []types.StoreKey, codec.BinaryCodec) (baseapp.StreamingService, error)
+>>>>>>> c6189bb63 (refactor: cleanup store/streaming/constructor.go #14044)
 
 // ServiceType enum for specifying the type of StreamingService
 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
@@ -47,6 +60,7 @@ func ServiceTypeFromString(name string) ServiceType {
 	switch strings.ToLower(name) {
 	case "file", "f":
 		return File
+
 	default:
 		return Unknown
 	}
@@ -57,25 +71,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 +104,7 @@ func NewFileStreamingService(
 	opts serverTypes.AppOptions,
 	keys []types.StoreKey,
 	marshaller codec.BinaryCodec,
+<<<<<<< HEAD
 	logger log.Logger,
 ) (baseapp.StreamingService, error) {
 	homePath := cast.ToString(opts.Get(flags.FlagHome))
@@ -107,6 +127,13 @@ func NewFileStreamingService(
 	}
 
 	return file.NewStreamingService(fileDir, filePrefix, keys, marshaller, logger, outputMetadata, stopNodeOnErr, fsync)
+=======
+) (baseapp.StreamingService, error) {
+	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
@@ -117,19 +144,27 @@ func LoadStreamingServices(
 	bApp *baseapp.BaseApp,
 	appOpts serverTypes.AppOptions,
 	appCodec codec.BinaryCodec,
+<<<<<<< HEAD
 	logger log.Logger,
+=======
+>>>>>>> c6189bb63 (refactor: cleanup store/streaming/constructor.go #14044)
 	keys map[string]*types.KVStoreKey,
 ) ([]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)
@@ -142,45 +177,50 @@ 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.
+<<<<<<< HEAD
 		streamingService, err := constructor(appOpts, exposeStoreKeys, appCodec, logger)
+=======
+		streamingService, err := constructor(appOpts, exposeStoreKeys, appCodec)
+>>>>>>> c6189bb63 (refactor: cleanup store/streaming/constructor.go #14044)
 		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,8 +6,9 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/cosmos/cosmos-sdk/types/kv"
 	log "github.com/tendermint/tendermint/libs/log"
+
+	"github.com/cosmos/cosmos-sdk/types/kv"
 )
 
 // SortedJSON takes any JSON and returns it sorted by keys. Also, all white-spaces
@@ -141,3 +142,15 @@ func LogDeferred(logger log.Logger, f func() error) {
 		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
+}