Stuff

src/github.com/matrix-org/dendrite/roomserver/input/latest_events.go

@@ -17,7 +17,6 @@ package input
 import (
 	"bytes"
 	"context"
-	"sync"
 
 	"github.com/matrix-org/dendrite/common"
 	"github.com/matrix-org/dendrite/roomserver/api"
@@ -27,8 +26,7 @@ import (
 	"github.com/matrix-org/util"
 )
 
-type sendValue struct {
+type EventSenderValue struct {
-	finishedChan  chan<- struct{}
 	stateAtEvent  types.StateAtEvent
 	event         gomatrixserverlib.Event
 	sendAsServer  string
@@ -38,31 +36,37 @@ type sendValue struct {
 type EventSender struct {
 	db           RoomEventDatabase
 	outputWriter OutputRoomEventWriter
-	sendingMutex sync.Mutex
+	linearizer   common.Linearizer
-	sending      map[types.RoomNID][]sendValue
 }
 
-func (e *EventSender) send(
+func (e *EventSender) Send(
 	ctx context.Context,
 	roomNID types.RoomNID,
 	stateAtEvent types.StateAtEvent,
 	event gomatrixserverlib.Event,
 	sendAsServer string,
 	transactionID *api.TransactionID,
-) {
+) (err error) {
-	e.sendingMutex.Lock()
+	e.linearizer.Await(event.RoomID(), func() {
-	defer e.sendingMutex.Unlock()
+		err = updateLatestEvents(ctx, e.db, e.outputWriter, roomNID, stateAtEvent, event, sendAsServer, transactionID)
-
-	finishedChan := make(chan struct{})
-	e.sending[roomNID] = append(e.sending[roomNID], sendValue{
-		finishedChan:  finishedChan,
-		stateAtEvent:  stateAtEvent,
-		event:         event,
-		sendAsServer:  sendAsServer,
-		transactionID: transactionID,
 	})
 
-	<-finishedChan
+	return
+}
+
+func (e *EventSender) SendMany(
+	ctx context.Context,
+	roomNID types.RoomNID,
+	events []EventSenderValue,
+) (err error) {
+	e.linearizer.Await(event.RoomID(), func() {
+		for i := range events {
+			err = updateLatestEvents(ctx, e.db, e.outputWriter, roomNID, stateAtEvent, event, sendAsServer, transactionID)
+			if err != nil {
+				return
+			}
+		}
+	})
 }
 
 // updateLatestEvents updates the list of latest events for this room in the database and writes the