From 8d4e4c2e847363cd6cc0a366f7df14ca98946117 Mon Sep 17 00:00:00 2001 From: QiuSW <105186638@qq.com> Date: Thu, 13 Aug 2026 14:57:56 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E4=BF=AE=E5=A4=8D=E5=AE=9E=E6=97=B6?= =?UTF-8?q?=E7=9B=91=E7=9C=8B=E6=8C=89=E9=9C=80=E6=8B=89=E6=B5=81=E5=BE=AA?= =?UTF-8?q?=E7=8E=AF=E7=AD=89=E5=BE=85=20(#54)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- Sense/server/app/sense/liveview/service.go | 5 +-- .../server/app/sense/liveview/service_test.go | 24 +++++++++++++ Sense/server/app/sense/media/route_port.go | 7 ++++ Sense/server/app/sense/media/service.go | 34 +++++++++++++++++++ Sense/server/app/sense/media/service_test.go | 24 +++++++++++++ .../sense/liveview/StreamPlayer.vue | 8 ++--- 6 files changed, 96 insertions(+), 6 deletions(-) diff --git a/Sense/server/app/sense/liveview/service.go b/Sense/server/app/sense/liveview/service.go index 454ecd7..2270752 100644 --- a/Sense/server/app/sense/liveview/service.go +++ b/Sense/server/app/sense/liveview/service.go @@ -47,6 +47,7 @@ type Service struct { base *url.URL ttl time.Duration route func(context.Context, string) (media.Route, error) + refresh func(context.Context, string) (media.Route, error) routes func(context.Context) ([]media.Route, error) mu sync.RWMutex sessions map[string]Session @@ -64,7 +65,7 @@ func NewService(rawBase string, ttl time.Duration) (*Service, error) { if ttl <= 0 || ttl > 10*time.Minute { ttl = 2 * time.Minute } - return &Service{base: parsed, ttl: ttl, route: media.PlaybackRoute, routes: media.PlaybackRoutes, sessions: map[string]Session{}, now: time.Now}, nil + return &Service{base: parsed, ttl: ttl, route: media.PlaybackRoute, refresh: media.RefreshPlaybackRoute, routes: media.PlaybackRoutes, sessions: map[string]Session{}, now: time.Now}, nil } func (s *Service) Routes(ctx context.Context) ([]Route, error) { items, err := s.routes(ctx) @@ -117,7 +118,7 @@ func (s *Service) Get(ctx context.Context, owner, id string) (Session, error) { if !ok || session.OwnerID != owner || !s.now().Before(session.ExpiresAt) { return Session{}, fmt.Errorf("playback session expired") } - route, err := s.route(ctx, session.RouteID) + route, err := s.refresh(ctx, session.RouteID) if err != nil { return Session{}, err } diff --git a/Sense/server/app/sense/liveview/service_test.go b/Sense/server/app/sense/liveview/service_test.go index 7bd13f9..5a3e5a9 100644 --- a/Sense/server/app/sense/liveview/service_test.go +++ b/Sense/server/app/sense/liveview/service_test.go @@ -44,3 +44,27 @@ func TestTTLIsBounded(t *testing.T) { t.Fatalf("ttl=%v", service.ttl) } } + +func TestSessionPollRefreshesOnDemandMediaState(t *testing.T) { + service, err := NewService("http://127.0.0.1:8889", time.Minute) + if err != nil { + t.Fatal(err) + } + route := media.Route{ID: "device:main", DeviceID: "device", ProfileToken: "main", Path: "sense_device_main", Desired: "running", Actual: "waiting"} + service.route = func(context.Context, string) (media.Route, error) { return route, nil } + service.refresh = func(context.Context, string) (media.Route, error) { + updated := route + updated.Actual = "ready" + updated.Detail = "上游拉流正常" + updated.Readers = 1 + return updated, nil + } + session, err := service.Create(context.Background(), "owner", route.ID) + if err != nil || session.Status != "waiting" || session.PlayerURL == "" { + t.Fatalf("session=%#v err=%v", session, err) + } + updated, err := service.Get(context.Background(), "owner", session.ID) + if err != nil || updated.Status != "ready" || updated.Detail != "上游拉流正常" { + t.Fatalf("updated=%#v err=%v", updated, err) + } +} diff --git a/Sense/server/app/sense/media/route_port.go b/Sense/server/app/sense/media/route_port.go index 024f88f..1ad9cff 100644 --- a/Sense/server/app/sense/media/route_port.go +++ b/Sense/server/app/sense/media/route_port.go @@ -19,6 +19,13 @@ func PlaybackRoute(ctx context.Context, id string) (Route, error) { } return service.store.Get(ctx, id) } +func RefreshPlaybackRoute(ctx context.Context, id string) (Route, error) { + service := activeService.Load() + if service == nil { + return Route{}, fmt.Errorf("media service is not ready") + } + return service.Refresh(ctx, id) +} func PlaybackRoutes(ctx context.Context) ([]Route, error) { service := activeService.Load() if service == nil { diff --git a/Sense/server/app/sense/media/service.go b/Sense/server/app/sense/media/service.go index 666009c..513fb10 100644 --- a/Sense/server/app/sense/media/service.go +++ b/Sense/server/app/sense/media/service.go @@ -125,6 +125,40 @@ func (s *Service) Reconcile(ctx context.Context, actor identity.Principal, id st identity.RecordAudit(ctx, actor.UserID, "media.reconcile", id, "success", map[string]any{"actual": route.Actual}) return route, nil } + +// Refresh reads the MediaMTX runtime state without reapplying configuration or +// exposing the source URI. It is safe to call while a playback session polls. +func (s *Service) Refresh(ctx context.Context, id string) (Route, error) { + route, err := s.store.Get(ctx, id) + if err != nil { + return Route{}, err + } + if route.Desired != "running" { + return route, nil + } + status, err := s.controller.Status(ctx, route.Path) + if err != nil { + return route, nil + } + actual := "waiting" + detail := "等待播放器连接并按需拉流" + if status.Ready { + actual = "ready" + detail = "上游拉流正常" + } + if route.Actual == actual && route.Detail == detail && route.Readers == status.Readers { + return route, nil + } + route.Actual = actual + route.Detail = detail + route.Readers = status.Readers + route.Version++ + route.UpdatedAt = s.now().UTC() + if err := s.store.Save(ctx, route); err != nil { + return Route{}, err + } + return route, nil +} func (s *Service) Stop(ctx context.Context, actor identity.Principal, id string) (Route, error) { route, err := s.store.Get(ctx, id) if err != nil { diff --git a/Sense/server/app/sense/media/service_test.go b/Sense/server/app/sense/media/service_test.go index 612175d..3f19420 100644 --- a/Sense/server/app/sense/media/service_test.go +++ b/Sense/server/app/sense/media/service_test.go @@ -40,6 +40,13 @@ type fakeController struct { status mediamtx.PathStatus } +type refreshController struct{ status mediamtx.PathStatus } + +func (f refreshController) Apply(context.Context, mediamtx.Source) error { return nil } +func (f refreshController) Status(context.Context, string) (mediamtx.PathStatus, error) { + return f.status, nil +} + func (f fakeController) Apply(context.Context, mediamtx.Source) error { return f.applyErr } func (f fakeController) Status(context.Context, string) (mediamtx.PathStatus, error) { if f.status.Name == "error" { @@ -142,4 +149,21 @@ func TestConfigureProfilesIsIdempotentAndReportsMediaState(t *testing.T) { } } +func TestRefreshTracksOnDemandReaderWithoutReapplyingRoute(t *testing.T) { + store := NewMemoryStore() + route := Route{ID: "device:main", DeviceID: "device", ProfileToken: "main", Path: "sense_device_main", Desired: "running", Actual: "waiting", Detail: "等待上游拉流", Version: 2} + if err := store.Save(context.Background(), route); err != nil { + t.Fatal(err) + } + service := NewService(store, &fakeProcess{}, refreshController{status: mediamtx.PathStatus{Name: route.Path, Ready: true, Readers: 1}}) + result, err := service.Refresh(context.Background(), route.ID) + if err != nil || result.Actual != "ready" || result.Readers != 1 || result.Detail != "上游拉流正常" { + t.Fatalf("result=%#v err=%v", result, err) + } + unchanged, err := service.Refresh(context.Background(), route.ID) + if err != nil || unchanged.Version != result.Version { + t.Fatalf("unchanged=%#v err=%v", unchanged, err) + } +} + func rtspReady() rtsp.Result { return rtsp.Result{Status: "ready"} } diff --git a/Sense/ui/src/components/sense/liveview/StreamPlayer.vue b/Sense/ui/src/components/sense/liveview/StreamPlayer.vue index 3dd5e38..fdfc44d 100644 --- a/Sense/ui/src/components/sense/liveview/StreamPlayer.vue +++ b/Sense/ui/src/components/sense/liveview/StreamPlayer.vue @@ -1,14 +1,14 @@