Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion ausocean_test_index.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -80,7 +80,7 @@ indexes:
- kind: MtsMediaV2
properties:
- name: BroadcastID
- name: Timestamp
- name: Date

- kind: Text
properties:
Expand Down
2 changes: 1 addition & 1 deletion cmd/oceanmedia/server.go
Original file line number Diff line number Diff line change
Expand Up @@ -101,7 +101,7 @@ func newServer(store datastore.Store, signer urlSigner, jwtSecret []byte, signed
return model.GetBroadcastEvent(ctx, s.store, broadcastID)
}
s.loadSegments = func(ctx context.Context, broadcastID string) ([]model.MtsMediaV2, error) {
return model.GetMtsMediaV2ByBroadcast(ctx, s.store, broadcastID, nil, 0)
return model.GetMtsMediaV2ByBroadcast(ctx, s.store, broadcastID, 0)
}
return s
}
Expand Down
34 changes: 13 additions & 21 deletions model/mtsmediav2.go
Original file line number Diff line number Diff line change
Expand Up @@ -101,14 +101,12 @@ func DeleteMtsMediaV2(ctx context.Context, store datastore.Store, mid int64, tim
}

// GetMtsMediaV2ByBroadcast retrieves the MtsMediaV2 entities belonging
// to a broadcast, optionally filtered by timestamp(s). One timestamp
// represents a lower bound (inclusive) on the segment timestamp,
// whereas two represents a time range. Results are ordered by segment
// timestamp ascending.
// to a broadcast, ordered by Date, i.e. the order in which they were
// recorded.
//
// A limit of zero or less uses a default size to cover the whole
// broadcast.
func GetMtsMediaV2ByBroadcast(ctx context.Context, store datastore.Store, broadcastID string, ts []int64, limit int) ([]MtsMediaV2, error) {
func GetMtsMediaV2ByBroadcast(ctx context.Context, store datastore.Store, broadcastID string, limit int) ([]MtsMediaV2, error) {
if broadcastID == "" {
return nil, errors.New("empty broadcast ID")
}
Expand All @@ -123,19 +121,7 @@ func GetMtsMediaV2ByBroadcast(ctx context.Context, store datastore.Store, broadc
if err != nil {
return nil, fmt.Errorf("failed to filter MtsMediaV2 by broadcast ID: %w", err)
}
if len(ts) > 0 {
err = q.FilterField("Timestamp", ">=", ts[0])
if err != nil {
return nil, fmt.Errorf("failed to filter MtsMediaV2 by timestamp: %w", err)
}
}
if len(ts) > 1 {
err = q.FilterField("Timestamp", "<", ts[1])
if err != nil {
return nil, fmt.Errorf("failed to filter MtsMediaV2 by timestamp: %w", err)
}
}
q.Order("Timestamp")
q.Order("Date")
q.Limit(limit)

var media []MtsMediaV2
Expand All @@ -144,10 +130,16 @@ func GetMtsMediaV2ByBroadcast(ctx context.Context, store datastore.Store, broadc
return nil, fmt.Errorf("failed to get MtsMediaV2 for broadcast %s: %w", broadcastID, err)
}

// FileStore orders by key name, which is not guaranteed to match a
// segment's timestamp, so sort explicitly.
// FileStore orders by key name, which is not guaranteed to match the
// Date, so sort explicitly. Timestamp and MID are used to break ties.
sort.Slice(media, func(i, j int) bool {
return media[i].Timestamp < media[j].Timestamp
if !media[i].Date.Equal(media[j].Date) {
return media[i].Date.Before(media[j].Date)
}
if media[i].Timestamp != media[j].Timestamp {
return media[i].Timestamp < media[j].Timestamp
}
return media[i].MID < media[j].MID
})
return media, nil
}
52 changes: 25 additions & 27 deletions model/mtsmediav2_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,7 @@ package model
import (
"context"
"testing"
"time"

"github.com/ausocean/cloud/datastore"
"github.com/stretchr/testify/assert"
Expand All @@ -38,13 +39,14 @@ const (
testBcv2MID = int64(16)
)

// newMtsMediaV2 creates a new MtsMediaV2 entity with the given broadcast
// ID and timestamp, registering cleanup with t.
func newMtsMediaV2(t *testing.T, ctx context.Context, store datastore.Store, broadcastID string, mid, ts int64) {
// newMtsMediaV2 creates a new MtsMediaV2 entity, registering cleanup with t.
// Its Date (used for ordering) is set to now by CreateMtsMediaV2.
func newMtsMediaV2(t *testing.T, ctx context.Context, store datastore.Store, broadcastID string, mid, ts, pts int64) {
t.Helper()
_, err := CreateMtsMediaV2(ctx, store, MtsMediaV2{
MID: mid,
Timestamp: ts,
PTS: pts,
Duration: 2541,
StorageURI: "https://example.com/bucket/" + broadcastID + "/segment.ts",
BroadcastID: broadcastID,
Expand All @@ -61,10 +63,16 @@ func TestGetMtsMediaV2ByBroadcast(t *testing.T) {
t.Run(kind, func(t *testing.T) {
ctx, store := setupStore(t, kind)

for _, ts := range []int64{1000, 1005, 1010, 1015} {
newMtsMediaV2(t, ctx, store, testBcv2BroadcastA, testBcv2MID, ts)
}
newMtsMediaV2(t, ctx, store, testBcv2BroadcastB, testBcv2MID, 2000)
// Creation order deliberately disagrees with timestamp and PTS
// order, so that ordering by Date (creation order) is exercised.
newMtsMediaV2(t, ctx, store, testBcv2BroadcastA, testBcv2MID, 1010, 400)
time.Sleep(time.Millisecond)
newMtsMediaV2(t, ctx, store, testBcv2BroadcastA, testBcv2MID, 1000, 300)
time.Sleep(time.Millisecond)
newMtsMediaV2(t, ctx, store, testBcv2BroadcastA, testBcv2MID, 1015, 200)
time.Sleep(time.Millisecond)
newMtsMediaV2(t, ctx, store, testBcv2BroadcastA, testBcv2MID, 1005, 100)
newMtsMediaV2(t, ctx, store, testBcv2BroadcastB, testBcv2MID, 2000, 10)

timestamps := func(media []MtsMediaV2) []int64 {
out := make([]int64, len(media))
Expand All @@ -74,44 +82,34 @@ func TestGetMtsMediaV2ByBroadcast(t *testing.T) {
return out
}

t.Run("AllOrdered", func(t *testing.T) {
got, err := GetMtsMediaV2ByBroadcast(ctx, store, testBcv2BroadcastA, nil, 0)
t.Run("OrderedByDate", func(t *testing.T) {
got, err := GetMtsMediaV2ByBroadcast(ctx, store, testBcv2BroadcastA, 0)
assert.NoError(t, err)
assert.Equal(t, []int64{1000, 1005, 1010, 1015}, timestamps(got))
assert.Equal(t, []int64{1010, 1000, 1015, 1005}, timestamps(got))
})

t.Run("BroadcastFilter", func(t *testing.T) {
got, err := GetMtsMediaV2ByBroadcast(ctx, store, testBcv2BroadcastB, nil, 0)
got, err := GetMtsMediaV2ByBroadcast(ctx, store, testBcv2BroadcastB, 0)
assert.NoError(t, err)
assert.Equal(t, []int64{2000}, timestamps(got))
})

t.Run("FromTimestamp", func(t *testing.T) {
got, err := GetMtsMediaV2ByBroadcast(ctx, store, testBcv2BroadcastA, []int64{1005}, 0)
assert.NoError(t, err)
assert.Equal(t, []int64{1005, 1010, 1015}, timestamps(got))
})

t.Run("TimestampRange", func(t *testing.T) {
got, err := GetMtsMediaV2ByBroadcast(ctx, store, testBcv2BroadcastA, []int64{1005, 1015}, 0)
assert.NoError(t, err)
assert.Equal(t, []int64{1005, 1010}, timestamps(got))
})

t.Run("Limit", func(t *testing.T) {
got, err := GetMtsMediaV2ByBroadcast(ctx, store, testBcv2BroadcastA, nil, 2)
// The limit is a safety cap. FileStore applies it before
// sorting in the model layer, so only assert the cap here.
got, err := GetMtsMediaV2ByBroadcast(ctx, store, testBcv2BroadcastA, 2)
assert.NoError(t, err)
assert.Equal(t, []int64{1000, 1005}, timestamps(got))
assert.Len(t, got, 2)
})

t.Run("NoResults", func(t *testing.T) {
got, err := GetMtsMediaV2ByBroadcast(ctx, store, "no-such-broadcast", nil, 0)
got, err := GetMtsMediaV2ByBroadcast(ctx, store, "no-such-broadcast", 0)
assert.NoError(t, err)
assert.Empty(t, got)
})

t.Run("EmptyBroadcastID", func(t *testing.T) {
_, err := GetMtsMediaV2ByBroadcast(ctx, store, "", nil, 0)
_, err := GetMtsMediaV2ByBroadcast(ctx, store, "", 0)
assert.Error(t, err)
})
})
Expand Down
2 changes: 1 addition & 1 deletion vidgrind_index.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -8,7 +8,7 @@ indexes:
- kind: MtsMediaV2
properties:
- name: BroadcastID
- name: Timestamp
- name: Date

- kind: Text
properties:
Expand Down
Loading