-
-
Notifications
You must be signed in to change notification settings - Fork 799
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Implement Scrobble buffering/retrying
- Loading branch information
Showing
17 changed files
with
514 additions
and
97 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,115 @@ | ||
| package scrobbler | ||
|
|
||
| import ( | ||
| "context" | ||
| "time" | ||
|
|
||
| "github.com/navidrome/navidrome/log" | ||
| "github.com/navidrome/navidrome/model" | ||
| ) | ||
|
|
||
| func NewBufferedScrobbler(ds model.DataStore, s Scrobbler, service string) *bufferedScrobbler { | ||
| b := &bufferedScrobbler{ds: ds, wrapped: s, service: service} | ||
| b.wakeSignal = make(chan struct{}, 1) | ||
| go b.run() | ||
| return b | ||
| } | ||
|
|
||
| type bufferedScrobbler struct { | ||
| ds model.DataStore | ||
| wrapped Scrobbler | ||
| service string | ||
| wakeSignal chan struct{} | ||
| } | ||
|
|
||
| func (b *bufferedScrobbler) IsAuthorized(ctx context.Context, userId string) bool { | ||
| return b.wrapped.IsAuthorized(ctx, userId) | ||
| } | ||
|
|
||
| func (b *bufferedScrobbler) NowPlaying(ctx context.Context, userId string, track *model.MediaFile) error { | ||
| return b.wrapped.NowPlaying(ctx, userId, track) | ||
| } | ||
|
|
||
| func (b *bufferedScrobbler) Scrobble(ctx context.Context, userId string, s Scrobble) error { | ||
| err := b.ds.ScrobbleBuffer(ctx).Enqueue(b.service, userId, s.ID, s.TimeStamp) | ||
| if err != nil { | ||
| return err | ||
| } | ||
|
|
||
| b.sendWakeSignal() | ||
| return nil | ||
| } | ||
|
|
||
| func (b *bufferedScrobbler) sendWakeSignal() { | ||
| // Don't block if the previous signal was not read yet | ||
| select { | ||
| case b.wakeSignal <- struct{}{}: | ||
| default: | ||
| } | ||
| } | ||
|
|
||
| func (b *bufferedScrobbler) run() { | ||
| ctx := context.Background() | ||
| for { | ||
| if !b.processQueue(ctx) { | ||
| time.AfterFunc(5*time.Second, func() { | ||
| b.sendWakeSignal() | ||
| }) | ||
| } | ||
| <-b.wakeSignal | ||
| } | ||
| } | ||
|
|
||
| func (b *bufferedScrobbler) processQueue(ctx context.Context) bool { | ||
| buffer := b.ds.ScrobbleBuffer(ctx) | ||
| userIds, err := buffer.UserIDs(b.service) | ||
| if err != nil { | ||
| log.Error(ctx, "Error retrieving userIds from scrobble buffer", "scrobbler", b.service, err) | ||
| return false | ||
| } | ||
| result := true | ||
| for _, userId := range userIds { | ||
| if !b.processUserQueue(ctx, userId) { | ||
| result = false | ||
| } | ||
| } | ||
| return result | ||
| } | ||
|
|
||
| func (b *bufferedScrobbler) processUserQueue(ctx context.Context, userId string) bool { | ||
| buffer := b.ds.ScrobbleBuffer(ctx) | ||
| for { | ||
| entry, err := buffer.Next(b.service, userId) | ||
| if err != nil { | ||
| log.Error(ctx, "Error reading from scrobble buffer", "scrobbler", b.service, err) | ||
| return false | ||
| } | ||
| if entry == nil { | ||
| return true | ||
| } | ||
| log.Debug(ctx, "Sending scrobble", "scrobbler", b.service, "track", entry.Title, "artist", entry.Artist) | ||
| err = b.wrapped.Scrobble(ctx, entry.UserID, Scrobble{ | ||
| MediaFile: entry.MediaFile, | ||
| TimeStamp: entry.PlayTime, | ||
| }) | ||
| if err != nil { | ||
| switch err { | ||
| case ErrRetryLater: | ||
| log.Warn(ctx, "Could not send scrobble. Will be retried", "userId", entry.UserID, | ||
| "track", entry.Title, "artist", entry.Artist, "scrobbler", b.service, err) | ||
| return false | ||
| default: | ||
| log.Error(ctx, "Error sending scrobble to service. Discarding", "scrobbler", b.service, | ||
| "userId", entry.UserID, "artist", entry.Artist, "track", entry.Title, err) | ||
| } | ||
| } | ||
| err = buffer.Dequeue(entry) | ||
| if err != nil { | ||
| log.Error(ctx, "Error removing entry from scrobble buffer", "userId", entry.UserID, | ||
| "track", entry.Title, "artist", entry.Artist, "scrobbler", b.service, err) | ||
| return false | ||
| } | ||
| } | ||
| } | ||
|
|
||
| var _ Scrobbler = (*bufferedScrobbler)(nil) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.