Add Setup method to PresenceStreamProvider

This commit is contained in:
Till Faelligen 2021-08-03 15:32:34 +02:00
parent 7003ae501d
commit 642ac3e4ad
2 changed files with 15 additions and 1 deletions

View file

@ -66,6 +66,7 @@ func NewSyncStreamProviders(
streams.SendToDeviceStreamProvider.Setup()
streams.AccountDataStreamProvider.Setup()
streams.DeviceListStreamProvider.Setup()
streams.PresenceDataStreamProdiver.Setup()
return streams
}
@ -79,5 +80,6 @@ func (s *Streams) Latest(ctx context.Context) types.StreamingToken {
SendToDevicePosition: s.SendToDeviceStreamProvider.LatestPosition(ctx),
AccountDataPosition: s.AccountDataStreamProvider.LatestPosition(ctx),
DeviceListPosition: s.DeviceListStreamProvider.LatestPosition(ctx),
PresenceDataPosition: s.PresenceDataStreamProdiver.LatestPosition(ctx),
}
}

View file

@ -15,8 +15,17 @@ type PresenceStreamProvider struct {
UserAPI userapi.UserInternalAPI
}
func (p *PresenceStreamProvider) Setup() {
p.StreamProvider.Setup()
res := userapi.QueryMaxPresenceIDResponse{}
if err := p.UserAPI.QueryMaxPresenceID(context.Background(), &userapi.QueryMaxPresenceIDRequest{}, &res); err != nil {
panic(err)
}
p.latest = types.StreamPosition(res.ID)
}
func (p *PresenceStreamProvider) CompleteSync(ctx context.Context, req *types.SyncRequest) types.StreamPosition {
req.Log.Debug(" CompleteSyncrequested for presence!")
return p.IncrementalSync(ctx, req, 0, p.LatestPosition(ctx))
}
@ -34,6 +43,9 @@ func (p *PresenceStreamProvider) IncrementalSync(ctx context.Context, req *types
req.Log.WithError(err).Error("unable to fetch presence after")
return from
}
if len(res.Presences) == 0 {
return to
}
evs := []gomatrixserverlib.ClientEvent{}
var maxPos int64
for _, presence := range res.Presences {