chore: store audit

store/cachekv/memiterator.go

@@ -8,7 +8,7 @@ import (
 	"github.com/cosmos/cosmos-sdk/store/types"
 )
 
-// Iterates over iterKVCache items.
+// memIterator iterates over iterKVCache items.
 // if key is nil, means it was deleted.
 // Implements Iterator.
 type memIterator struct {

store/cachekv/store.go

@@ -15,14 +15,16 @@ import (
 	"github.com/cosmos/cosmos-sdk/types/kv"
 )
 
-// If value is nil but deleted is false, it means the parent doesn't have the
-// key.  (No need to delete upon Write())
+// cValue represents a cached value.
+// If dirty is true, it indicates the cached value is different from the underlying value.
 type cValue struct {
 	value []byte
 	dirty bool
 }
 
 // Store wraps an in-memory cache around an underlying types.KVStore.
+// If a cached value is nil but deleted is defined for the corresponding key,
+// it means the parent doesn't have the key. (No need to delete upon Write())
 type Store struct {
 	mtx           sync.Mutex
 	cache         map[string]*cValue

store/iavl/store.go

@@ -380,7 +380,7 @@ func getProofFromTree(tree *iavl.MutableTree, key []byte, exists bool) *tmcrypto
 
 //----------------------------------------
 
-// Implements types.Iterator.
+// iavlIterator implements types.Iterator.
 type iavlIterator struct {
 	*iavl.Iterator
 }

store/streaming/constructor_test.go

@@ -1,13 +1,19 @@
-package streaming
+package streaming_test
 
 import (
 	"testing"
 
+	"github.com/cosmos/cosmos-sdk/baseapp"
 	"github.com/cosmos/cosmos-sdk/codec"
 	codecTypes "github.com/cosmos/cosmos-sdk/codec/types"
+	serverTypes "github.com/cosmos/cosmos-sdk/server/types"
+	"github.com/cosmos/cosmos-sdk/simapp"
+	"github.com/cosmos/cosmos-sdk/store/streaming"
 	"github.com/cosmos/cosmos-sdk/store/streaming/file"
 	"github.com/cosmos/cosmos-sdk/store/types"
 	sdk "github.com/cosmos/cosmos-sdk/types"
+	"github.com/tendermint/tendermint/libs/log"
+	dbm "github.com/tendermint/tm-db"
 
 	"github.com/stretchr/testify/require"
 )
@@ -24,12 +30,12 @@ var (
 )
 
 func TestStreamingServiceConstructor(t *testing.T) {
-	_, err := NewServiceConstructor("unexpectedName")
+	_, err := streaming.NewServiceConstructor("unexpectedName")
 	require.NotNil(t, err)
 
-	constructor, err := NewServiceConstructor("file")
+	constructor, err := streaming.NewServiceConstructor("file")
 	require.Nil(t, err)
-	var expectedType ServiceConstructor
+	var expectedType streaming.ServiceConstructor
 	require.IsType(t, expectedType, constructor)
 
 	serv, err := constructor(mockOptions, mockKeys, testMarshaller)
@@ -41,3 +47,54 @@ func TestStreamingServiceConstructor(t *testing.T) {
 		require.True(t, ok)
 	}
 }
+
+func TestLoadStreamingServices(t *testing.T) {
+	db := dbm.NewMemDB()
+	encCdc := simapp.MakeTestEncodingConfig()
+	keys := sdk.NewKVStoreKeys("mockKey1", "mockKey2")
+	bApp := baseapp.NewBaseApp("appName", log.NewNopLogger(), db)
+
+	testCases := map[string]struct {
+		appOpts            serverTypes.AppOptions
+		activeStreamersLen int
+	}{
+		"empty app options": {
+			appOpts: simapp.EmptyAppOptions{},
+		},
+		"all StoreKeys exposed": {
+			appOpts:            streamingAppOptions{keys: []string{"*"}},
+			activeStreamersLen: 1,
+		},
+		"some StoreKey exposed": {
+			appOpts:            streamingAppOptions{keys: []string{"mockKey1"}},
+			activeStreamersLen: 1,
+		},
+		"not exposing anything": {
+			appOpts: streamingAppOptions{keys: []string{"mockKey3"}},
+		},
+	}
+
+	for name, tc := range testCases {
+		t.Run(name, func(t *testing.T) {
+			activeStreamers, _, err := streaming.LoadStreamingServices(bApp, tc.appOpts, encCdc.Codec, keys)
+			require.NoError(t, err)
+			require.Equal(t, tc.activeStreamersLen, len(activeStreamers))
+		})
+	}
+
+}
+
+type streamingAppOptions struct {
+	keys []string
+}
+
+func (ao streamingAppOptions) Get(o string) interface{} {
+	switch o {
+	case "store.streamers":
+		return []string{"file"}
+	case "streamers.file.keys":
+		return ao.keys
+	default:
+		return nil
+	}
+}

store/streaming/file/service.go

@@ -39,7 +39,7 @@ type IntermediateWriter struct {
 	outChan chan<- []byte
 }
 
-// NewIntermediateWriter create an instance of an intermediateWriter that sends to the provided channel
+// NewIntermediateWriter create an instance of an IntermediateWriter that sends to the provided channel
 func NewIntermediateWriter(outChan chan<- []byte) *IntermediateWriter {
 	return &IntermediateWriter{
 		outChan: outChan,
@@ -62,7 +62,7 @@ func NewStreamingService(writeDir, filePrefix string, storeKeys []types.StoreKey
 	for _, key := range storeKeys {
 		listeners[key] = append(listeners[key], listener)
 	}
-	// check that the writeDir exists and is writeable so that we can catch the error here at initialization if it is not
+	// check that the writeDir exists and is writable so that we can catch the error here at initialization if it is not
 	// we don't open a dstFile until we receive our first ABCI message
 	if err := isDirWriteable(writeDir); err != nil {
 		return nil, err

store/streaming/file/service_test.go

@@ -372,7 +372,7 @@ func readInFile(name string) ([]byte, error) {
 	return ioutil.ReadFile(path)
 }
 
-// Returns all of the protobuf messages contained in the byte array as an array of byte arrays
+// segmentBytes returns all of the protobuf messages contained in the byte array as an array of byte arrays
 // The messages have their length prefix removed
 func segmentBytes(bz []byte) ([][]byte, error) {
 	var err error
@@ -388,7 +388,7 @@ func segmentBytes(bz []byte) ([][]byte, error) {
 	return segments, nil
 }
 
-// Returns the bytes for the leading protobuf object in the byte array (removing the length prefix) and returns the remainder of the byte array
+// getHeadSegment returns the bytes for the leading protobuf object in the byte array (removing the length prefix) and returns the remainder of the byte array
 func getHeadSegment(bz []byte) ([]byte, []byte, error) {
 	size, prefixSize := binary.Uvarint(bz)
 	if prefixSize < 0 {