mirrors/dendrite
1
0
Fork 0
mirror of https://github.com/matrix-org/dendrite.git synced 2025-12-07 23:13:11 -06:00
Code Issues Packages Projects Releases Wiki Activity

Start adding tests

Browse source 
This is annoying to do because we don't know WHEN the goroutine begins to block.
This commit is contained in:
Kegan Dougal 2017-05-12 17:56:33 +01:00
parent b3ad8475ac
commit 5fce7e23b0
2 changed files with 10 additions and 4 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

10
src/github.com/matrix-org/dendrite/syncapi/sync/notifier.go
View file

@ -99,8 +99,10 @@ func (n *Notifier) OnNewEvent(ev *gomatrixserverlib.Event, pos types.StreamPosit
}
}
// WaitForEvents blocks until there are new events for this request.
func (n *Notifier) WaitForEvents(req syncRequest) types.StreamPosition {
// WaitForEvents blocks until there are new events for this request. If forceBlock is true, the request
// will be forcibly waited until a new event wakes it up. This is typically only useful for testing
// blocking code.
func (n *Notifier) WaitForEvents(req syncRequest, forceBlock bool) types.StreamPosition {
// Do what synapse does: https://github.com/matrix-org/synapse/blob/v0.20.0/synapse/notifier.py#L298
// - Bucket request into a lookup map keyed off a list of joined room IDs and separately a user ID
// - Incoming events wake requests for a matching room ID
@ -109,6 +111,7 @@ func (n *Notifier) WaitForEvents(req syncRequest) types.StreamPosition {
// TODO: v1 /events 'peeking' has an 'explicit room ID' which is also tracked,
// but given we don't do /events, let's pretend it doesn't exist.
var hasBlocked bool
for {
// In a guard, check if the /sync request should block, and block it until we get woken up
n.currPosMutex.RLock()
@ -119,12 +122,15 @@ func (n *Notifier) WaitForEvents(req syncRequest) types.StreamPosition {
// with a pos which contains no new events for this user. We should probably re-wait for events
// automatically in this case.
if req.since != currentPos {
if !forceBlock || (forceBlock && hasBlocked) {
return currentPos
}
}
// wait to be woken up, and then re-check the stream position
req.log.WithField("user_id", req.userID).Info("Waiting for event")
n.blockUser(req.userID)
hasBlocked = true
}
}

2
src/github.com/matrix-org/dendrite/syncapi/sync/requestpool.go
View file

@ -99,7 +99,7 @@ func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request) util.JSONRespons
}
func (rp *RequestPool) currentSyncForUser(req syncRequest) (*types.Response, error) {
currentPos := rp.notifier.WaitForEvents(req)
currentPos := rp.notifier.WaitForEvents(req, false)
if req.since == types.StreamPosition(0) {
pos, data, err := rp.db.CompleteSync(req.userID, req.limit)