dendrite/syncapi/streams/stream_devicelist.go
Neil Alexander 62a325ded8
Complete sync performance (#1741)
* Parallelise PDU stream fetching for complete sync

* Fixes

* Fixes

* Worker queue

* Workers

* Don't populate device list changes on complete sync

* Don't fast-forward typing notifications either on complete sync

* Revert "Don't fast-forward typing notifications either on complete sync"
This reverts commit 01471f78431cdd840915111f71bd2b5176e584a8.

* Comments
2021-01-29 16:32:54 +00:00

44 lines
1.1 KiB
Go

package streams
import (
"context"
keyapi "github.com/matrix-org/dendrite/keyserver/api"
"github.com/matrix-org/dendrite/roomserver/api"
"github.com/matrix-org/dendrite/syncapi/internal"
"github.com/matrix-org/dendrite/syncapi/types"
)
type DeviceListStreamProvider struct {
PartitionedStreamProvider
rsAPI api.RoomserverInternalAPI
keyAPI keyapi.KeyInternalAPI
}
func (p *DeviceListStreamProvider) CompleteSync(
ctx context.Context,
req *types.SyncRequest,
) types.LogPosition {
return p.LatestPosition(ctx)
}
func (p *DeviceListStreamProvider) IncrementalSync(
ctx context.Context,
req *types.SyncRequest,
from, to types.LogPosition,
) types.LogPosition {
var err error
to, _, err = internal.DeviceListCatchup(context.Background(), p.keyAPI, p.rsAPI, req.Device.UserID, req.Response, from, to)
if err != nil {
req.Log.WithError(err).Error("internal.DeviceListCatchup failed")
return from
}
err = internal.DeviceOTKCounts(req.Context, p.keyAPI, req.Device.UserID, req.Device.ID, req.Response)
if err != nil {
req.Log.WithError(err).Error("internal.DeviceListCatchup failed")
return from
}
return to
}