cwtch/storage/message_store.go

package storage

import (
	"bufio"
	"cwtch.im/cwtch/protocol"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"sync"
)

// MessageStoreInterface defines an interface to interact with a store of cwtch messages.
type MessageStoreInterface interface {
	AddMessage(protocol.GroupMessage)
	FetchMessages() []*protocol.GroupMessage
}

// MessageStore is a file-backed implementation of MessageStoreInterface
type MessageStore struct {
	file     *os.File
	lock     sync.Mutex
	messages []*protocol.GroupMessage
}

// Close closes the message store and underlying resources.
func (ms *MessageStore) Close() {
	ms.lock.Lock()
	ms.messages = nil
	ms.file.Close()
	ms.lock.Unlock()
}

// Init sets up a MessageStore backed by filename
func (ms *MessageStore) Init(filename string) {
	f, err := os.OpenFile(filename, os.O_CREATE|os.O_APPEND|os.O_RDWR, 0600)
	if err != nil {
		panic(err)
	}
	ms.file = f

	scanner := bufio.NewScanner(f)
	for scanner.Scan() {
		gms := scanner.Text()
		gm := &protocol.GroupMessage{}
		err := json.Unmarshal([]byte(gms), gm)
		if err == nil {
			ms.messages = append(ms.messages, gm)
		} else {
			panic(err)
		}
	}

	if err := scanner.Err(); err != nil {
		panic(err)
	}

}

// FetchMessages returns all messages from the backing file.
func (ms *MessageStore) FetchMessages() (messages []*protocol.GroupMessage) {
	messages = make([]*protocol.GroupMessage, len(ms.messages))
	ms.lock.Lock()
	copy(messages, ms.messages)
	ms.lock.Unlock()
	return
}

// AddMessage adds a GroupMessage to the store
func (ms *MessageStore) AddMessage(gm protocol.GroupMessage) {
	ms.lock.Lock()
	ms.messages = append(ms.messages, &gm)
	s, err := json.Marshal(gm)
	if err != nil {
		log.Printf("[ERROR] Failed to unmarshal group message %v\n", err)
	}
	fmt.Fprintf(ms.file, "%s\n", s)
	ms.lock.Unlock()
}