mirror of
https://github.com/matrix-org/dendrite
synced 2024-12-15 02:03:45 +01:00
5e9dce1c0c
* syncapi: Rename and split out tokens Previously we used the badly named `PaginationToken` which was used for both `/sync` and `/messages` requests. This quickly became confusing because named fields like `PDUPosition` meant different things depending on the token type. Instead, we now have two token types: `TopologyToken` and `StreamingToken`, both of which have fields which make more sense for their specific situations. Updated the codebase to use one or the other. `PaginationToken` still lives on as `syncToken`, an unexported type which both tokens rely on. This allows us to guarantee that the specific mappings of positions to a string remain solely under the control of the `types` package. This enables us to move high-level conceptual things like "decrement this topological token" to function calls e.g `TopologicalToken.Decrement()`. Currently broken because `/messages` seemingly used both stream and topological tokens, though I need to confirm this. * final tweaks/hacks * spurious logging * Review comments and linting
160 lines
4.4 KiB
Go
160 lines
4.4 KiB
Go
// Copyright 2017 Vector Creations Ltd
|
|
//
|
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
|
// you may not use this file except in compliance with the License.
|
|
// You may obtain a copy of the License at
|
|
//
|
|
// http://www.apache.org/licenses/LICENSE-2.0
|
|
//
|
|
// Unless required by applicable law or agreed to in writing, software
|
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
// See the License for the specific language governing permissions and
|
|
// limitations under the License.
|
|
|
|
package sync
|
|
|
|
import (
|
|
"context"
|
|
"runtime"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/matrix-org/dendrite/syncapi/types"
|
|
)
|
|
|
|
// UserStream represents a communication mechanism between the /sync request goroutine
|
|
// and the underlying sync server goroutines.
|
|
// Goroutines can get a UserStreamListener to wait for updates, and can Broadcast()
|
|
// updates.
|
|
type UserStream struct {
|
|
UserID string
|
|
// The lock that protects changes to this struct
|
|
lock sync.Mutex
|
|
// Closed when there is an update.
|
|
signalChannel chan struct{}
|
|
// The last sync position that there may have been an update for the user
|
|
pos types.StreamingToken
|
|
// The last time when we had some listeners waiting
|
|
timeOfLastChannel time.Time
|
|
// The number of listeners waiting
|
|
numWaiting uint
|
|
}
|
|
|
|
// UserStreamListener allows a sync request to wait for updates for a user.
|
|
type UserStreamListener struct {
|
|
userStream *UserStream
|
|
|
|
// Whether the stream has been closed
|
|
hasClosed bool
|
|
}
|
|
|
|
// NewUserStream creates a new user stream
|
|
func NewUserStream(userID string, currPos types.StreamingToken) *UserStream {
|
|
return &UserStream{
|
|
UserID: userID,
|
|
timeOfLastChannel: time.Now(),
|
|
pos: currPos,
|
|
signalChannel: make(chan struct{}),
|
|
}
|
|
}
|
|
|
|
// GetListener returns UserStreamListener that a sync request can use to wait
|
|
// for new updates with.
|
|
// UserStreamListener must be closed
|
|
func (s *UserStream) GetListener(ctx context.Context) UserStreamListener {
|
|
s.lock.Lock()
|
|
defer s.lock.Unlock()
|
|
|
|
s.numWaiting++ // We decrement when UserStreamListener is closed
|
|
|
|
listener := UserStreamListener{
|
|
userStream: s,
|
|
}
|
|
|
|
// Lets be a bit paranoid here and check that Close() is being called
|
|
runtime.SetFinalizer(&listener, func(l *UserStreamListener) {
|
|
if !l.hasClosed {
|
|
l.Close()
|
|
}
|
|
})
|
|
|
|
return listener
|
|
}
|
|
|
|
// Broadcast a new sync position for this user.
|
|
func (s *UserStream) Broadcast(pos types.StreamingToken) {
|
|
s.lock.Lock()
|
|
defer s.lock.Unlock()
|
|
|
|
s.pos = pos
|
|
|
|
close(s.signalChannel)
|
|
|
|
s.signalChannel = make(chan struct{})
|
|
}
|
|
|
|
// NumWaiting returns the number of goroutines waiting for waiting for updates.
|
|
// Used for metrics and testing.
|
|
func (s *UserStream) NumWaiting() uint {
|
|
s.lock.Lock()
|
|
defer s.lock.Unlock()
|
|
return s.numWaiting
|
|
}
|
|
|
|
// TimeOfLastNonEmpty returns the last time that the number of waiting listeners
|
|
// was non-empty, may be time.Now() if number of waiting listeners is currently
|
|
// non-empty.
|
|
func (s *UserStream) TimeOfLastNonEmpty() time.Time {
|
|
s.lock.Lock()
|
|
defer s.lock.Unlock()
|
|
|
|
if s.numWaiting > 0 {
|
|
return time.Now()
|
|
}
|
|
|
|
return s.timeOfLastChannel
|
|
}
|
|
|
|
// GetSyncPosition returns last sync position which the UserStream was
|
|
// notified about
|
|
func (s *UserStreamListener) GetSyncPosition() types.StreamingToken {
|
|
s.userStream.lock.Lock()
|
|
defer s.userStream.lock.Unlock()
|
|
|
|
return s.userStream.pos
|
|
}
|
|
|
|
// GetNotifyChannel returns a channel that is closed when there may be an
|
|
// update for the user.
|
|
// sincePos specifies from which point we want to be notified about. If there
|
|
// has already been an update after sincePos we'll return a closed channel
|
|
// immediately.
|
|
func (s *UserStreamListener) GetNotifyChannel(sincePos types.StreamingToken) <-chan struct{} {
|
|
s.userStream.lock.Lock()
|
|
defer s.userStream.lock.Unlock()
|
|
|
|
if s.userStream.pos.IsAfter(sincePos) {
|
|
// If the listener is behind, i.e. missed a potential update, then we
|
|
// want them to wake up immediately. We do this by returning a new
|
|
// closed stream, which returns immediately when selected.
|
|
closedChannel := make(chan struct{})
|
|
close(closedChannel)
|
|
return closedChannel
|
|
}
|
|
|
|
return s.userStream.signalChannel
|
|
}
|
|
|
|
// Close cleans up resources used
|
|
func (s *UserStreamListener) Close() {
|
|
s.userStream.lock.Lock()
|
|
defer s.userStream.lock.Unlock()
|
|
|
|
if !s.hasClosed {
|
|
s.userStream.numWaiting--
|
|
s.userStream.timeOfLastChannel = time.Now()
|
|
}
|
|
|
|
s.hasClosed = true
|
|
}
|