Constructor for TransactionWriter

internal/sql.go

@@ -18,6 +18,7 @@ package internal
 
 import (
 	"database/sql"
+	"errors"
 	"fmt"
 	"runtime"
 	"time"
@@ -120,6 +121,12 @@ type TransactionWriter struct {
 	todo    chan transactionWriterTask
 }
 
+func NewTransactionWriter() *TransactionWriter {
+	return &TransactionWriter{
+		todo: make(chan transactionWriterTask),
+	}
+}
+
 // transactionWriterTask represents a specific task.
 type transactionWriterTask struct {
 	db   *sql.DB
@@ -132,7 +139,7 @@ type transactionWriterTask struct {
 // database parameter. This will block until the task is finished.
 func (w *TransactionWriter) Do(db *sql.DB, f func(txn *sql.Tx) error) error {
 	if w.todo == nil {
-		w.todo = make(chan transactionWriterTask)
+		return errors.New("not initialised")
 	}
 	if !w.running.Load() {
 		go w.run()

syncapi/storage/postgres/syncserver.go

@@ -82,6 +82,7 @@ func NewDatabase(dbDataSourceName string, dbProperties internal.DbProperties) (*
 		CurrentRoomState:    currState,
 		BackwardExtremities: backwardExtremities,
 		SendToDevice:        sendToDevice,
+		SendToDeviceWriter:  internal.NewTransactionWriter(),
 		EDUCache:            cache.New(),
 	}
 	return &d, nil

syncapi/storage/shared/syncserver.go

@@ -42,7 +42,7 @@ type Database struct {
 	CurrentRoomState    tables.CurrentRoomState
 	BackwardExtremities tables.BackwardsExtremities
 	SendToDevice        tables.SendToDevice
-	SendToDeviceWriter  internal.TransactionWriter
+	SendToDeviceWriter  *internal.TransactionWriter
 	EDUCache            *cache.EDUCache
 }
 

syncapi/storage/sqlite3/syncserver.go

@@ -108,6 +108,7 @@ func (d *SyncServerDatasource) prepare() (err error) {
 		CurrentRoomState:    roomState,
 		Topology:            topology,
 		SendToDevice:        sendToDevice,
+		SendToDeviceWriter:  internal.NewTransactionWriter(),
 		EDUCache:            cache.New(),
 	}
 	return nil