Merge branch 'main' into dependabot/go_modules/go.mongodb.org/mongo-driver-1.11.6

Author: ackleymi

config/configuration.go

@@ -50,6 +50,7 @@ const (
 	HeartBtIntOverride           string = "HeartBtIntOverride"
 	FileLogPath                  string = "FileLogPath"
 	FileStorePath                string = "FileStorePath"
+	FileStoreSync                string = "FileStoreSync"
 	SQLStoreDriver               string = "SQLStoreDriver"
 	SQLStoreDataSourceName       string = "SQLStoreDataSourceName"
 	SQLStoreConnMaxLifetime      string = "SQLStoreConnMaxLifetime"

field_map.go

@@ -227,6 +227,14 @@ func (m *FieldMap) SetString(tag Tag, value string) *FieldMap {
 	return m.SetBytes(tag, []byte(value))
 }
 
+// Remove removes a tag from field map.
+func (m *FieldMap) Remove(tag Tag) {
+	m.rwLock.Lock()
+	defer m.rwLock.Unlock()
+
+	delete(m.tagLookup, tag)
+}
+
 // Clear purges all fields from field map.
 func (m *FieldMap) Clear() {
 	m.rwLock.Lock()

field_map_test.go

@@ -190,3 +190,15 @@ func TestFieldMap_CopyInto(t *testing.T) {
 	assert.Nil(t, err)
 	assert.Equal(t, "a", s)
 }
+
+func TestFieldMap_Remove(t *testing.T) {
+	var fMap FieldMap
+	fMap.init()
+
+	fMap.SetField(1, FIXString("hello"))
+	fMap.SetField(2, FIXString("world"))
+
+	fMap.Remove(1)
+	assert.False(t, fMap.Has(1))
+	assert.True(t, fMap.Has(2))
+}

filestore.go

@@ -52,6 +52,7 @@ type fileStore struct {
 	sessionFile        *os.File
 	senderSeqNumsFile  *os.File
 	targetSeqNumsFile  *os.File
+	fileSync           bool
 }
 
 // NewFileStoreFactory returns a file-based implementation of MessageStoreFactory.
@@ -77,10 +78,19 @@ func (f fileStoreFactory) Create(sessionID SessionID) (msgStore MessageStore, er
 	if err != nil {
 		return nil, err
 	}
-	return newFileStore(sessionID, dirname)
+	var fsync bool
+	if sessionSettings.HasSetting(config.FileStoreSync) {
+		fsync, err = sessionSettings.BoolSetting(config.FileStoreSync)
+		if err != nil {
+			return nil, err
+		}
+	} else {
+		fsync = true //existing behavior is to fsync writes
+	}
+	return newFileStore(sessionID, dirname, fsync)
 }
 
-func newFileStore(sessionID SessionID, dirname string) (*fileStore, error) {
+func newFileStore(sessionID SessionID, dirname string, fileSync bool) (*fileStore, error) {
 	if err := os.MkdirAll(dirname, os.ModePerm); err != nil {
 		return nil, err
 	}
@@ -96,6 +106,7 @@ func newFileStore(sessionID SessionID, dirname string) (*fileStore, error) {
 		sessionFname:       path.Join(dirname, fmt.Sprintf("%s.%s", sessionPrefix, "session")),
 		senderSeqNumsFname: path.Join(dirname, fmt.Sprintf("%s.%s", sessionPrefix, "senderseqnums")),
 		targetSeqNumsFname: path.Join(dirname, fmt.Sprintf("%s.%s", sessionPrefix, "targetseqnums")),
+		fileSync:           fileSync,
 	}
 
 	if err := store.Refresh(); err != nil {
@@ -232,8 +243,10 @@ func (store *fileStore) setSession() error {
 	if _, err := store.sessionFile.Write(data); err != nil {
 		return fmt.Errorf("unable to write to file: %s: %s", store.sessionFname, err.Error())
 	}
-	if err := store.sessionFile.Sync(); err != nil {
-		return fmt.Errorf("unable to flush file: %s: %s", store.sessionFname, err.Error())
+	if store.fileSync {
+		if err := store.sessionFile.Sync(); err != nil {
+			return fmt.Errorf("unable to flush file: %s: %s", store.sessionFname, err.Error())
+		}
 	}
 	return nil
 }
@@ -245,8 +258,10 @@ func (store *fileStore) setSeqNum(f *os.File, seqNum int) error {
 	if _, err := fmt.Fprintf(f, "%019d", seqNum); err != nil {
 		return fmt.Errorf("unable to write to file: %s: %s", f.Name(), err.Error())
 	}
-	if err := f.Sync(); err != nil {
-		return fmt.Errorf("unable to flush file: %s: %s", f.Name(), err.Error())
+	if store.fileSync {
+		if err := f.Sync(); err != nil {
+			return fmt.Errorf("unable to flush file: %s: %s", f.Name(), err.Error())
+		}
 	}
 	return nil
 }
@@ -313,11 +328,13 @@ func (store *fileStore) SaveMessage(seqNum int, msg []byte) error {
 	if _, err := store.bodyFile.Write(msg); err != nil {
 		return fmt.Errorf("unable to write to file: %s: %s", store.bodyFname, err.Error())
 	}
-	if err := store.bodyFile.Sync(); err != nil {
-		return fmt.Errorf("unable to flush file: %s: %s", store.bodyFname, err.Error())
-	}
-	if err := store.headerFile.Sync(); err != nil {
-		return fmt.Errorf("unable to flush file: %s: %s", store.headerFname, err.Error())
+	if store.fileSync {
+		if err := store.bodyFile.Sync(); err != nil {
+			return fmt.Errorf("unable to flush file: %s: %s", store.bodyFname, err.Error())
+		}
+		if err := store.headerFile.Sync(); err != nil {
+			return fmt.Errorf("unable to flush file: %s: %s", store.headerFname, err.Error())
+		}
 	}
 
 	store.offsets[seqNum] = msgDef{offset: offset, size: len(msg)}

go.mod

@@ -7,7 +7,7 @@ require (
 	github.com/mattn/go-sqlite3 v1.14.16
 	github.com/pkg/errors v0.9.1
 	github.com/shopspring/decimal v1.3.1
-	github.com/stretchr/testify v1.8.1
+	github.com/stretchr/testify v1.8.2
 	go.mongodb.org/mongo-driver v1.11.6
 	golang.org/x/net v0.5.0
 )

go.sum

@@ -37,8 +37,8 @@ github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpE
 github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
-github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
-github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
+github.com/stretchr/testify v1.8.2 h1:+h33VjcLVPDHtOdpUCuF+7gSuG3yGIftsP1YvFihtJ8=
+github.com/stretchr/testify v1.8.2/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
 github.com/tidwall/pretty v1.0.0 h1:HsD+QiTn7sK6flMKIvNmpqz1qrpP3Ps6jOKIKMooyg4=
 github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
 github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c=

message.go

@@ -272,7 +272,6 @@ func ParseMessageWithDataDictionary(
 	}
 
 	return
-
 }
 
 func isHeaderField(tag Tag, dataDict *datadictionary.DataDictionary) bool {
@@ -390,6 +389,14 @@ func extractField(parsedFieldBytes *TagValue, buffer []byte) (remBytes []byte, e
 	return buffer[(endIndex + 1):], err
 }
 
+func (m *Message) Bytes() []byte {
+	if m.rawMessage != nil {
+		return m.rawMessage.Bytes()
+	}
+
+	return m.build()
+}
+
 func (m *Message) String() string {
 	if m.rawMessage != nil {
 		return m.rawMessage.String()

message_test.go

@@ -222,12 +222,14 @@ func (s *MessageSuite) TestCopyIntoMessage() {
 	s.Nil(ParseMessage(s.msg, bytes.NewBufferString(newMsgString)))
 	s.True(s.msg.IsMsgTypeOf("A"))
 	s.Equal(s.msg.String(), newMsgString)
+	s.Equal(string(s.msg.Bytes()), newMsgString)
 
 	// clear the source buffer also
 	msgBuf.Reset()
 
 	s.True(dest.IsMsgTypeOf("D"))
 	s.Equal(dest.String(), renderedString)
+	s.Equal(string(dest.Bytes()), renderedString)
 }
 
 func checkFieldInt(s *MessageSuite, fields FieldMap, tag, expected int) {