diff --git a/store/streaming/constructor.go b/store/streaming/constructor.go index 408c5c8362fa..979ce89786ea 100644 --- a/store/streaming/constructor.go +++ b/store/streaming/constructor.go @@ -13,13 +13,21 @@ 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 @@ -27,11 +35,11 @@ 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" @@ -39,6 +47,11 @@ const ( 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 } diff --git a/types/utils.go b/types/utils.go index 88c19f8d0069..468c02e5663b 100644 --- a/types/utils.go +++ b/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 +}