MirrorHub/dendrite
0
0
Fork 0
mirror of https://github.com/matrix-org/dendrite synced 2024-11-19 00:00:55 +01:00
Code Issues Releases Wiki Activity

Tweak typing

Browse source
This commit is contained in:
Neil Alexander 2020-12-14 16:55:55 +00:00
parent 1bc1347ce4
commit e8d485d881
No known key found for this signature in database
GPG key ID: A02A2019A2BB0944
3 changed files with 20 additions and 9 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
syncapi/consumers/eduserver_receipts.go
View file

@ -88,7 +88,7 @@ func (s *OutputReceiptEventConsumer) onMessage(msg *sarama.ConsumerMessage) erro
return err
}
// update stream position
s.notifier.OnNewReceipt(types.StreamingToken{ReceiptPosition: streamPos})
s.notifier.OnNewReceipt(output.RoomID, types.StreamingToken{ReceiptPosition: streamPos})
return nil
}

9
syncapi/consumers/eduserver_typing.go
View file

@ -64,12 +64,7 @@ func NewOutputTypingEventConsumer(
// Start consuming from EDU api
func (s *OutputTypingEventConsumer) Start() error {
s.db.SetTypingTimeoutCallback(func(userID, roomID string, latestSyncPosition int64) {
s.notifier.OnNewEvent(
nil, roomID, nil,
types.StreamingToken{
TypingPosition: types.StreamPosition(latestSyncPosition),
},
)
s.notifier.OnNewTyping(roomID, types.StreamingToken{TypingPosition: types.StreamPosition(latestSyncPosition)})
})
return s.typingConsumer.Start()
@ -97,6 +92,6 @@ func (s *OutputTypingEventConsumer) onMessage(msg *sarama.ConsumerMessage) error
typingPos = s.db.RemoveTypingUser(typingEvent.UserID, typingEvent.RoomID)
}
s.notifier.OnNewEvent(nil, output.Event.RoomID, nil, types.StreamingToken{TypingPosition: typingPos})
s.notifier.OnNewTyping(output.Event.RoomID, types.StreamingToken{TypingPosition: typingPos})
return nil
}

18
syncapi/sync/notifier.go
View file

@ -162,13 +162,29 @@ func (n *Notifier) OnNewSendToDevice(
}
// OnNewReceipt updates the current position
func (n *Notifier) OnNewReceipt(
func (n *Notifier) OnNewTyping(
roomID string,
posUpdate types.StreamingToken,
) {
n.streamLock.Lock()
defer n.streamLock.Unlock()
latestPos := n.currPos.WithUpdates(posUpdate)
n.currPos = latestPos
n.wakeupUsers(n.joinedUsers(roomID), nil, latestPos)
}
// OnNewReceipt updates the current position
func (n *Notifier) OnNewReceipt(
roomID string,
posUpdate types.StreamingToken,
) {
n.streamLock.Lock()
defer n.streamLock.Unlock()
latestPos := n.currPos.WithUpdates(posUpdate)
n.currPos = latestPos
n.wakeupUsers(n.joinedUsers(roomID), nil, latestPos)
}
func (n *Notifier) OnNewKeyChange(