From e5fc3a8bce10a0ead73d251620f3dd0a3f4fccf6 Mon Sep 17 00:00:00 2001
From: qydysky <32743305+qydysky@users.noreply.github.com>
Date: Sun, 16 Oct 2022 03:20:09 +0800
Subject: [PATCH] fix

---
 websocket/Recoder.go     | 20 +++++++++++++++++---
 websocket/Server_test.go |  3 ++-
 2 files changed, 19 insertions(+), 4 deletions(-)

diff --git a/websocket/Recoder.go b/websocket/Recoder.go
index 0faf427..c09efd9 100644
--- a/websocket/Recoder.go
+++ b/websocket/Recoder.go
@@ -55,6 +55,10 @@ func (t *Recorder) Start(filePath string) error {
 				}
 				return false
 			},
+			`close`: func(data interface{}) bool {
+				t.Stop()
+				return true
+			},
 		})
 		t.stopflag.Wait()
 	}()
@@ -68,9 +72,19 @@ func (t *Recorder) Stop() {
 	t.onlyOnce.UnSet()
 }
 
-func Play(filePath string, perReadSize int, maxReadSize int) (s *Server) {
+func Play(filePath string, perReadSize int, maxReadSize int) (s *Server, close func()) {
+	sg := signal.Init()
+
 	s = New_server()
 
+	close = func() {
+		s.Interface().Push_tag(`close`, uinterface{
+			Id:   0,
+			Data: `rev_close`,
+		})
+		sg.Done()
+	}
+
 	go func() {
 		f := file.New(filePath, 0, false)
 		defer f.Close()
@@ -78,10 +92,10 @@ func Play(filePath string, perReadSize int, maxReadSize int) (s *Server) {
 		startT := time.Now()
 		timer := time.NewTicker(time.Second)
 
-		for {
+		for sg.Islive() {
 			cu := (<-timer.C).Sub(startT).Seconds()
 
-			for {
+			for sg.Islive() {
 				if data, err := f.ReadUntil('\n', perReadSize, maxReadSize); err != nil {
 					panic(err)
 				} else if len(data) != 0 {
diff --git a/websocket/Server_test.go b/websocket/Server_test.go
index ccf3689..a7ec681 100644
--- a/websocket/Server_test.go
+++ b/websocket/Server_test.go
@@ -65,7 +65,8 @@ func Test_Server(t *testing.T) {
 }
 
 func Test_Recoder(t *testing.T) {
-	s := Play("l.csv", 50, 5000)
+	s, cf := Play("l.csv", 50, 5000)
+	defer cf()
 
 	w := web.Easy_boot()
 	open.Run("http://" + w.Server.Addr)
-- 
2.39.2