Remove sqlite persistence
Currently there seem to be a bug with publishing the contact-code, and each attempt result in the error `database is closed` (non-thread safe?). As discussed previously we probably don't need to have fs persistence, so just changing this to have it in memory.
This commit is contained in:
parent
e85b4509ba
commit
ceab77db19
|
@ -121,7 +121,7 @@ func NewWithEncryptorConfig(
|
|||
ProtocolVersion: protocolVersion,
|
||||
InstallationID: installationID,
|
||||
}),
|
||||
publisher: publisher.New(db, logger),
|
||||
publisher: publisher.New(logger),
|
||||
onAddedBundlesHandler: addedBundlesHandler,
|
||||
onNewSharedSecretHandler: onNewSharedSecretHandler,
|
||||
onSendContactCodeHandler: onSendContactCodeHandler,
|
||||
|
|
|
@ -1,53 +1,37 @@
|
|||
package publisher
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"encoding/hex"
|
||||
"sync"
|
||||
)
|
||||
|
||||
type sqlitePersistence struct {
|
||||
db *sql.DB
|
||||
type persistence struct {
|
||||
lastAcksMutex sync.Mutex
|
||||
lastPublished int64
|
||||
lastAcks map[string]int64
|
||||
}
|
||||
|
||||
func newSQLitePersistence(db *sql.DB) *sqlitePersistence {
|
||||
return &sqlitePersistence{
|
||||
db: db,
|
||||
func newPersistence() *persistence {
|
||||
return &persistence{
|
||||
lastAcks: make(map[string]int64),
|
||||
}
|
||||
}
|
||||
|
||||
func (s *sqlitePersistence) lastPublished() (int64, error) {
|
||||
var lastPublished int64
|
||||
statement := "SELECT last_published FROM contact_code_config LIMIT 1"
|
||||
err := s.db.QueryRow(statement).Scan(&lastPublished)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return lastPublished, nil
|
||||
func (s *persistence) getLastPublished() int64 {
|
||||
return s.lastPublished
|
||||
}
|
||||
|
||||
func (s *sqlitePersistence) setLastPublished(lastPublished int64) error {
|
||||
statement := "UPDATE contact_code_config SET last_published = ?"
|
||||
stmt, err := s.db.Prepare(statement)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer stmt.Close()
|
||||
|
||||
_, err = stmt.Exec(lastPublished)
|
||||
return err
|
||||
func (s *persistence) setLastPublished(lastPublished int64) {
|
||||
s.lastPublished = lastPublished
|
||||
}
|
||||
|
||||
func (s *sqlitePersistence) lastAck(identity []byte) (int64, error) {
|
||||
func (s *persistence) lastAck(identity []byte) int64 {
|
||||
s.lastAcksMutex.Lock()
|
||||
defer s.lastAcksMutex.Unlock()
|
||||
return s.lastAcks[hex.EncodeToString(identity)], nil
|
||||
return s.lastAcks[hex.EncodeToString(identity)]
|
||||
}
|
||||
|
||||
func (s *sqlitePersistence) setLastAck(identity []byte, lastAck int64) {
|
||||
func (s *persistence) setLastAck(identity []byte, lastAck int64) {
|
||||
s.lastAcksMutex.Lock()
|
||||
defer s.lastAcksMutex.Unlock()
|
||||
s.lastAcks[hex.EncodeToString(identity)] = lastAck
|
||||
|
|
|
@ -2,7 +2,6 @@ package publisher
|
|||
|
||||
import (
|
||||
"crypto/ecdsa"
|
||||
"database/sql"
|
||||
"errors"
|
||||
"time"
|
||||
|
||||
|
@ -24,19 +23,19 @@ var (
|
|||
)
|
||||
|
||||
type Publisher struct {
|
||||
persistence *sqlitePersistence
|
||||
persistence *persistence
|
||||
logger *zap.Logger
|
||||
notifyCh chan struct{}
|
||||
quit chan struct{}
|
||||
}
|
||||
|
||||
func New(db *sql.DB, logger *zap.Logger) *Publisher {
|
||||
func New(logger *zap.Logger) *Publisher {
|
||||
if logger == nil {
|
||||
logger = zap.NewNop()
|
||||
}
|
||||
|
||||
return &Publisher{
|
||||
persistence: newSQLitePersistence(db),
|
||||
persistence: newPersistence(),
|
||||
logger: logger.With(zap.Namespace("Publisher")),
|
||||
}
|
||||
}
|
||||
|
@ -93,10 +92,7 @@ func (p *Publisher) tickerLoop() {
|
|||
}
|
||||
|
||||
func (p *Publisher) notify() error {
|
||||
lastPublished, err := p.persistence.lastPublished()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
lastPublished := p.persistence.getLastPublished()
|
||||
|
||||
now := time.Now().Unix()
|
||||
|
||||
|
@ -106,15 +102,13 @@ func (p *Publisher) notify() error {
|
|||
|
||||
p.notifyCh <- struct{}{}
|
||||
|
||||
return p.persistence.setLastPublished(now)
|
||||
p.persistence.setLastPublished(now)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (p *Publisher) ShouldAdvertiseBundle(publicKey *ecdsa.PublicKey, now int64) (bool, error) {
|
||||
identity := crypto.CompressPubkey(publicKey)
|
||||
lastAcked, err := p.persistence.lastAck(identity)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
lastAcked := p.persistence.lastAck(identity)
|
||||
return now-lastAcked < deviceNotFoundAckInterval, nil
|
||||
}
|
||||
|
||||
|
|
|
@ -1,8 +1,6 @@
|
|||
package publisher
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"io/ioutil"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/suite"
|
||||
|
@ -22,14 +20,8 @@ type PublisherTestSuite struct {
|
|||
}
|
||||
|
||||
func (p *PublisherTestSuite) SetupTest(installationID string) {
|
||||
dir, err := ioutil.TempDir("", "publisher-test")
|
||||
p.Require().NoError(err)
|
||||
|
||||
db, err := sql.Open("sqlite3", dir)
|
||||
p.Require().NoError(err)
|
||||
|
||||
p.logger = tt.MustCreateTestLogger()
|
||||
p.publisher = New(db, p.logger)
|
||||
p.publisher = New(p.logger)
|
||||
}
|
||||
|
||||
func (p *PublisherTestSuite) TearDownTest() {
|
||||
|
|
Loading…
Reference in New Issue