Rename TransactionWriters to Writers

This commit is contained in:
Neil Alexander 2020-08-21 09:31:33 +01:00
parent c97f3d3f83
commit bc1023ea19
No known key found for this signature in database
GPG key ID: A02A2019A2BB0944
49 changed files with 103 additions and 103 deletions

View file

@ -32,7 +32,7 @@ type Database struct {
events eventsStatements
txnID txnStatements
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
}
// NewDatabase opens a new database
@ -42,7 +42,7 @@ func NewDatabase(dbProperties *config.DatabaseOptions) (*Database, error) {
if result.db, err = sqlutil.Open(dbProperties); err != nil {
return nil, err
}
result.writer = sqlutil.NewDummyTransactionWriter()
result.writer = sqlutil.NewDummyWriter()
if err = result.prepare(); err != nil {
return nil, err
}

View file

@ -67,7 +67,7 @@ const (
type eventsStatements struct {
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
selectEventsByApplicationServiceIDStmt *sql.Stmt
countEventsByApplicationServiceIDStmt *sql.Stmt
insertEventStmt *sql.Stmt
@ -75,7 +75,7 @@ type eventsStatements struct {
deleteEventsBeforeAndIncludingIDStmt *sql.Stmt
}
func (s *eventsStatements) prepare(db *sql.DB, writer sqlutil.TransactionWriter) (err error) {
func (s *eventsStatements) prepare(db *sql.DB, writer sqlutil.Writer) (err error) {
s.db = db
s.writer = writer
_, err = db.Exec(appserviceEventsSchema)

View file

@ -32,7 +32,7 @@ type Database struct {
events eventsStatements
txnID txnStatements
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
}
// NewDatabase opens a new database
@ -42,7 +42,7 @@ func NewDatabase(dbProperties *config.DatabaseOptions) (*Database, error) {
if result.db, err = sqlutil.Open(dbProperties); err != nil {
return nil, err
}
result.writer = sqlutil.NewTransactionWriter()
result.writer = sqlutil.NewExclusiveWriter()
if err = result.prepare(); err != nil {
return nil, err
}

View file

@ -38,11 +38,11 @@ const selectTxnIDSQL = `
type txnStatements struct {
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
selectTxnIDStmt *sql.Stmt
}
func (s *txnStatements) prepare(db *sql.DB, writer sqlutil.TransactionWriter) (err error) {
func (s *txnStatements) prepare(db *sql.DB, writer sqlutil.Writer) (err error) {
s.db = db
s.writer = writer
_, err = db.Exec(txnIDSchema)

View file

@ -11,7 +11,7 @@ import (
type Database struct {
shared.Database
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
sqlutil.PartitionOffsetStatements
}
@ -22,7 +22,7 @@ func NewDatabase(dbProperties *config.DatabaseOptions) (*Database, error) {
if d.db, err = sqlutil.Open(dbProperties); err != nil {
return nil, err
}
d.writer = sqlutil.NewDummyTransactionWriter()
d.writer = sqlutil.NewDummyWriter()
if err = d.PartitionOffsetStatements.Prepare(d.db, d.writer, "currentstate"); err != nil {
return nil, err
}

View file

@ -27,7 +27,7 @@ import (
type Database struct {
DB *sql.DB
Writer sqlutil.TransactionWriter
Writer sqlutil.Writer
CurrentRoomState tables.CurrentRoomState
}

View file

@ -83,7 +83,7 @@ const selectKnownUsersSQL = "" +
type currentRoomStateStatements struct {
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
upsertRoomStateStmt *sql.Stmt
deleteRoomStateByEventIDStmt *sql.Stmt
selectRoomIDsWithMembershipStmt *sql.Stmt
@ -96,7 +96,7 @@ type currentRoomStateStatements struct {
func NewSqliteCurrentRoomStateTable(db *sql.DB) (tables.CurrentRoomState, error) {
s := &currentRoomStateStatements{
db: db,
writer: sqlutil.NewTransactionWriter(),
writer: sqlutil.NewExclusiveWriter(),
}
_, err := db.Exec(currentRoomStateSchema)
if err != nil {

View file

@ -11,7 +11,7 @@ import (
type Database struct {
shared.Database
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
sqlutil.PartitionOffsetStatements
}
@ -23,7 +23,7 @@ func NewDatabase(dbProperties *config.DatabaseOptions) (*Database, error) {
if d.db, err = sqlutil.Open(dbProperties); err != nil {
return nil, err
}
d.writer = sqlutil.NewTransactionWriter()
d.writer = sqlutil.NewExclusiveWriter()
if err = d.PartitionOffsetStatements.Prepare(d.db, d.writer, "currentstate"); err != nil {
return nil, err
}

View file

@ -28,7 +28,7 @@ type Database struct {
shared.Database
sqlutil.PartitionOffsetStatements
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
}
// NewDatabase opens a new database
@ -38,7 +38,7 @@ func NewDatabase(dbProperties *config.DatabaseOptions) (*Database, error) {
if d.db, err = sqlutil.Open(dbProperties); err != nil {
return nil, err
}
d.writer = sqlutil.NewDummyTransactionWriter()
d.writer = sqlutil.NewDummyWriter()
joinedHosts, err := NewPostgresJoinedHostsTable(d.db)
if err != nil {
return nil, err

View file

@ -28,7 +28,7 @@ import (
type Database struct {
DB *sql.DB
Writer sqlutil.TransactionWriter
Writer sqlutil.Writer
FederationSenderQueuePDUs tables.FederationSenderQueuePDUs
FederationSenderQueueEDUs tables.FederationSenderQueueEDUs
FederationSenderQueueJSON tables.FederationSenderQueueJSON

View file

@ -30,7 +30,7 @@ type Database struct {
shared.Database
sqlutil.PartitionOffsetStatements
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
}
// NewDatabase opens a new database
@ -40,7 +40,7 @@ func NewDatabase(dbProperties *config.DatabaseOptions) (*Database, error) {
if d.db, err = sqlutil.Open(dbProperties); err != nil {
return nil, err
}
d.writer = sqlutil.NewTransactionWriter()
d.writer = sqlutil.NewExclusiveWriter()
joinedHosts, err := NewSQLiteJoinedHostsTable(d.db)
if err != nil {
return nil, err

View file

@ -54,7 +54,7 @@ const upsertPartitionOffsetsSQL = "" +
// PartitionOffsetStatements represents a set of statements that can be run on a partition_offsets table.
type PartitionOffsetStatements struct {
db *sql.DB
writer TransactionWriter
writer Writer
selectPartitionOffsetsStmt *sql.Stmt
upsertPartitionOffsetStmt *sql.Stmt
}
@ -62,7 +62,7 @@ type PartitionOffsetStatements struct {
// Prepare converts the raw SQL statements into prepared statements.
// Takes a prefix to prepend to the table name used to store the partition offsets.
// This allows multiple components to share the same database schema.
func (s *PartitionOffsetStatements) Prepare(db *sql.DB, writer TransactionWriter, prefix string) (err error) {
func (s *PartitionOffsetStatements) Prepare(db *sql.DB, writer Writer, prefix string) (err error) {
s.db = db
s.writer = writer
_, err = db.Exec(strings.Replace(partitionOffsetsSchema, "${prefix}", prefix, -1))

View file

@ -104,6 +104,6 @@ func SQLiteDriverName() string {
return "sqlite3"
}
type TransactionWriter interface {
type Writer interface {
Do(db *sql.DB, txn *sql.Tx, f func(txn *sql.Tx) error) error
}

View file

@ -4,15 +4,15 @@ import (
"database/sql"
)
type DummyTransactionWriter struct {
type DummyWriter struct {
}
func NewDummyTransactionWriter() TransactionWriter {
return &DummyTransactionWriter{}
func NewDummyWriter() Writer {
return &DummyWriter{}
}
func (w *DummyTransactionWriter) Do(db *sql.DB, txn *sql.Tx, f func(txn *sql.Tx) error) error {
if txn == nil {
func (w *DummyWriter) Do(db *sql.DB, txn *sql.Tx, f func(txn *sql.Tx) error) error {
if db != nil && txn == nil {
return WithTransaction(db, func(txn *sql.Tx) error {
return f(txn)
})

View file

@ -7,16 +7,16 @@ import (
"go.uber.org/atomic"
)
// ExclusiveTransactionWriter allows queuing database writes so that you don't
// ExclusiveWriter allows queuing database writes so that you don't
// contend on database locks in, e.g. SQLite. Only one task will run
// at a time on a given ExclusiveTransactionWriter.
type ExclusiveTransactionWriter struct {
// at a time on a given ExclusiveWriter.
type ExclusiveWriter struct {
running atomic.Bool
todo chan transactionWriterTask
}
func NewTransactionWriter() TransactionWriter {
return &ExclusiveTransactionWriter{
func NewExclusiveWriter() Writer {
return &ExclusiveWriter{
todo: make(chan transactionWriterTask),
}
}
@ -34,7 +34,7 @@ type transactionWriterTask struct {
// txn parameter if one is supplied, and if not, will take out a
// new transaction from the database supplied in the database
// parameter. Either way, this will block until the task is done.
func (w *ExclusiveTransactionWriter) Do(db *sql.DB, txn *sql.Tx, f func(txn *sql.Tx) error) error {
func (w *ExclusiveWriter) Do(db *sql.DB, txn *sql.Tx, f func(txn *sql.Tx) error) error {
if w.todo == nil {
return errors.New("not initialised")
}
@ -55,15 +55,15 @@ func (w *ExclusiveTransactionWriter) Do(db *sql.DB, txn *sql.Tx, f func(txn *sql
// of these goroutines will run at a time. A transaction will be
// opened using the database object from the task and then this will
// be passed as a parameter to the task function.
func (w *ExclusiveTransactionWriter) run() {
func (w *ExclusiveWriter) run() {
if !w.running.CAS(false, true) {
return
}
defer w.running.Store(false)
for task := range w.todo {
if task.txn != nil {
if task.db != nil && task.txn != nil {
task.wait <- task.f(task.txn)
} else if task.db != nil {
} else if task.db != nil && task.txn == nil {
task.wait <- WithTransaction(task.db, func(txn *sql.Tx) error {
return task.f(txn)
})

View file

@ -63,7 +63,7 @@ const deleteAllDeviceKeysSQL = "" +
type deviceKeysStatements struct {
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
upsertDeviceKeysStmt *sql.Stmt
selectDeviceKeysStmt *sql.Stmt
selectBatchDeviceKeysStmt *sql.Stmt
@ -71,7 +71,7 @@ type deviceKeysStatements struct {
deleteAllDeviceKeysStmt *sql.Stmt
}
func NewSqliteDeviceKeysTable(db *sql.DB, writer sqlutil.TransactionWriter) (tables.DeviceKeys, error) {
func NewSqliteDeviceKeysTable(db *sql.DB, writer sqlutil.Writer) (tables.DeviceKeys, error) {
s := &deviceKeysStatements{
db: db,
writer: writer,

View file

@ -52,12 +52,12 @@ const selectKeyChangesSQL = "" +
type keyChangesStatements struct {
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
upsertKeyChangeStmt *sql.Stmt
selectKeyChangesStmt *sql.Stmt
}
func NewSqliteKeyChangesTable(db *sql.DB, writer sqlutil.TransactionWriter) (tables.KeyChanges, error) {
func NewSqliteKeyChangesTable(db *sql.DB, writer sqlutil.Writer) (tables.KeyChanges, error) {
s := &keyChangesStatements{
db: db,
writer: writer,

View file

@ -60,7 +60,7 @@ const selectKeyByAlgorithmSQL = "" +
type oneTimeKeysStatements struct {
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
upsertKeysStmt *sql.Stmt
selectKeysStmt *sql.Stmt
selectKeysCountStmt *sql.Stmt
@ -68,7 +68,7 @@ type oneTimeKeysStatements struct {
deleteOneTimeKeyStmt *sql.Stmt
}
func NewSqliteOneTimeKeysTable(db *sql.DB, writer sqlutil.TransactionWriter) (tables.OneTimeKeys, error) {
func NewSqliteOneTimeKeysTable(db *sql.DB, writer sqlutil.Writer) (tables.OneTimeKeys, error) {
s := &oneTimeKeysStatements{
db: db,
writer: writer,

View file

@ -51,13 +51,13 @@ const selectStaleDeviceListsSQL = "" +
type staleDeviceListsStatements struct {
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
upsertStaleDeviceListStmt *sql.Stmt
selectStaleDeviceListsWithDomainsStmt *sql.Stmt
selectStaleDeviceListsStmt *sql.Stmt
}
func NewSqliteStaleDeviceListsTable(db *sql.DB, writer sqlutil.TransactionWriter) (tables.StaleDeviceLists, error) {
func NewSqliteStaleDeviceListsTable(db *sql.DB, writer sqlutil.Writer) (tables.StaleDeviceLists, error) {
s := &staleDeviceListsStatements{
db: db,
writer: writer,

View file

@ -25,7 +25,7 @@ func NewDatabase(dbProperties *config.DatabaseOptions) (*shared.Database, error)
if err != nil {
return nil, err
}
writer := sqlutil.NewTransactionWriter()
writer := sqlutil.NewExclusiveWriter()
otk, err := NewSqliteOneTimeKeysTable(db, writer)
if err != nil {
return nil, err

View file

@ -62,12 +62,12 @@ SELECT content_type, file_size_bytes, creation_ts, upload_name, base64hash, user
type mediaStatements struct {
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
insertMediaStmt *sql.Stmt
selectMediaStmt *sql.Stmt
}
func (s *mediaStatements) prepare(db *sql.DB, writer sqlutil.TransactionWriter) (err error) {
func (s *mediaStatements) prepare(db *sql.DB, writer sqlutil.Writer) (err error) {
s.db = db
s.writer = writer

View file

@ -26,7 +26,7 @@ type statements struct {
thumbnail thumbnailStatements
}
func (s *statements) prepare(db *sql.DB, writer sqlutil.TransactionWriter) (err error) {
func (s *statements) prepare(db *sql.DB, writer sqlutil.Writer) (err error) {
if err = s.media.prepare(db, writer); err != nil {
return
}

View file

@ -31,13 +31,13 @@ import (
type Database struct {
statements statements
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
}
// Open opens a postgres database.
func Open(dbProperties *config.DatabaseOptions) (*Database, error) {
d := Database{
writer: sqlutil.NewTransactionWriter(),
writer: sqlutil.NewExclusiveWriter(),
}
var err error
if d.db, err = sqlutil.Open(dbProperties); err != nil {

View file

@ -59,13 +59,13 @@ SELECT content_type, file_size_bytes, creation_ts, width, height, resize_method
type thumbnailStatements struct {
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
insertThumbnailStmt *sql.Stmt
selectThumbnailStmt *sql.Stmt
selectThumbnailsStmt *sql.Stmt
}
func (s *thumbnailStatements) prepare(db *sql.DB, writer sqlutil.TransactionWriter) (err error) {
func (s *thumbnailStatements) prepare(db *sql.DB, writer sqlutil.Writer) (err error) {
_, err = db.Exec(thumbnailSchema)
if err != nil {
return

View file

@ -98,7 +98,7 @@ func Open(dbProperties *config.DatabaseOptions) (*Database, error) {
}
d.Database = shared.Database{
DB: db,
Writer: sqlutil.NewDummyTransactionWriter(),
Writer: sqlutil.NewDummyWriter(),
EventTypesTable: eventTypes,
EventStateKeysTable: eventStateKeys,
EventJSONTable: eventJSON,

View file

@ -27,7 +27,7 @@ const redactionsArePermanent = false
type Database struct {
DB *sql.DB
Writer sqlutil.TransactionWriter
Writer sqlutil.Writer
EventsTable tables.Events
EventJSONTable tables.EventJSON
EventTypesTable tables.EventTypes

View file

@ -41,7 +41,7 @@ type Database struct {
invites tables.Invites
membership tables.Membership
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
}
// Open a sqlite database.
@ -52,7 +52,7 @@ func Open(dbProperties *config.DatabaseOptions) (*Database, error) {
if d.db, err = sqlutil.Open(dbProperties); err != nil {
return nil, err
}
d.writer = sqlutil.NewTransactionWriter()
d.writer = sqlutil.NewExclusiveWriter()
//d.db.Exec("PRAGMA journal_mode=WAL;")
//d.db.Exec("PRAGMA read_uncommitted = true;")
@ -120,7 +120,7 @@ func Open(dbProperties *config.DatabaseOptions) (*Database, error) {
}
d.Database = shared.Database{
DB: d.db,
Writer: sqlutil.NewTransactionWriter(),
Writer: sqlutil.NewExclusiveWriter(),
EventsTable: d.events,
EventTypesTable: d.eventTypes,
EventStateKeysTable: d.eventStateKeys,

View file

@ -30,7 +30,7 @@ import (
// A Database implements gomatrixserverlib.KeyDatabase and is used to store
// the public keys for other matrix servers.
type Database struct {
writer sqlutil.TransactionWriter
writer sqlutil.Writer
statements serverKeyStatements
}
@ -49,7 +49,7 @@ func NewDatabase(
return nil, err
}
d := &Database{
writer: sqlutil.NewTransactionWriter(),
writer: sqlutil.NewExclusiveWriter(),
}
err = d.statements.prepare(db, d.writer)
if err != nil {

View file

@ -63,12 +63,12 @@ const upsertServerKeysSQL = "" +
type serverKeyStatements struct {
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
bulkSelectServerKeysStmt *sql.Stmt
upsertServerKeysStmt *sql.Stmt
}
func (s *serverKeyStatements) prepare(db *sql.DB, writer sqlutil.TransactionWriter) (err error) {
func (s *serverKeyStatements) prepare(db *sql.DB, writer sqlutil.Writer) (err error) {
s.db = db
s.writer = writer
_, err = db.Exec(serverKeysSchema)

View file

@ -31,7 +31,7 @@ import (
type SyncServerDatasource struct {
shared.Database
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
sqlutil.PartitionOffsetStatements
}
@ -42,7 +42,7 @@ func NewDatabase(dbProperties *config.DatabaseOptions) (*SyncServerDatasource, e
if d.db, err = sqlutil.Open(dbProperties); err != nil {
return nil, err
}
d.writer = sqlutil.NewDummyTransactionWriter()
d.writer = sqlutil.NewDummyWriter()
if err = d.PartitionOffsetStatements.Prepare(d.db, d.writer, "syncapi"); err != nil {
return nil, err
}
@ -88,7 +88,7 @@ func NewDatabase(dbProperties *config.DatabaseOptions) (*SyncServerDatasource, e
BackwardExtremities: backwardExtremities,
Filter: filter,
SendToDevice: sendToDevice,
SendToDeviceWriter: sqlutil.NewTransactionWriter(),
SendToDeviceWriter: sqlutil.NewExclusiveWriter(),
EDUCache: cache.New(),
}
return &d, nil

View file

@ -45,7 +45,7 @@ type Database struct {
BackwardExtremities tables.BackwardsExtremities
SendToDevice tables.SendToDevice
Filter tables.Filter
SendToDeviceWriter sqlutil.TransactionWriter
SendToDeviceWriter sqlutil.Writer
EDUCache *cache.EDUCache
}

View file

@ -51,14 +51,14 @@ const selectMaxAccountDataIDSQL = "" +
type accountDataStatements struct {
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
streamIDStatements *streamIDStatements
insertAccountDataStmt *sql.Stmt
selectMaxAccountDataIDStmt *sql.Stmt
selectAccountDataInRangeStmt *sql.Stmt
}
func NewSqliteAccountDataTable(db *sql.DB, writer sqlutil.TransactionWriter, streamID *streamIDStatements) (tables.AccountData, error) {
func NewSqliteAccountDataTable(db *sql.DB, writer sqlutil.Writer, streamID *streamIDStatements) (tables.AccountData, error) {
s := &accountDataStatements{
db: db,
writer: writer,

View file

@ -49,13 +49,13 @@ const deleteBackwardExtremitySQL = "" +
type backwardExtremitiesStatements struct {
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
insertBackwardExtremityStmt *sql.Stmt
selectBackwardExtremitiesForRoomStmt *sql.Stmt
deleteBackwardExtremityStmt *sql.Stmt
}
func NewSqliteBackwardsExtremitiesTable(db *sql.DB, writer sqlutil.TransactionWriter) (tables.BackwardsExtremities, error) {
func NewSqliteBackwardsExtremitiesTable(db *sql.DB, writer sqlutil.Writer) (tables.BackwardsExtremities, error) {
s := &backwardExtremitiesStatements{
db: db,
writer: writer,

View file

@ -85,7 +85,7 @@ const selectEventsWithEventIDsSQL = "" +
type currentRoomStateStatements struct {
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
streamIDStatements *streamIDStatements
upsertRoomStateStmt *sql.Stmt
deleteRoomStateByEventIDStmt *sql.Stmt
@ -95,7 +95,7 @@ type currentRoomStateStatements struct {
selectStateEventStmt *sql.Stmt
}
func NewSqliteCurrentRoomStateTable(db *sql.DB, writer sqlutil.TransactionWriter, streamID *streamIDStatements) (tables.CurrentRoomState, error) {
func NewSqliteCurrentRoomStateTable(db *sql.DB, writer sqlutil.Writer, streamID *streamIDStatements) (tables.CurrentRoomState, error) {
s := &currentRoomStateStatements{
db: db,
writer: writer,

View file

@ -52,13 +52,13 @@ const insertFilterSQL = "" +
type filterStatements struct {
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
selectFilterStmt *sql.Stmt
selectFilterIDByContentStmt *sql.Stmt
insertFilterStmt *sql.Stmt
}
func NewSqliteFilterTable(db *sql.DB, writer sqlutil.TransactionWriter) (tables.Filter, error) {
func NewSqliteFilterTable(db *sql.DB, writer sqlutil.Writer) (tables.Filter, error) {
_, err := db.Exec(filterSchema)
if err != nil {
return nil, err

View file

@ -59,7 +59,7 @@ const selectMaxInviteIDSQL = "" +
type inviteEventsStatements struct {
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
streamIDStatements *streamIDStatements
insertInviteEventStmt *sql.Stmt
selectInviteEventsInRangeStmt *sql.Stmt
@ -67,7 +67,7 @@ type inviteEventsStatements struct {
selectMaxInviteIDStmt *sql.Stmt
}
func NewSqliteInvitesTable(db *sql.DB, writer sqlutil.TransactionWriter, streamID *streamIDStatements) (tables.Invites, error) {
func NewSqliteInvitesTable(db *sql.DB, writer sqlutil.Writer, streamID *streamIDStatements) (tables.Invites, error) {
s := &inviteEventsStatements{
db: db,
writer: writer,

View file

@ -105,7 +105,7 @@ const selectStateInRangeSQL = "" +
type outputRoomEventsStatements struct {
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
streamIDStatements *streamIDStatements
insertEventStmt *sql.Stmt
selectEventsStmt *sql.Stmt
@ -117,7 +117,7 @@ type outputRoomEventsStatements struct {
updateEventJSONStmt *sql.Stmt
}
func NewSqliteEventsTable(db *sql.DB, writer sqlutil.TransactionWriter, streamID *streamIDStatements) (tables.Events, error) {
func NewSqliteEventsTable(db *sql.DB, writer sqlutil.Writer, streamID *streamIDStatements) (tables.Events, error) {
s := &outputRoomEventsStatements{
db: db,
writer: writer,

View file

@ -67,7 +67,7 @@ const selectMaxPositionInTopologySQL = "" +
type outputRoomEventsTopologyStatements struct {
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
insertEventInTopologyStmt *sql.Stmt
selectEventIDsInRangeASCStmt *sql.Stmt
selectEventIDsInRangeDESCStmt *sql.Stmt
@ -75,7 +75,7 @@ type outputRoomEventsTopologyStatements struct {
selectMaxPositionInTopologyStmt *sql.Stmt
}
func NewSqliteTopologyTable(db *sql.DB, writer sqlutil.TransactionWriter) (tables.Topology, error) {
func NewSqliteTopologyTable(db *sql.DB, writer sqlutil.Writer) (tables.Topology, error) {
s := &outputRoomEventsTopologyStatements{
db: db,
writer: writer,

View file

@ -73,13 +73,13 @@ const deleteSendToDeviceMessagesSQL = `
type sendToDeviceStatements struct {
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
insertSendToDeviceMessageStmt *sql.Stmt
selectSendToDeviceMessagesStmt *sql.Stmt
countSendToDeviceMessagesStmt *sql.Stmt
}
func NewSqliteSendToDeviceTable(db *sql.DB, writer sqlutil.TransactionWriter) (tables.SendToDevice, error) {
func NewSqliteSendToDeviceTable(db *sql.DB, writer sqlutil.Writer) (tables.SendToDevice, error) {
s := &sendToDeviceStatements{
db: db,
writer: writer,

View file

@ -28,12 +28,12 @@ const selectStreamIDStmt = "" +
type streamIDStatements struct {
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
increaseStreamIDStmt *sql.Stmt
selectStreamIDStmt *sql.Stmt
}
func (s *streamIDStatements) prepare(db *sql.DB, writer sqlutil.TransactionWriter) (err error) {
func (s *streamIDStatements) prepare(db *sql.DB, writer sqlutil.Writer) (err error) {
s.db = db
s.writer = writer
_, err = db.Exec(streamIDTableSchema)

View file

@ -32,7 +32,7 @@ import (
type SyncServerDatasource struct {
shared.Database
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
sqlutil.PartitionOffsetStatements
streamID streamIDStatements
}
@ -45,7 +45,7 @@ func NewDatabase(dbProperties *config.DatabaseOptions) (*SyncServerDatasource, e
if d.db, err = sqlutil.Open(dbProperties); err != nil {
return nil, err
}
d.writer = sqlutil.NewTransactionWriter()
d.writer = sqlutil.NewExclusiveWriter()
if err = d.prepare(); err != nil {
return nil, err
}
@ -101,7 +101,7 @@ func (d *SyncServerDatasource) prepare() (err error) {
Topology: topology,
Filter: filter,
SendToDevice: sendToDevice,
SendToDeviceWriter: sqlutil.NewTransactionWriter(),
SendToDeviceWriter: sqlutil.NewExclusiveWriter(),
EDUCache: cache.New(),
}
return nil

View file

@ -35,7 +35,7 @@ import (
// Database represents an account database
type Database struct {
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
sqlutil.PartitionOffsetStatements
accounts accountsStatements
profiles profilesStatements
@ -53,7 +53,7 @@ func NewDatabase(dbProperties *config.DatabaseOptions, serverName gomatrixserver
d := &Database{
serverName: serverName,
db: db,
writer: sqlutil.NewDummyTransactionWriter(),
writer: sqlutil.NewDummyWriter(),
}
if err = d.PartitionOffsetStatements.Prepare(db, d.writer, "account"); err != nil {
return nil, err

View file

@ -51,13 +51,13 @@ const selectAccountDataByTypeSQL = "" +
type accountDataStatements struct {
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
insertAccountDataStmt *sql.Stmt
selectAccountDataStmt *sql.Stmt
selectAccountDataByTypeStmt *sql.Stmt
}
func (s *accountDataStatements) prepare(db *sql.DB, writer sqlutil.TransactionWriter) (err error) {
func (s *accountDataStatements) prepare(db *sql.DB, writer sqlutil.Writer) (err error) {
s.db = db
s.writer = writer
_, err = db.Exec(accountDataSchema)

View file

@ -59,7 +59,7 @@ const selectNewNumericLocalpartSQL = "" +
type accountsStatements struct {
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
insertAccountStmt *sql.Stmt
selectAccountByLocalpartStmt *sql.Stmt
selectPasswordHashStmt *sql.Stmt
@ -67,7 +67,7 @@ type accountsStatements struct {
serverName gomatrixserverlib.ServerName
}
func (s *accountsStatements) prepare(db *sql.DB, writer sqlutil.TransactionWriter, server gomatrixserverlib.ServerName) (err error) {
func (s *accountsStatements) prepare(db *sql.DB, writer sqlutil.Writer, server gomatrixserverlib.ServerName) (err error) {
s.db = db
s.writer = writer
_, err = db.Exec(accountsSchema)

View file

@ -53,7 +53,7 @@ const selectProfilesBySearchSQL = "" +
type profilesStatements struct {
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
insertProfileStmt *sql.Stmt
selectProfileByLocalpartStmt *sql.Stmt
setAvatarURLStmt *sql.Stmt
@ -61,7 +61,7 @@ type profilesStatements struct {
selectProfilesBySearchStmt *sql.Stmt
}
func (s *profilesStatements) prepare(db *sql.DB, writer sqlutil.TransactionWriter) (err error) {
func (s *profilesStatements) prepare(db *sql.DB, writer sqlutil.Writer) (err error) {
s.db = db
s.writer = writer
_, err = db.Exec(profilesSchema)

View file

@ -34,7 +34,7 @@ import (
// Database represents an account database
type Database struct {
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
sqlutil.PartitionOffsetStatements
accounts accountsStatements
@ -58,7 +58,7 @@ func NewDatabase(dbProperties *config.DatabaseOptions, serverName gomatrixserver
d := &Database{
serverName: serverName,
db: db,
writer: sqlutil.NewTransactionWriter(),
writer: sqlutil.NewExclusiveWriter(),
}
partitions := sqlutil.PartitionOffsetStatements{}
if err = partitions.Prepare(db, d.writer, "account"); err != nil {

View file

@ -54,14 +54,14 @@ const deleteThreePIDSQL = "" +
type threepidStatements struct {
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
selectLocalpartForThreePIDStmt *sql.Stmt
selectThreePIDsForLocalpartStmt *sql.Stmt
insertThreePIDStmt *sql.Stmt
deleteThreePIDStmt *sql.Stmt
}
func (s *threepidStatements) prepare(db *sql.DB, writer sqlutil.TransactionWriter) (err error) {
func (s *threepidStatements) prepare(db *sql.DB, writer sqlutil.Writer) (err error) {
s.db = db
s.writer = writer
_, err = db.Exec(threepidSchema)

View file

@ -78,7 +78,7 @@ const selectDevicesByIDSQL = "" +
type devicesStatements struct {
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
insertDeviceStmt *sql.Stmt
selectDevicesCountStmt *sql.Stmt
selectDeviceByTokenStmt *sql.Stmt
@ -91,7 +91,7 @@ type devicesStatements struct {
serverName gomatrixserverlib.ServerName
}
func (s *devicesStatements) prepare(db *sql.DB, writer sqlutil.TransactionWriter, server gomatrixserverlib.ServerName) (err error) {
func (s *devicesStatements) prepare(db *sql.DB, writer sqlutil.Writer, server gomatrixserverlib.ServerName) (err error) {
s.db = db
s.writer = writer
_, err = db.Exec(devicesSchema)

View file

@ -34,7 +34,7 @@ var deviceIDByteLength = 6
// Database represents a device database.
type Database struct {
db *sql.DB
writer sqlutil.TransactionWriter
writer sqlutil.Writer
devices devicesStatements
}
@ -44,7 +44,7 @@ func NewDatabase(dbProperties *config.DatabaseOptions, serverName gomatrixserver
if err != nil {
return nil, err
}
writer := sqlutil.NewTransactionWriter()
writer := sqlutil.NewExclusiveWriter()
d := devicesStatements{}
if err = d.prepare(db, writer, serverName); err != nil {
return nil, err