cmd/mxl-player/main.go

@@ -17,7 +17,6 @@ import (
 
 	cimgui "github.com/AllenDang/cimgui-go/imgui"
 	vk "github.com/christerso/vulkan-go/vk"
-	"github.com/qvest-digital/go-mxl/mxl"
 	pflag "github.com/spf13/pflag"
 )
 
@@ -78,8 +77,10 @@ func checkMXLargs(args appArgs) {
 }
 
 func main() {
-	// video: 5fbec3b1-1b0f-417d-9059-8b94a47197ed
+	// timelapse video: 5fbec3b1-1b0f-417d-9059-8b94a47197ed
-	// audio: 5fbec3b1-1b0f-417d-9059-8b94a47197ec
+	// timelapse audio: 5fbec3b1-1b0f-417d-9059-8b94a47197ec
+	// f1 video: 5fbec3b1-1b0f-417d-9059-8b94a47197ef
+	// f1 audio: 5fbec3b1-1b0f-417d-9059-8b94a47197eb
 	var args appArgs
 	flagSet := pflag.NewFlagSet(APP_NAME, pflag.ContinueOnError)
 	flagSet.Usage = func() { printUsage(os.Stderr) }
@@ -259,7 +260,6 @@ func main() {
 		fmt.Printf("sync: video %dx%d audio %dch batch=%d\n",
 			syncSrc.Width(), syncSrc.Height(), aChans, audioBatch)
 	case args.VideoFlowId != "":
-
 		videoSrc, err = source.Open(args.Domain, args.VideoFlowId)
 		if err != nil {
 			log.Fatalf("source: %v", err)
@@ -338,30 +338,43 @@ func main() {
 		defer vkDevice.WaitIdle()
 	}
 
+	// GUI state (accessible from doReconnect + goroutine)
+	var (
+		domainStr string = args.Domain
+		videoStr  string = args.VideoFlowId
+		audioStr  string = args.AudioFlowId
+		showStats bool   = true
+	)
+
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
-	grant := make(chan struct{}, 1)
+	type reconnectParams struct {
+		domain string
+		video  string
+		audio  string
+	}
+	// One control channel: grant (empty params) or reconnect (with params).
+	control := make(chan reconnectParams, 1)
 	staged := make(chan uint64)
-	failed := make(chan struct{})
 
-	reopen := func() error {
+	reopen := func(params reconnectParams) error {
+		// Close current sources
 		if syncSrc != nil {
 			_ = syncSrc.Close()
+			syncSrc = nil
 		}
 		if videoSrc != nil {
 			_ = videoSrc.Close()
+			videoSrc = nil
 		}
 		if audioSrc != nil {
 			_ = audioSrc.Close()
+			audioSrc = nil
 		}
-		for {
+		// Try once. Return error if fails — caller loops back to select
-			select {
+		// and can pick up new reconnect params or a new grant.
-			case <-ctx.Done():
+		if params.video != "" && params.audio != "" {
-				return ctx.Err()
+			s, e := source.OpenSync(params.domain, params.video, params.audio)
-			default:
-			}
-			if syncSrc != nil {
-				s, e := source.OpenSync(args.Domain, args.VideoFlowId, args.AudioFlowId)
 			if e == nil {
 				if r != nil {
 					newSize := vk.DeviceSize(s.Stride()) * vk.DeviceSize(s.Height())
@@ -372,11 +385,16 @@ func main() {
 					}
 				}
 				syncSrc = s
+				aChans = s.Channels()
+				audioBatch = uint64(s.AudioRate().Num) / uint64(s.Rate().Num)
+				if audioBatch == 0 {
+					audioBatch = 1
+				}
 				return nil
 			}
-				log.Printf("source: reopen retry: %v", e)
+			return e
-			} else if videoSrc != nil {
+		} else if params.video != "" {
-				s, e := source.Open(args.Domain, args.VideoFlowId)
+			s, e := source.Open(params.domain, params.video)
 			if e == nil {
 				if r != nil {
 					newSize := vk.DeviceSize(s.Stride()) * vk.DeviceSize(s.Height())
@@ -389,17 +407,29 @@ func main() {
 				videoSrc = s
 				return nil
 			}
-				log.Printf("source: reopen retry: %v", e)
+			return e
-			} else if audioSrc != nil {
+		} else if params.audio != "" {
-				s, e := source.OpenAudio(args.Domain, args.AudioFlowId)
+			s, e := source.OpenAudio(params.domain, params.audio)
 			if e == nil {
 				audioSrc = s
+				aChans = s.Channels()
+				audioBatch = uint64(s.Rate().Num) / (100 * uint64(s.Rate().Den))
+				if audioBatch == 0 {
+					audioBatch = 1
+				}
 				return nil
 			}
-				log.Printf("source: reopen retry: %v", e)
+			return e
 		}
-			time.Sleep(500 * time.Millisecond)
+		return fmt.Errorf("reopen: no flow specified")
 	}
+
+	doReconnect := func() {
+		select {
+		case <-control:
+		default:
+		}
+		control <- reconnectParams{domain: domainStr, video: videoStr, audio: audioStr}
 	}
 
 	go func() {
@@ -409,6 +439,25 @@ func main() {
 				select {
 				case <-ctx.Done():
 					return
+				case params := <-control:
+					if params.video != "" || params.audio != "" {
+						if rerr := reopen(params); rerr != nil {
+							if errors.Is(rerr, context.Canceled) {
+								return
+							}
+							log.Printf("source: reopen failed: %v, retrying", rerr)
+							select {
+							case <-time.After(500 * time.Millisecond):
+							case <-ctx.Done():
+								return
+							}
+							select {
+							case control <- params:
+							default:
+							}
+						}
+					}
+					continue
 				default:
 				}
 				queued := sdl.GetAudioStreamQueued(audioStream)
@@ -426,19 +475,23 @@ func main() {
 					if errors.Is(err, context.Canceled) {
 						return
 					}
-					if errors.Is(err, mxl.ErrFlowInvalid) {
+					log.Printf("source: %v", err)
-						log.Printf("source: flow invalid, reopening")
+					params := reconnectParams{domain: domainStr, video: videoStr, audio: audioStr}
-						if rerr := reopen(); rerr != nil {
+					select {
-							log.Printf("source: reopen failed: %v", rerr)
+					case <-control:
-							cancel()
+					default:
+					}
+					select {
+					case <-time.After(500 * time.Millisecond):
+					case <-ctx.Done():
 						return
 					}
+					select {
+					case control <- params:
+					default:
+					}
 					continue
 				}
-					log.Printf("source: %v", err)
-					cancel()
-					return
-				}
 				if f.Samples != nil && audioStream != 0 {
 					sdl.PutAudioStreamData(audioStream, interleaveAudio(f.Samples))
 				}
@@ -447,11 +500,30 @@ func main() {
 
 		// Video (with or without sync) mode: grant/staged handshake.
 		for {
+			params := <-control
+			if params.video != "" || params.audio != "" {
+				// Reconnect request from Connect button or auto-retry.
 				select {
-			case <-grant:
+				case <-control: // drain any pending grant
+				default:
+				}
+				if rerr := reopen(params); rerr != nil {
+					if errors.Is(rerr, context.Canceled) {
+						return
+					}
+					log.Printf("source: reopen failed: %v, retrying", rerr)
+					select {
+					case <-time.After(500 * time.Millisecond):
 					case <-ctx.Done():
 						return
 					}
+					select {
+					case control <- params:
+					default:
+					}
+				}
+				continue
+			}
 
 			var payload []byte
 			var grainIdx uint64
@@ -462,24 +534,19 @@ func main() {
 					if errors.Is(err, context.Canceled) {
 						return
 					}
-					if errors.Is(err, mxl.ErrFlowInvalid) {
+					log.Printf("source: %v", err)
-						log.Printf("source: flow invalid, reopening")
+					// Drain any pending grant, then send reconnect.
-						if rerr := reopen(); rerr != nil {
+					select {
-							log.Printf("source: reopen failed: %v", rerr)
+					case <-control:
-							cancel()
+					default:
-							return
 					}
 					select {
-						case failed <- struct{}{}:
+					case control <- reconnectParams{domain: domainStr, video: videoStr, audio: audioStr}:
 					case <-ctx.Done():
 						return
 					}
 					continue
 				}
-					log.Printf("source: %v", err)
-					cancel()
-					return
-				}
 				payload = vFrame.Payload
 				grainIdx = vFrame.Index
 				if aFrame.Samples != nil && audioStream != 0 {
@@ -492,27 +559,20 @@ func main() {
 					if errors.Is(err, context.Canceled) {
 						return
 					}
-					if errors.Is(err, mxl.ErrFlowInvalid) {
+					log.Printf("source: %v", err)
-						log.Printf("source: flow invalid, reopening")
+					select {
-						if rerr := reopen(); rerr != nil {
+					case <-control:
-							log.Printf("source: reopen failed: %v", rerr)
+					default:
-							cancel()
-							return
 					}
 					select {
-						case failed <- struct{}{}:
+					case control <- reconnectParams{domain: domainStr, video: videoStr, audio: audioStr}:
 					case <-ctx.Done():
 						return
 					}
 					continue
 				}
-					log.Printf("source: %v", err)
-					cancel()
-					return
-				}
 				payload = f.Payload
 				grainIdx = f.Index
-
 			}
 
 			if r != nil {
@@ -540,11 +600,6 @@ func main() {
 		frameCount uint64
 		lastReport time.Time
 		lastFrame  time.Time
-		// ImGui stats
-		domainStr string = args.Domain
-		videoStr  string = args.VideoFlowId
-		audioStr  string = args.AudioFlowId
-		showStats bool   = true
 	)
 	lastFrame = time.Now()
 
@@ -559,8 +614,13 @@ func main() {
 			case sdl.EventWindowResized, sdl.EventPixelSizeChanged:
 				resized = true
 			case sdl.EventKeyDown:
+				if gui.IO().WantCaptureKeyboard() {
+					break
+				}
 				key := *(*int32)(unsafe.Pointer(&event[28]))
 				switch uint32(key) {
+				case sdl.KeyQ:
+					fallthrough
 				case sdl.KeyEscape:
 					running = false
 				case sdl.KeyF:
@@ -588,7 +648,7 @@ func main() {
 		}
 		if !granted {
 			select {
-			case grant <- struct{}{}:
+			case control <- reconnectParams{}:
 				granted = true
 			case <-ctx.Done():
 				running = false
@@ -596,20 +656,21 @@ func main() {
 			}
 		}
 		var shownIndex uint64
+		hasFrame := false
 		select {
 		case shownIndex = <-staged:
 			granted = false
-		case <-failed:
+			hasFrame = true
-			granted = false
-			continue
 		case <-ctx.Done():
 			running = false
 			continue
 		case <-time.After(100 * time.Millisecond):
-			continue
+			// No frame staged. Reset granted so we re-grant on next iteration.
+			granted = false
 		}
 
 		// stats
+		if hasFrame {
 			if lastIndex != 0 && shownIndex > lastIndex {
 				if g := shownIndex - lastIndex - 1; g > 0 {
 					dropped += g
@@ -626,6 +687,7 @@ func main() {
 				dropped = 0
 				lastReport = now
 			}
+		}
 		// end of stats
 		if r != nil {
 			gui.BeginFrame(time.Since(lastFrame), int32(r.Extent().Width), int32(r.Extent().Height))
@@ -633,17 +695,24 @@ func main() {
 			// cimgui.Begin("Test")
 			if showStats {
 				cimgui.SetNextWindowPos(cimgui.Vec2{X: 10, Y: 10})
-				cimgui.SetNextWindowSize(cimgui.Vec2{X: 300, Y: 300})
+				cimgui.SetNextWindowSize(cimgui.Vec2{X: 200, Y: 200})
 				cimgui.BeginV("Stats", &showStats,
-					cimgui.WindowFlagsNoTitleBar|cimgui.WindowFlagsNoBackground|cimgui.WindowFlagsNoResize|cimgui.WindowFlagsNoScrollbar)
+					cimgui.WindowFlagsNoTitleBar|cimgui.WindowFlagsNoResize|cimgui.WindowFlagsNoScrollbar)
 				cimgui.Text(fmt.Sprintf("FPS: %.1f", fps))
 				cimgui.Text(fmt.Sprintf("Dropped: %d", dropped))
 				cimgui.Text(fmt.Sprintf("Index: %d", shownIndex))
+				if videoSrc != nil {
+					cimgui.Text(fmt.Sprintf("Video: %dx%d %.2fp", videoSrc.Width(), videoSrc.Height(),
+						float32(videoSrc.Rate().Num/videoSrc.Rate().Den)))
+				}
 				if syncSrc != nil {
 					cimgui.Text(fmt.Sprintf("Video: %dx%d %.2fp", syncSrc.Width(), syncSrc.Height(),
-						float32(syncSrc.Rate().Num/syncSrc.AudioRate().Den)))
+						float32(syncSrc.Rate().Num/syncSrc.Rate().Den)))
 					cimgui.Text(fmt.Sprintf("Audio: %dch %dkHz", syncSrc.Channels(), syncSrc.AudioRate().Num))
 				}
+				cimgui.Text("\nPress F1 to hide stats")
+				cimgui.Text("Q or Esc to quit")
+				cimgui.Text("F for fullscreen")
 				cimgui.End()
 			}
 			cimgui.Begin("Connection")
@@ -651,6 +720,9 @@ func main() {
 			cimgui.InputTextWithHint("Video UUID", "", &videoStr, 0, nil)
 			cimgui.InputTextWithHint("Audio UUID", "", &audioStr, 0, nil)
 			cimgui.Checkbox("Show stats", &showStats)
+			if cimgui.Button("Connect") {
+				doReconnect()
+			}
 			cimgui.End()
 			gui.EndFrame()
 			lastFrame = time.Now()

fixes.md

@@ -1,14 +1,23 @@
+# Useful links
+https://pthom.github.io/imgui_explorer/
+
 # List of bugs, that need to be fixed
 
 ## Major
+- check how it looks like with more than 2 audio channels
+
 ## Minor
 - hotkeys on keyDown, especially toggle keys. Way to recreate: press F and hold
 
 ## TODO:
+- some sort of playlist with id's
 - CLI option to run fullscreen
 - fabrics bridge reader. Step by step. Start with local
 - basic UI: stats, fields for domain, flow ids, label, etc.
 - snapshot
 - waveform, vectorscope
 - some image, when audio only
-- q for quit
+
+## Done
+- [x] resize broken again
+- [x] q for quit

imgui.ini

@@ -10,11 +10,11 @@ Collapsed=0
 
 [Window][Stats]
 Pos=10,10
-Size=300,300
+Size=200,200
 Collapsed=0
 
 [Window][Connection]
-Pos=676,497
+Pos=425,351
-Size=413,137
+Size=523,153
 Collapsed=0
 

internal/sdl/sdl.go

@@ -27,8 +27,10 @@ const (
 
 	EventTextInput uint32 = 0x303
 
+	// it's about SDL_keycode, not SDL_scancode
 	KeyEscape uint32 = 0x1B
 	KeyF      uint32 = 0x66
+	KeyQ      uint32 = 0x71
 	KeyF1     uint32 = 0x4000003A
 
 	InitAudio                  uint32 = 0x00000010

internal/source/source.go

@@ -403,6 +403,7 @@ func (s *SyncSource) Close() error {
 
 // NextSync reads both at a synced timestamp. Returns video Frame + audio AudioFrame
 func (s *SyncSource) NextSync(ctx context.Context, audioBatch uint64, timeout time.Duration) (Frame, AudioFrame, error) {
+	var timeouts int
 	for {
 		select {
 		case <-ctx.Done():
@@ -421,7 +422,19 @@ func (s *SyncSource) NextSync(ctx context.Context, audioBatch uint64, timeout ti
 					s.idx = mxl.CurrentIndex(s.rate)
 					continue
 				}
-				return Frame{}, AudioFrame{}, fmt.Errorf("GetGraing: %w", gerr)
+				if errors.Is(gerr, mxl.ErrOutOfRangeLate) {
+					s.idx = mxl.CurrentIndex(s.rate)
+					continue
+				}
+				if errors.Is(gerr, mxl.ErrOutOfRangeEarly) {
+					select {
+					case <-time.After(5 * time.Millisecond):
+					case <-ctx.Done():
+						return Frame{}, AudioFrame{}, ctx.Err()
+					}
+					continue
+				}
+				return Frame{}, AudioFrame{}, fmt.Errorf("GetGrain: %w", gerr)
 			}
 			// read audio at the same timestamp
 			aIdx := mxl.TimestampToIndex(s.aRate, ts)
@@ -451,14 +464,19 @@ func (s *SyncSource) NextSync(ctx context.Context, audioBatch uint64, timeout ti
 			}
 			// even if audio failed, video returns
 			return vFrame, aFrame, nil
-		case errors.Is(err, mxl.ErrTimeout), errors.Is(err, mxl.ErrOutOfRangeEarly):
+		case errors.Is(err, mxl.ErrTimeout), errors.Is(err, mxl.ErrOutOfRangeEarly), errors.Is(err, mxl.ErrOutOfRangeLate):
+			timeouts++
+			if timeouts > 10 {
+				timeouts = 0
+				s.idx = mxl.CurrentIndex(s.rate)
+				return Frame{}, AudioFrame{}, fmt.Errorf("sync: feeds not responding")
+			}
+			s.idx = mxl.CurrentIndex(s.rate)
 			select {
 			case <-time.After(5 * time.Millisecond):
 			case <-ctx.Done():
 				return Frame{}, AudioFrame{}, ctx.Err()
 			}
-		case errors.Is(err, mxl.ErrOutOfRangeLate):
-			s.idx = mxl.CurrentIndex(s.rate)
 		default:
 			return Frame{}, AudioFrame{}, fmt.Errorf("WaitForDataAt: %w", err)
 		}

mxl-gst-scripts/imgui.ini

@@ -0,0 +1,15 @@
+[Window][Debug##Default]
+Pos=60,60
+Size=400,400
+Collapsed=0
+
+[Window][Stats]
+Pos=10,10
+Size=200,200
+Collapsed=0
+
+[Window][Connection]
+Pos=60,60
+Size=110,146
+Collapsed=0
+

mxl-gst-scripts/loop-feed2.sh

@@ -0,0 +1,12 @@
+#!/bin/bash
+VIDEO_ID="5fbec3b1-1b0f-417d-9059-8b94a47197ef"
+AUDIO_ID="5fbec3b1-1b0f-417d-9059-8b94a47197eb"
+VIDEO_URI=$1
+if [[ -z "${VIDEO_URI}" ]] then
+    VIDEO_URI="${HOME}/Videos/test-vid/f1.ts"
+fi
+export GST_PLUGIN_PATH="${HOME}/.gst-plugin:${GST_PLUGIN_PATH}"
+mxl-gst-looping-filesrc -d /dev/shm/mxl -i "${VIDEO_URI}" --video-id "${VIDEO_ID}" --audio-id "${AUDIO_ID}" 2>/dev/null
+# sleep 5
+# kill -9 $(pidof "mxl-gst-looping-filesrc")
+# echo -e "\ngst-looping-filesrc killed"

mxl-gst-scripts/test-reconnect.sh

@@ -0,0 +1,20 @@
+#!/bin/bash
+#!/bin/bash
+VIDEO_ID="5fbec3b1-1b0f-417d-9059-8b94a47197ef"
+AUDIO_ID="5fbec3b1-1b0f-417d-9059-8b94a47197eb"
+VIDEO_URI=$1
+if [[ -z "${VIDEO_URI}" ]] then
+    VIDEO_URI="${HOME}/Videos/test-vid/f1.ts"
+fi
+export GST_PLUGIN_PATH="${HOME}/.gst-plugin:${GST_PLUGIN_PATH}"
+mxl-gst-looping-filesrc -d /dev/shm/mxl -i "${VIDEO_URI}" --video-id "${VIDEO_ID}" --audio-id "${AUDIO_ID}" 2>/dev/null &
+sleep 1
+echo "Writer has been started"
+go run ../cmd/mxl-player -d /dev/shm/mxl -v "${VIDEO_ID}" -a "${AUDIO_ID}" &> /tmp/player.log &
+echo "mxl-player has been started"
+sleep 5
+pkill mxl-gst
+echo "Writer stopped" >> /tmp/player.log
+sleep 5
+echo "Writer has been started again" >> /tmp/player.log
+mxl-gst-looping-filesrc -d /dev/shm/mxl -i "${VIDEO_URI}" --video-id "${VIDEO_ID}" --audio-id "${AUDIO_ID}" 2>/dev/null