Support putting data on over int16 timecode jump (#318)

seqsense · Oct 26, 2022 · 4296b50 · 4296b50
1 parent 5935929
commit 4296b50
Show file tree

Hide file tree

Showing 2 changed files with 41 additions and 18 deletions.
diff --git a/provider.go b/provider.go
@@ -257,34 +257,48 @@ func (p *Provider) PutMedia(opts ...PutMediaOption) (BlockWriter, error) {
 		case <-ctx.Done():
 		}
 	}
+	prepareNextConn := func() {
+		nextConn = newConnection(options)
+		chConnection <- nextConn
+	}
+	switchToNextConn := func(startTime uint64) {
+		if conn != nil {
+			conn.close()
+		}
+		timeout.Stop()
+		conn = nextConn
+		conn.initialize(startTime, options)
+		resetTimeout()
+		nextConn = nil
+	}
+
 	writer := &blockWriter{
 		fnWrite: func(bt *BlockWithBaseTimecode) error {
 			absTime := uint64(bt.AbsTimecode())
 			if lastAbsTime != 0 {
 				diff := int64(absTime - lastAbsTime)
-				if diff < 0 || diff > math.MaxInt16 {
+				if diff < 0 {
 					return fmt.Errorf(`stream_id=%s, timecode=%d, last=%d, diff=%d: %w`,
 						p.streamID, bt.AbsTimecode(), lastAbsTime, diff,
 						ErrInvalidTimecode,
 					)
 				}
+				if diff > math.MaxInt16 {
+					options.logger.Debugf(`Forcing next connection: { StreamID: "%s", AbsTime: %d, LastAbsTime: %d, Diff: %d }`,
+						p.streamID, bt.AbsTimecode(), lastAbsTime, diff,
+					)
+					prepareNextConn()
+					switchToNextConn(absTime)
+				}
 			}
 
 			if conn == nil || (nextConn == nil && int16(absTime-conn.baseTimecode) > 8000) {
 				options.logger.Debugf(`Prepare next connection: { StreamID: "%s" }`, p.streamID)
-				nextConn = newConnection(options)
-				chConnection <- nextConn
+				prepareNextConn()
 			}
 			if conn == nil || int16(absTime-conn.baseTimecode) > 9000 {
 				options.logger.Debugf(`Switch to next connection: { StreamID: "%s", AbsTime: %d }`, p.streamID, absTime)
-				if conn != nil {
-					conn.close()
-				}
-				timeout.Stop()
-				conn = nextConn
-				conn.initialize(absTime, options)
-				resetTimeout()
-				nextConn = nil
+				switchToNextConn(absTime)
 			}
 			bt.Block.Timecode = int16(absTime - conn.baseTimecode)
 			select {

diff --git a/provider_test.go b/provider_test.go
@@ -63,6 +63,13 @@ func TestProvider(t *testing.T) {
 		},
 		Tags: newTags([]kvm.SimpleTag{{TagName: "TEST_TAG", TagString: "2"}}),
 	}
+	expected2 := kvsm.FragmentTest{
+		Cluster: kvsm.ClusterTest{
+			Timecode:    startTimestampInMillis + 43000,
+			SimpleBlock: []ebml.Block{newBlock(0), newBlock(1)},
+		},
+		Tags: newTags([]kvm.SimpleTag{{TagName: "TEST_TAG", TagString: "3"}}),
+	}
 
 	testCases := map[string]struct {
 		mockServerOpts func(*testing.T, map[uint64]bool, *bool, func()) []kvsm.KinesisVideoServerOption
@@ -72,7 +79,7 @@ func TestProvider(t *testing.T) {
 	}{
 		"NoError": {
 			mockServerOpts: func(*testing.T, map[uint64]bool, *bool, func()) []kvsm.KinesisVideoServerOption { return nil },
-			expected:       []kvsm.FragmentTest{expected0, expected1},
+			expected:       []kvsm.FragmentTest{expected0, expected1, expected2},
 		},
 		"HTTPErrorRetry": {
 			mockServerOpts: func(t *testing.T, dropped map[uint64]bool, _ *bool, _ func()) []kvsm.KinesisVideoServerOption {
@@ -94,7 +101,7 @@ func TestProvider(t *testing.T) {
 				}
 			},
 			putMediaOpts: []kvm.PutMediaOption{retryOpt},
-			expected:     []kvsm.FragmentTest{expected0, expected1},
+			expected:     []kvsm.FragmentTest{expected0, expected1, expected2},
 		},
 		"DelayedHTTPErrorRetry": {
 			mockServerOpts: func(t *testing.T, dropped map[uint64]bool, _ *bool, _ func()) []kvsm.KinesisVideoServerOption {
@@ -117,7 +124,7 @@ func TestProvider(t *testing.T) {
 				}
 			},
 			putMediaOpts: []kvm.PutMediaOption{retryOpt},
-			expected:     []kvsm.FragmentTest{expected0, expected1},
+			expected:     []kvsm.FragmentTest{expected0, expected1, expected2},
 		},
 		"KinesisErrorRetry": {
 			mockServerOpts: func(t *testing.T, dropped map[uint64]bool, _ *bool, _ func()) []kvsm.KinesisVideoServerOption {
@@ -138,7 +145,7 @@ func TestProvider(t *testing.T) {
 				}
 			},
 			putMediaOpts: []kvm.PutMediaOption{retryOpt},
-			expected:     []kvsm.FragmentTest{expected0, expected0, expected1, expected1},
+			expected:     []kvsm.FragmentTest{expected0, expected0, expected1, expected1, expected2, expected2},
 		},
 		"KinesisFailDumpShort": {
 			mockServerOpts: func(t *testing.T, dropped map[uint64]bool, _ *bool, _ func()) []kvsm.KinesisVideoServerOption {
@@ -272,7 +279,7 @@ func TestProvider(t *testing.T) {
 				}
 			},
 			putMediaOpts: []kvm.PutMediaOption{retryOpt},
-			expected:     []kvsm.FragmentTest{expected0, expected0, expected1, expected1},
+			expected:     []kvsm.FragmentTest{expected0, expected0, expected1, expected1, expected2, expected2},
 		},
 		"DisconnectRetry": {
 			mockServerOpts: func(t *testing.T, _ map[uint64]bool, disconnected *bool, disconnect func()) []kvsm.KinesisVideoServerOption {
@@ -291,7 +298,7 @@ func TestProvider(t *testing.T) {
 				}
 			},
 			putMediaOpts: []kvm.PutMediaOption{retryOpt},
-			expected:     []kvsm.FragmentTest{expected0, expected1},
+			expected:     []kvsm.FragmentTest{expected0, expected1, expected2},
 		},
 		"DelayedDisconnectRetry": {
 			mockServerOpts: func(t *testing.T, _ map[uint64]bool, disconnected *bool, disconnect func()) []kvsm.KinesisVideoServerOption {
@@ -311,7 +318,7 @@ func TestProvider(t *testing.T) {
 				}
 			},
 			putMediaOpts: []kvm.PutMediaOption{retryOpt},
-			expected:     []kvsm.FragmentTest{expected0, expected1},
+			expected:     []kvsm.FragmentTest{expected0, expected1, expected2},
 		},
 	}
 
@@ -337,6 +344,8 @@ func TestProvider(t *testing.T) {
 				10000,
 				10001, // switch to the next fragment here
 				10002,
+				43000, // force next fragment (jump >32768)
+				43001,
 			}
 
 			var response []kvm.FragmentEvent