Expiry items in NowPlaying

engine/nowplaying.go

@@ -59,7 +59,7 @@ func (r *nowPlayingRepository) Enqueue(info *NowPlayingInfo) error {
 
 func (r *nowPlayingRepository) Dequeue(playerId int) (*NowPlayingInfo, error) {
 	l := r.getList(playerId)
-	e := l.Back()
+	e := checkExpired(l, l.Back)
 	if e == nil {
 		return nil, nil
 	}
@@ -69,7 +69,7 @@ func (r *nowPlayingRepository) Dequeue(playerId int) (*NowPlayingInfo, error) {
 
 func (r *nowPlayingRepository) Head(playerId int) (*NowPlayingInfo, error) {
 	l := r.getList(playerId)
-	e := l.Front()
+	e := checkExpired(l, l.Front)
 	if e == nil {
 		return nil, nil
 	}
@@ -78,7 +78,7 @@ func (r *nowPlayingRepository) Head(playerId int) (*NowPlayingInfo, error) {
 
 func (r *nowPlayingRepository) Tail(playerId int) (*NowPlayingInfo, error) {
 	l := r.getList(playerId)
-	e := l.Back()
+	e := checkExpired(l, l.Back)
 	if e == nil {
 		return nil, nil
 	}
@@ -94,9 +94,25 @@ func (r *nowPlayingRepository) GetAll() ([]*NowPlayingInfo, error) {
 	var all []*NowPlayingInfo
 	playerMap.Range(func(playerId, l interface{}) bool {
 		ll := l.(*list.List)
-		e := ll.Front()
+		e := checkExpired(ll, ll.Front)
+		if e != nil {
 			all = append(all, e.Value.(*NowPlayingInfo))
+		}
 		return true
 	})
 	return all, nil
 }
+
+func checkExpired(l *list.List, f func() *list.Element) *list.Element {
+	for {
+		e := f()
+		if e == nil {
+			return nil
+		}
+		start := e.Value.(*NowPlayingInfo).Start
+		if time.Now().Sub(start) < NowPlayingExpire {
+			return e
+		}
+		l.Remove(e)
+	}
+}

engine/nowplaying_repository_test.go

@@ -2,6 +2,7 @@ package engine
 
 import (
 	"sync"
+	"time"
 
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
@@ -9,6 +10,8 @@ import (
 
 var _ = Describe("NowPlayingRepository", func() {
 	var repo NowPlayingRepository
+	var now = time.Now()
+	var past = time.Time{}
 
 	BeforeEach(func() {
 		playerMap = sync.Map{}
@@ -16,34 +19,43 @@ var _ = Describe("NowPlayingRepository", func() {
 	})
 
 	It("enqueues and dequeues records", func() {
-		Expect(repo.Enqueue(&NowPlayingInfo{PlayerId: 1, TrackID: "AAA"})).To(BeNil())
+		Expect(repo.Enqueue(&NowPlayingInfo{PlayerId: 1, TrackID: "AAA", Start: now})).To(BeNil())
-		Expect(repo.Enqueue(&NowPlayingInfo{PlayerId: 1, TrackID: "BBB"})).To(BeNil())
+		Expect(repo.Enqueue(&NowPlayingInfo{PlayerId: 1, TrackID: "BBB", Start: now})).To(BeNil())
 
-		Expect(repo.Tail(1)).To(Equal(&NowPlayingInfo{PlayerId: 1, TrackID: "AAA"}))
+		Expect(repo.Tail(1)).To(Equal(&NowPlayingInfo{PlayerId: 1, TrackID: "AAA", Start: now}))
-		Expect(repo.Head(1)).To(Equal(&NowPlayingInfo{PlayerId: 1, TrackID: "BBB"}))
+		Expect(repo.Head(1)).To(Equal(&NowPlayingInfo{PlayerId: 1, TrackID: "BBB", Start: now}))
 
 		Expect(repo.Count(1)).To(Equal(int64(2)))
 
-		Expect(repo.Dequeue(1)).To(Equal(&NowPlayingInfo{PlayerId: 1, TrackID: "AAA"}))
+		Expect(repo.Dequeue(1)).To(Equal(&NowPlayingInfo{PlayerId: 1, TrackID: "AAA", Start: now}))
 		Expect(repo.Count(1)).To(Equal(int64(1)))
 	})
 
 	It("handles multiple players", func() {
-		Expect(repo.Enqueue(&NowPlayingInfo{PlayerId: 1, TrackID: "AAA"})).To(BeNil())
+		Expect(repo.Enqueue(&NowPlayingInfo{PlayerId: 1, TrackID: "AAA", Start: now})).To(BeNil())
-		Expect(repo.Enqueue(&NowPlayingInfo{PlayerId: 1, TrackID: "BBB"})).To(BeNil())
+		Expect(repo.Enqueue(&NowPlayingInfo{PlayerId: 1, TrackID: "BBB", Start: now})).To(BeNil())
 
-		Expect(repo.Enqueue(&NowPlayingInfo{PlayerId: 2, TrackID: "CCC"})).To(BeNil())
+		Expect(repo.Enqueue(&NowPlayingInfo{PlayerId: 2, TrackID: "CCC", Start: now})).To(BeNil())
-		Expect(repo.Enqueue(&NowPlayingInfo{PlayerId: 2, TrackID: "DDD"})).To(BeNil())
+		Expect(repo.Enqueue(&NowPlayingInfo{PlayerId: 2, TrackID: "DDD", Start: now})).To(BeNil())
 
 		Expect(repo.GetAll()).To(ConsistOf([]*NowPlayingInfo{
-			{PlayerId: 1, TrackID: "BBB"},
+			{PlayerId: 1, TrackID: "BBB", Start: now},
-			{PlayerId: 2, TrackID: "DDD"},
+			{PlayerId: 2, TrackID: "DDD", Start: now},
 		}))
 
 		Expect(repo.Count(2)).To(Equal(int64(2)))
 		Expect(repo.Count(2)).To(Equal(int64(2)))
 
-		Expect(repo.Tail(1)).To(Equal(&NowPlayingInfo{PlayerId: 1, TrackID: "AAA"}))
+		Expect(repo.Tail(1)).To(Equal(&NowPlayingInfo{PlayerId: 1, TrackID: "AAA", Start: now}))
-		Expect(repo.Head(2)).To(Equal(&NowPlayingInfo{PlayerId: 2, TrackID: "DDD"}))
+		Expect(repo.Head(2)).To(Equal(&NowPlayingInfo{PlayerId: 2, TrackID: "DDD", Start: now}))
+	})
+
+	It("handles expired items", func() {
+		Expect(repo.Enqueue(&NowPlayingInfo{PlayerId: 1, TrackID: "AAA", Start: past})).To(BeNil())
+		Expect(repo.Enqueue(&NowPlayingInfo{PlayerId: 2, TrackID: "BBB", Start: now})).To(BeNil())
+
+		Expect(repo.GetAll()).To(ConsistOf([]*NowPlayingInfo{
+			{PlayerId: 2, TrackID: "BBB", Start: now},
+		}))
 	})
 })

engine/scrobbler.go

@@ -9,11 +9,6 @@ import (
 	"github.com/cloudsonic/sonic-server/model"
 )
 
-const (
-	minSkipped = 3 * time.Second
-	maxSkipped = 20 * time.Second
-)
-
 type Scrobbler interface {
 	Register(ctx context.Context, playerId int, trackId string, playDate time.Time) (*model.MediaFile, error)
 	NowPlaying(ctx context.Context, playerId int, playerName, trackId, username string) (*model.MediaFile, error)
@@ -29,19 +24,24 @@ type scrobbler struct {
 }
 
 func (s *scrobbler) Register(ctx context.Context, playerId int, trackId string, playTime time.Time) (*model.MediaFile, error) {
-	// TODO Add transaction
+	var mf *model.MediaFile
-	mf, err := s.ds.MediaFile().Get(trackId)
+	var err error
+	err = s.ds.WithTx(func(tx model.DataStore) error {
+		mf, err = s.ds.MediaFile().Get(trackId)
 		if err != nil {
-		return nil, err
+			return err
 		}
 		err = s.ds.MediaFile().MarkAsPlayed(trackId, playTime)
 		if err != nil {
-		return nil, err
+			return err
 		}
 		err = s.ds.Album().MarkAsPlayed(mf.AlbumID, playTime)
+		return err
+	})
 	return mf, err
 }
 
+// TODO Validate if NowPlaying still works after all refactorings
 func (s *scrobbler) NowPlaying(ctx context.Context, playerId int, playerName, trackId, username string) (*model.MediaFile, error) {
 	mf, err := s.ds.MediaFile().Get(trackId)
 	if err != nil {