fix persist add store transaction

Author: sami-sweng

filestore.go

@@ -301,6 +301,14 @@ func (store *fileStore) SaveMessage(seqNum int, msg []byte) error {
 	return nil
 }
 
+func (store *fileStore) SaveMessageAndIncrNextSenderMsgSeqNum(seqNum int, msg []byte) error {
+	err := store.SaveMessage(seqNum, msg)
+	if err != nil {
+		return err
+	}
+	return store.IncrNextSenderMsgSeqNum()
+}
+
 func (store *fileStore) getMessage(seqNum int) (msg []byte, found bool, err error) {
 	msgInfo, found := store.offsets[seqNum]
 	if !found {

mongostore.go

@@ -248,6 +248,15 @@ func (store *mongoStore) SaveMessage(seqNum int, msg []byte) (err error) {
 	return
 }
 
+func (store *mongoStore) SaveMessageAndIncrNextSenderMsgSeqNum(seqNum int, msg []byte) error {
+	// TODO add transaction
+	err := store.SaveMessage(seqNum, msg)
+	if err != nil {
+		return err
+	}
+	return store.IncrNextSenderMsgSeqNum()
+}
+
 func (store *mongoStore) GetMessages(beginSeqNum, endSeqNum int) (msgs [][]byte, err error) {
 	msgFilter := generateMessageFilter(&store.sessionID)
 	//Marshal into database form

session.go

@@ -321,12 +321,10 @@ func (s *session) prepMessageForSend(msg *Message, inReplyTo *Message) (msgBytes
 
 func (s *session) persist(seqNum int, msgBytes []byte) error {
 	if !s.DisableMessagePersist {
-		if err := s.store.SaveMessage(seqNum, msgBytes); err != nil {
-			return err
-		}
+		return s.store.SaveMessageAndIncrNextSenderMsgSeqNum(seqNum, msgBytes)
+	} else {
+		return s.store.IncrNextSenderMsgSeqNum()
 	}
-
-	return s.store.IncrNextSenderMsgSeqNum()
 }
 
 func (s *session) sendQueued() {

sqlstore.go

@@ -274,6 +274,53 @@ func (store *sqlStore) SaveMessage(seqNum int, msg []byte) error {
 	return err
 }
 
+func (store *sqlStore) SaveMessageAndIncrNextSenderMsgSeqNum(seqNum int, msg []byte) error {
+	s := store.sessionID
+
+	tx, err := store.db.Begin()
+	if err != nil {
+		return err
+	}
+	defer tx.Rollback()
+
+	_, err = tx.Exec(sqlString(`INSERT INTO messages (
+			msgseqnum, message,
+			beginstring, session_qualifier,
+			sendercompid, sendersubid, senderlocid,
+			targetcompid, targetsubid, targetlocid)
+			VALUES(?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, store.placeholder),
+		seqNum, string(msg),
+		s.BeginString, s.Qualifier,
+		s.SenderCompID, s.SenderSubID, s.SenderLocationID,
+		s.TargetCompID, s.TargetSubID, s.TargetLocationID)
+	if err != nil {
+		return err
+	}
+
+	if err := store.cache.IncrNextSenderMsgSeqNum(); err != nil {
+		return errors.Wrap(err, "cache incr next")
+	}
+
+	next := store.cache.NextSenderMsgSeqNum()
+	_, err = store.db.Exec(sqlString(`UPDATE sessions SET outgoing_seqnum = ?
+		WHERE beginstring=? AND session_qualifier=?
+		AND sendercompid=? AND sendersubid=? AND senderlocid=?
+		AND targetcompid=? AND targetsubid=? AND targetlocid=?`, store.placeholder),
+		next, s.BeginString, s.Qualifier,
+		s.SenderCompID, s.SenderSubID, s.SenderLocationID,
+		s.TargetCompID, s.TargetSubID, s.TargetLocationID)
+	if err != nil {
+		return err
+	}
+
+	err = store.cache.SetNextSenderMsgSeqNum(next)
+	if err != nil {
+		return err
+	}
+
+	return tx.Commit()
+}
+
 func (store *sqlStore) GetMessages(beginSeqNum, endSeqNum int) ([][]byte, error) {
 	s := store.sessionID
 	var msgs [][]byte

store.go

@@ -20,6 +20,7 @@ type MessageStore interface {
 	CreationTime() time.Time
 
 	SaveMessage(seqNum int, msg []byte) error
+	SaveMessageAndIncrNextSenderMsgSeqNum(seqNum int, msg []byte) error
 	GetMessages(beginSeqNum, endSeqNum int) ([][]byte, error)
 
 	Refresh() error
@@ -97,6 +98,14 @@ func (store *memoryStore) SaveMessage(seqNum int, msg []byte) error {
 	return nil
 }
 
+func (store *memoryStore) SaveMessageAndIncrNextSenderMsgSeqNum(seqNum int, msg []byte) error {
+	err := store.SaveMessage(seqNum, msg)
+	if err != nil {
+		return err
+	}
+	return store.IncrNextSenderMsgSeqNum()
+}
+
 func (store *memoryStore) GetMessages(beginSeqNum, endSeqNum int) ([][]byte, error) {
 	var msgs [][]byte
 	for seqNum := beginSeqNum; seqNum <= endSeqNum; seqNum++ {