diff --git a/syncapi/sync/notifier_test.go b/syncapi/sync/notifier_test.go index 39124214a..d24da4636 100644 --- a/syncapi/sync/notifier_test.go +++ b/syncapi/sync/notifier_test.go @@ -335,7 +335,7 @@ func waitForEvents(n *Notifier, req syncRequest) (types.StreamingToken, error) { return types.StreamingToken{}, fmt.Errorf( "waitForEvents timed out waiting for %s (pos=%v)", req.device.UserID, req.since, ) - case <-listener.GetNotifyChannel(*req.since): + case <-listener.GetNotifyChannel(req.since): p := listener.GetSyncPosition() return p, nil } @@ -365,7 +365,7 @@ func newTestSyncRequest(userID, deviceID string, since types.StreamingToken) syn ID: deviceID, }, timeout: 1 * time.Minute, - since: &since, + since: since, wantFullState: false, limit: DefaultTimelineLimit, log: util.GetLogger(context.TODO()), diff --git a/syncapi/sync/request.go b/syncapi/sync/request.go index d5cf143d9..f2f2894be 100644 --- a/syncapi/sync/request.go +++ b/syncapi/sync/request.go @@ -46,7 +46,7 @@ type syncRequest struct { device userapi.Device limit int timeout time.Duration - since *types.StreamingToken // nil means that no since token was supplied + since types.StreamingToken // nil means that no since token was supplied wantFullState bool log *log.Entry } @@ -55,17 +55,13 @@ func newSyncRequest(req *http.Request, device userapi.Device, syncDB storage.Dat timeout := getTimeout(req.URL.Query().Get("timeout")) fullState := req.URL.Query().Get("full_state") wantFullState := fullState != "" && fullState != "false" - var since *types.StreamingToken - sinceStr := req.URL.Query().Get("since") + since, sinceStr := types.StreamingToken{}, req.URL.Query().Get("since") if sinceStr != "" { - tok, err := types.NewStreamTokenFromString(sinceStr) + var err error + since, err = types.NewStreamTokenFromString(sinceStr) if err != nil { return nil, err } - since = &tok - } - if since == nil { - since = &types.StreamingToken{} } timelineLimit := DefaultTimelineLimit // TODO: read from stored filters too diff --git a/syncapi/sync/requestpool.go b/syncapi/sync/requestpool.go index 5bc249006..c3637f177 100644 --- a/syncapi/sync/requestpool.go +++ b/syncapi/sync/requestpool.go @@ -185,7 +185,7 @@ func (rp *RequestPool) OnIncomingSyncRequest(req *http.Request, device *userapi. // respond with, so we skip the return an go back to waiting for content to // be sent down or the request timing out. var hasTimedOut bool - sincePos := *syncReq.since + sincePos := syncReq.since for { select { // Wait for notifier to wake us up @@ -279,7 +279,7 @@ func (rp *RequestPool) currentSyncForUser(req syncRequest, latestPos types.Strea res := types.NewResponse() // See if we have any new tasks to do for the send-to-device messaging. - events, updates, deletions, err := rp.db.SendToDeviceUpdatesForSync(req.ctx, req.device.UserID, req.device.ID, *req.since) + events, updates, deletions, err := rp.db.SendToDeviceUpdatesForSync(req.ctx, req.device.UserID, req.device.ID, req.since) if err != nil { return nil, fmt.Errorf("rp.db.SendToDeviceUpdatesForSync: %w", err) } @@ -291,7 +291,7 @@ func (rp *RequestPool) currentSyncForUser(req syncRequest, latestPos types.Strea return res, fmt.Errorf("rp.db.CompleteSync: %w", err) } } else { - res, err = rp.db.IncrementalSync(req.ctx, res, req.device, *req.since, latestPos, req.limit, req.wantFullState) + res, err = rp.db.IncrementalSync(req.ctx, res, req.device, req.since, latestPos, req.limit, req.wantFullState) if err != nil { return res, fmt.Errorf("rp.db.IncrementalSync: %w", err) } @@ -302,7 +302,7 @@ func (rp *RequestPool) currentSyncForUser(req syncRequest, latestPos types.Strea if err != nil { return res, fmt.Errorf("rp.appendAccountData: %w", err) } - res, err = rp.appendDeviceLists(res, req.device.UserID, *req.since, latestPos) + res, err = rp.appendDeviceLists(res, req.device.UserID, req.since, latestPos) if err != nil { return res, fmt.Errorf("rp.appendDeviceLists: %w", err) } @@ -316,7 +316,7 @@ func (rp *RequestPool) currentSyncForUser(req syncRequest, latestPos types.Strea // Then add the updates into the sync response. if len(updates) > 0 || len(deletions) > 0 { // Handle the updates and deletions in the database. - err = rp.db.CleanSendToDeviceUpdates(context.Background(), updates, deletions, *req.since) + err = rp.db.CleanSendToDeviceUpdates(context.Background(), updates, deletions, req.since) if err != nil { return res, fmt.Errorf("rp.db.CleanSendToDeviceUpdates: %w", err) } diff --git a/syncapi/types/types.go b/syncapi/types/types.go index 0882c6298..1e29a0389 100644 --- a/syncapi/types/types.go +++ b/syncapi/types/types.go @@ -116,6 +116,17 @@ type StreamingToken struct { DeviceListPosition LogPosition } +// This will be used as a fallback by json.Marshal. +func (s StreamingToken) MarshalText() ([]byte, error) { + return []byte(s.String()), nil +} + +// This will be used as a fallback by json.Unmarshal. +func (s *StreamingToken) UnmarshalText(text []byte) (err error) { + *s, err = NewStreamTokenFromString(string(text)) + return err +} + func (t StreamingToken) String() string { posStr := fmt.Sprintf( "s%d_%d_%d_%d",