Merge branch 'storagebugfix' of cwtch.im/cwtch into master

storage/profile_store.go

@@ -152,7 +152,7 @@ func (ps *profileStore) eventHandler() {
 			groupid := ev.Data[event.GroupID]
 			received, _ := time.Parse(time.RFC3339Nano, ev.Data[event.TimestampReceived])
 			sent, _ := time.Parse(time.RFC3339Nano, ev.Data[event.TimestampSent])
-			message := model.Message{Received: received, Timestamp: sent, Message: ev.Data[event.Data], PeerID: ev.Data[event.RemotePeer], Signature: []byte(ev.Data[event.Signature]), PreviousMessageSig:[]byte(ev.Data[event.PreviousSignature])}
+			message := model.Message{Received: received, Timestamp: sent, Message: ev.Data[event.Data], PeerID: ev.Data[event.RemotePeer], Signature: []byte(ev.Data[event.Signature]), PreviousMessageSig: []byte(ev.Data[event.PreviousSignature])}
 			ps.streamStores[groupid].Write(message)
 		default:
 			return

storage/stream_store.go

@@ -1,7 +1,6 @@
 package storage
 
 import (
-	"bufio"
 	"cwtch.im/cwtch/model"
 	"encoding/json"
 	"fmt"
@@ -53,24 +52,23 @@ func (ss *streamStore) initBuffer() {
 }
 
 func (ss *streamStore) initBufferFromStorage() error {
-	filename := path.Join(ss.storeDirectory, fmt.Sprintf("%s.%d", ss.filenameBase, 0))
+	filename := fmt.Sprintf("%s.%d", ss.filenameBase, 0)
-	f, err := os.OpenFile(filename, os.O_CREATE|os.O_APPEND|os.O_RDWR, 0600)
+
+	bytes, err := readEncryptedFile(ss.storeDirectory, filename, ss.password)
 	if err != nil {
-		log.Errorf("StreamStore could not open: %v: %v", filename, err)
+		log.Debugf("Failed to read encrypted file: %v", err)
+	}
+
+	msgs := []model.Message{}
+	err = json.Unmarshal([]byte(bytes), &msgs)
+	if err != nil {
+		log.Debugf("failed to init buffer from storage: %v", err)
 		return err
 	}
-	defer f.Close()
 
-	scanner := bufio.NewScanner(f)
+	for _, message := range msgs {
-	for scanner.Scan() {
+		ss.updateBuffer(message)
-		mText := scanner.Text()
-		m := model.Message{}
-		err := json.Unmarshal([]byte(mText), &m)
-		if err == nil {
-			ss.updateBuffer(m)
-		}
 	}
-
 	return nil
 }
 
@@ -118,6 +116,7 @@ func (ss *streamStore) Read() (messages []model.Message) {
 
 		bytes, err := readEncryptedFile(ss.storeDirectory, filename, ss.password)
 		if err != nil {
+			log.Debugf("Failed to read encrypted file: %v", err)
 			continue
 		}
 
@@ -125,6 +124,8 @@ func (ss *streamStore) Read() (messages []model.Message) {
 		err = json.Unmarshal([]byte(bytes), &msgs)
 		if err == nil {
 			resp = append(resp, msgs...)
+		} else {
+			log.Debugf("Failed to unmarshal messages: %v", err)
 		}
 	}
 
@@ -139,6 +140,7 @@ func (ss *streamStore) Write(m model.Message) {
 	ss.updateFile()
 
 	if ss.bufferByteCount > bytesPerFile {
+		log.Debugf("rotating log file")
 		ss.rotateFileStore()
 		ss.initBuffer()
 	}

storage/stream_store_test.go

@@ -2,6 +2,7 @@ package storage
 
 import (
 	"cwtch.im/cwtch/model"
+	"git.openprivacy.ca/openprivacy/libricochet-go/log"
 	"os"
 	"testing"
 )
@@ -12,6 +13,10 @@ const password = "asdfqwer"
 const line1 = "Hello from storage!"
 
 func TestStreamStoreWriteRead(t *testing.T) {
+
+	log.SetLevel(log.LevelDebug)
+
+	os.Remove(".test.json")
 	os.RemoveAll(testingDir)
 	os.Mkdir(testingDir, 0777)
 	ss1 := NewStreamStore(testingDir, filenameBase, password)
@@ -27,3 +32,25 @@ func TestStreamStoreWriteRead(t *testing.T) {
 		t.Errorf("Read message has wrong content. Expected: '%v' Actual: '%v'\n", line1, messages[0].Message)
 	}
 }
+
+func TestStreamStoreWriteReadRotate(t *testing.T) {
+
+	log.SetLevel(log.LevelDebug)
+	os.Remove(".test.json")
+	os.RemoveAll(testingDir)
+	os.Mkdir(testingDir, 0777)
+	ss1 := NewStreamStore(testingDir, filenameBase, password)
+	m := model.Message{Message: line1}
+	for i := 0; i < 400; i++ {
+		ss1.Write(m)
+	}
+
+	ss2 := NewStreamStore(testingDir, filenameBase, password)
+	messages := ss2.Read()
+	if len(messages) != 400 {
+		t.Errorf("Read messages has wrong length. Expected: 400 Actual: %d\n", len(messages))
+	}
+	if messages[0].Message != line1 {
+		t.Errorf("Read message has wrong content. Expected: '%v' Actual: '%v'\n", line1, messages[0].Message)
+	}
+}