Creates a Go-channel of all the changes made on a channel. Does NOT handle the Wait option. Does NOT check authorization.
(ctx context.Context, singleChannelCache SingleChannelCache, options ChangesOptions, to string)
| 442 | // Creates a Go-channel of all the changes made on a channel. |
| 443 | // Does NOT handle the Wait option. Does NOT check authorization. |
| 444 | func (db *DatabaseCollectionWithUser) changesFeed(ctx context.Context, singleChannelCache SingleChannelCache, options ChangesOptions, to string) <-chan *ChangeEntry { |
| 445 | |
| 446 | feed := make(chan *ChangeEntry, 1) |
| 447 | |
| 448 | queryLimit := db.channelQueryLimit() |
| 449 | requestLimit := options.Limit |
| 450 | |
| 451 | // Make a copy of the changesOptions so that query pagination can modify since and limit. Pagination uses safe sequence |
| 452 | // as starting point and can subsequently ignore LowSeq - it is added back to entries as needed when the main |
| 453 | // changes loop processes the channel's feed. |
| 454 | paginationOptions := options |
| 455 | paginationOptions.Since.Seq = options.Since.SafeSequence() |
| 456 | paginationOptions.Since.LowSeq = 0 |
| 457 | |
| 458 | go func() { |
| 459 | defer base.FatalPanicHandler() |
| 460 | defer close(feed) |
| 461 | var itemsSent int |
| 462 | var lastSeq uint64 |
| 463 | // Pagination based on ChannelQueryLimit. This loop may terminated in three ways (see return statements): |
| 464 | // 1. Query returns fewer rows than ChannelQueryLimit |
| 465 | // 2. A limit is specified on the incoming ChangesOptions, and that limit is reached |
| 466 | // 3. An error is returned when calling singleChannelCache.GetChanges |
| 467 | for { |
| 468 | if options.ChangesCtx.Err() != nil { |
| 469 | base.DebugfCtx(ctx, base.KeyChanges, "Terminating channel feed %s", base.UD(to)) |
| 470 | return |
| 471 | } |
| 472 | // Calculate limit for this iteration |
| 473 | if requestLimit == 0 { |
| 474 | paginationOptions.Limit = queryLimit |
| 475 | } else { |
| 476 | remainingLimit := requestLimit - itemsSent |
| 477 | paginationOptions.Limit = base.Min(remainingLimit, queryLimit) |
| 478 | } |
| 479 | |
| 480 | base.TracefCtx(ctx, base.KeyChanges, "Querying channel %q with options: %+v", base.UD(singleChannelCache.ChannelID().Name), paginationOptions) |
| 481 | changes, err := singleChannelCache.GetChanges(ctx, paginationOptions) |
| 482 | if err != nil { |
| 483 | base.WarnfCtx(ctx, "Error retrieving changes for channel %q: %v", base.UD(singleChannelCache.ChannelID().Name), err) |
| 484 | change := ChangeEntry{ |
| 485 | Err: base.ErrChannelFeed, |
| 486 | } |
| 487 | feed <- &change |
| 488 | return |
| 489 | } |
| 490 | base.DebugfCtx(ctx, base.KeyChanges, "[changesFeed] Found %d changes for channel %q", len(changes), base.UD(singleChannelCache.ChannelID().Name)) |
| 491 | |
| 492 | // Now write each log entry to the 'feed' channel in turn: |
| 493 | sentChanges := 0 |
| 494 | for _, logEntry := range changes { |
| 495 | if options.ChangesCtx.Err() != nil { |
| 496 | base.DebugfCtx(ctx, base.KeyChanges, "Terminating channel feed %s", base.UD(to)) |
| 497 | return |
| 498 | } |
| 499 | if logEntry.Sequence >= options.Since.TriggeredBy { |
| 500 | options.Since.TriggeredBy = 0 |
| 501 | } |
no test coverage detected