kanshan/speedtest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: add queued test waiting

Browse Source
This commit is contained in:
Test User
2026-06-06 03:48:59 +08:00
parent c14cea9e72
commit fba25d0e7a
9 changed files with 425 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
README.md
+2 -2
View File
@@ -5,7 +5,7 @@ NetStable 是一个独立的 Go Web 测速项目，用于测试“打开网页
## 功能
- 同一时间只允许一个用户真正进行测速。
- 其他用户发起请求时返回“有其他用户正在测速，请稍等”。
- 其他用户发起请求时会进入 FIFO 队列，页面和 CLI 会显示当前排第几位，并在前面的用户完成后自动开始。
- 启动时可设置程序理论最高带宽，例如 `-bandwidth-limit-mbps 30` 将测试流量限制到 30 Mbps。
- 页面展示当前设置的理论最高限值。
- 一次测试按阶段执行：先连续下载指定时长，再连续上传同样时长。页面固定提供 `15s` 和 `30s` 两档，默认 `30s`。
@@ -94,7 +94,7 @@ ssh root@SERVER 'systemctl daemon-reload && systemctl enable --now netstable'
## CLI 测试
同一个二进制也可以作为客户端，在其他服务器上直接测试到部署节点的上传、下载和延迟。CLI 会复用 Web API，所以仍然遵守“同一时间只允许一个用户测速”的锁。
同一个二进制也可以作为客户端，在其他服务器上直接测试到部署节点的上传、下载和延迟。CLI 会复用 Web API，所以仍然遵守“同一时间只允许一个用户测速”的锁；如果前面有人正在测，CLI 会显示当前排队位置并持续等待，轮到自己后自动开始。
HTTP 模式不依赖外部程序：
cmd/netstable/main.go
+3
View File
@@ -141,6 +141,9 @@ func runClient(args []string) {
Timeout: time.Duration(*timeoutMillis) * time.Millisecond,
DownloadBytes: *downloadBytes,
UploadBytes: *uploadBytes,
OnQueueWaiting: func(message string) {
_, _ = fmt.Fprintln(os.Stderr, message)
},
}
if !*quiet && !*jsonOutput {
options.OnSample = func(sample probe.Sample) {
docs/release-template.md
+1 -1
View File
@@ -40,7 +40,7 @@ shasum -a 256 -c checksums.txt
## 主要功能
- 单用户实时测速锁。
- 单用户实时测速锁和 FIFO 队列，排队用户可看到当前位置。
- 理论最高带宽限制和页面展示。
- 保存脱敏后的用户 IP、地区、运营商、测试摘要和完整样本曲线。
- SSE 心跳辅助连接用于显示连接状态。
internal/client/client.go
+93 -11
View File
@@ -22,26 +22,37 @@ const (
)
type Options struct {
ServerURL string
Target string
Duration time.Duration
Timeout time.Duration
DownloadBytes int
UploadBytes int
OnSample func(probe.Sample)
ServerURL string
Target string
Duration time.Duration
Timeout time.Duration
DownloadBytes int
UploadBytes int
QueueInterval time.Duration
OnSample func(probe.Sample)
OnQueueWaiting func(string)
}
type createResponse struct {
ID string `json:"id"`
DownloadURL string `json:"downloadUrl"`
UploadURL string `json:"uploadUrl"`
CompleteURL string `json:"completeUrl"`
ID string `json:"id"`
Status string `json:"status"`
QueuePosition int `json:"queuePosition"`
QueueURL string `json:"queueUrl"`
DownloadURL string `json:"downloadUrl"`
UploadURL string `json:"uploadUrl"`
CompleteURL string `json:"completeUrl"`
}
type completeResponse struct {
Summary probe.Summary `json:"summary"`
}
type queueResponse struct {
ID string `json:"id"`
Status string `json:"status"`
QueuePosition int `json:"queuePosition"`
}
type errorResponse struct {
Error string `json:"error"`
}
@@ -57,6 +68,9 @@ func Run(ctx context.Context, options Options) (probe.Summary, error) {
if err != nil {
return probe.Summary{}, err
}
if err := waitForQueue(ctx, &httpClient, server, session, options); err != nil {
return probe.Summary{}, err
}
var samples []probe.Sample
record := func(sample probe.Sample) {
@@ -108,6 +122,9 @@ func normalizeOptions(options Options) (Options, *url.URL, error) {
if options.UploadBytes <= 0 {
options.UploadBytes = DefaultUploadBytes
}
if options.QueueInterval <= 0 {
options.QueueInterval = 5 * time.Second
}
return options, server, nil
}
@@ -147,6 +164,71 @@ func createSession(ctx context.Context, httpClient *http.Client, server *url.URL
return createResponse{}, fmt.Errorf("create test session: %s", message)
}
func waitForQueue(ctx context.Context, httpClient *http.Client, server *url.URL, session createResponse, options Options) error {
if session.QueuePosition <= 0 && session.Status != "queued" {
return nil
}
queueURL := session.QueueURL
if strings.TrimSpace(queueURL) == "" {
return errors.New("server queued the test without a queue status URL")
}
position := session.QueuePosition
for {
if options.OnQueueWaiting != nil {
options.OnQueueWaiting(queueMessage(position))
}
timer := time.NewTimer(options.QueueInterval)
select {
case <-ctx.Done():
timer.Stop()
return ctx.Err()
case <-timer.C:
}
status, err := fetchQueueStatus(ctx, httpClient, server, queueURL)
if err != nil {
return err
}
if status.Status == "ready" || status.QueuePosition <= 0 {
return nil
}
position = status.QueuePosition
}
}
func fetchQueueStatus(ctx context.Context, httpClient *http.Client, server *url.URL, path string) (queueResponse, error) {
request, err := http.NewRequestWithContext(ctx, http.MethodGet, resolve(server, path), nil)
if err != nil {
return queueResponse{}, err
}
response, err := httpClient.Do(request)
if err != nil {
return queueResponse{}, err
}
defer response.Body.Close()
if response.StatusCode < 200 || response.StatusCode >= 300 {
message := decodeError(response.Body)
if message == "" {
message = response.Status
}
return queueResponse{}, fmt.Errorf("queue status: %s", message)
}
var status queueResponse
if err := json.NewDecoder(response.Body).Decode(&status); err != nil {
return queueResponse{}, err
}
return status, nil
}
func queueMessage(position int) string {
if position <= 0 {
return "正在排队，等待其他用户完成..."
}
return fmt.Sprintf("正在排队，当前排第 %d 位，等待其他用户完成...", position)
}
func runPhase(ctx context.Context, duration time.Duration, measure func() probe.Sample, record func(probe.Sample)) error {
deadline := time.Now().Add(duration)
samples := 0
internal/client/client_test.go
+40 -20
View File
@@ -50,30 +50,36 @@ func TestRunCompletesBrowserCompatibleSpeedTest(t *testing.T) {
}
}
func TestRunReturnsBusyError(t *testing.T) {
server := appweb.New(appweb.Options{Store: &memoryStore{}, Runner: noopRunner})
func TestRunWaitsInQueueUntilBusySessionCompletes(t *testing.T) {
store := &memoryStore{}
server := appweb.New(appweb.Options{Store: store, Runner: noopRunner})
httpServer := httptest.NewServer(server.Handler())
defer httpServer.Close()
_, err := Run(context.Background(), Options{
ServerURL: httpServer.URL,
Duration: 20 * time.Millisecond,
DownloadBytes: 1024,
UploadBytes: 512,
busy := createBusySession(t, httpServer.URL)
go func() {
time.Sleep(30 * time.Millisecond)
completeBusySession(t, httpServer.URL, busy)
}()
var queueMessages int
summary, err := Run(context.Background(), Options{
ServerURL: httpServer.URL,
Duration: 20 * time.Millisecond,
Timeout: time.Second,
DownloadBytes: 1024,
UploadBytes: 512,
QueueInterval: 10 * time.Millisecond,
OnQueueWaiting: func(message string) { queueMessages++ },
})
if err != nil {
t.Fatalf("first Run returned error: %v", err)
t.Fatalf("Run returned error: %v", err)
}
createBusySession(t, httpServer.URL)
_, err = Run(context.Background(), Options{
ServerURL: httpServer.URL,
Duration: 20 * time.Millisecond,
DownloadBytes: 1024,
UploadBytes: 512,
})
if err == nil || !strings.Contains(err.Error(), "正在测速") {
t.Fatalf("busy error = %v, want Chinese busy message", err)
if summary.ID == "" || summary.AvgDownloadMbps <= 0 || summary.AvgUploadMbps <= 0 {
t.Fatalf("summary = %#v, want completed queued test", summary)
}
if queueMessages == 0 {
t.Fatal("OnQueueWaiting was not called while queued")
}
}
@@ -120,14 +126,14 @@ func TestRunIperf3FailedCommandReleasesOneOffServerSession(t *testing.T) {
createBusySession(t, httpServer.URL)
}
func createBusySession(t *testing.T, serverURL string) {
func createBusySession(t *testing.T, serverURL string) createResponse {
t.Helper()
server, err := url.Parse(serverURL)
if err != nil {
t.Fatalf("parse server URL: %v", err)
}
httpClient := http.Client{Timeout: time.Second}
_, err = createSession(context.Background(), &httpClient, server, Options{
session, err := createSession(context.Background(), &httpClient, server, Options{
Target: server.Host,
Duration: time.Second,
Timeout: time.Second,
@@ -137,6 +143,20 @@ func createBusySession(t *testing.T, serverURL string) {
if err != nil {
t.Fatalf("create busy session: %v", err)
}
return session
}
func completeBusySession(t *testing.T, serverURL string, session createResponse) {
t.Helper()
server, err := url.Parse(serverURL)
if err != nil {
t.Fatalf("parse server URL: %v", err)
}
httpClient := http.Client{Timeout: time.Second}
now := time.Now().UTC()
if _, err := completeSession(context.Background(), &httpClient, server, session.CompleteURL, now, now, nil); err != nil {
t.Errorf("complete busy session: %v", err)
}
}
func hasKind(samples []probe.Sample, kind string) bool {
internal/web/server.go
+133 -21
View File
@@ -60,6 +60,7 @@ type Server struct {
mu sync.Mutex
sessions map[string]TestRequest
iperfSessions map[string]*iperfSession
queue []string
activeID string
}
@@ -125,11 +126,20 @@ type completeIperf3Request struct {
}
type createTestResponse struct {
ID string `json:"id"`
EventsURL string `json:"eventsUrl"`
DownloadURL string `json:"downloadUrl"`
UploadURL string `json:"uploadUrl"`
CompleteURL string `json:"completeUrl"`
ID string `json:"id"`
Status string `json:"status"`
QueuePosition int `json:"queuePosition"`
QueueURL string `json:"queueUrl"`
EventsURL string `json:"eventsUrl"`
DownloadURL string `json:"downloadUrl"`
UploadURL string `json:"uploadUrl"`
CompleteURL string `json:"completeUrl"`
}
type queueStatusResponse struct {
ID string `json:"id"`
Status string `json:"status"`
QueuePosition int `json:"queuePosition"`
}
type transferResponse struct {
@@ -665,30 +675,35 @@ func (s *Server) handleCreateTest(w http.ResponseWriter, r *http.Request) {
s.mu.Lock()
s.expireActiveLocked(time.Now().UTC())
status := "ready"
position := 0
if s.activeID != "" {
activeID := s.activeID
s.mu.Unlock()
writeJSON(w, http.StatusConflict, busyResponse{
Error: "有其他用户正在测速，请稍等。",
ActiveTestID: activeID,
})
return
status = "queued"
s.queue = append(s.queue, request.ID)
position = queuePositionLocked(s.queue, request.ID)
}
s.sessions[request.ID] = request
s.activeID = request.ID
if status == "ready" {
s.activeID = request.ID
}
s.mu.Unlock()
writeJSON(w, http.StatusAccepted, createTestResponse{
ID: request.ID,
EventsURL: "/api/tests/" + request.ID + "/events",
DownloadURL: "/api/download?testId=" + request.ID,
UploadURL: "/api/upload?testId=" + request.ID,
CompleteURL: "/api/tests/" + request.ID + "/complete",
ID: request.ID,
Status: status,
QueuePosition: position,
QueueURL: "/api/tests/" + request.ID + "/queue",
EventsURL: "/api/tests/" + request.ID + "/events",
DownloadURL: "/api/download?testId=" + request.ID,
UploadURL: "/api/upload?testId=" + request.ID,
CompleteURL: "/api/tests/" + request.ID + "/complete",
})
}
func (s *Server) handleTestRoute(w http.ResponseWriter, r *http.Request) {
switch {
case strings.HasSuffix(r.URL.Path, "/queue"):
s.handleQueueStatus(w, r)
case strings.HasSuffix(r.URL.Path, "/events"):
s.handleTestEvents(w, r)
case strings.HasSuffix(r.URL.Path, "/complete"):
@@ -698,6 +713,28 @@ func (s *Server) handleTestRoute(w http.ResponseWriter, r *http.Request) {
}
}
func (s *Server) handleQueueStatus(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
writeJSON(w, http.StatusMethodNotAllowed, errorResponse{Error: "method not allowed"})
return
}
id, ok := queueSessionID(r.URL.Path)
if !ok {
writeJSON(w, http.StatusNotFound, errorResponse{Error: "not found"})
return
}
s.mu.Lock()
s.expireActiveLocked(time.Now().UTC())
response, exists := s.queueStatusLocked(id)
s.mu.Unlock()
if !exists {
writeJSON(w, http.StatusNotFound, errorResponse{Error: "test session not found"})
return
}
writeJSON(w, http.StatusOK, response)
}
func (s *Server) handleCompleteTest(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodPost {
writeJSON(w, http.StatusMethodNotAllowed, errorResponse{Error: "method not allowed"})
@@ -752,10 +789,21 @@ func (s *Server) handleTestEvents(w http.ResponseWriter, r *http.Request) {
}
s.mu.Lock()
request, exists := s.sessions[id]
if exists {
delete(s.sessions, id)
_, exists := s.sessions[id]
active := s.activeID == id
s.mu.Unlock()
if !exists {
writeJSON(w, http.StatusNotFound, errorResponse{Error: "test session not found"})
return
}
if !active {
writeJSON(w, http.StatusConflict, errorResponse{Error: "该测速会话正在排队，请等待轮到后再开始。"})
return
}
s.mu.Lock()
request, exists := s.sessions[id]
delete(s.sessions, id)
s.mu.Unlock()
if !exists {
writeJSON(w, http.StatusNotFound, errorResponse{Error: "test session not found"})
@@ -872,6 +920,7 @@ func (s *Server) releaseActive(id string) {
defer s.mu.Unlock()
if s.activeID == id {
s.activeID = ""
s.promoteNextLocked(time.Now().UTC())
}
}
@@ -882,12 +931,14 @@ func (s *Server) takeSession(id string) (TestRequest, bool) {
if exists {
delete(s.sessions, id)
}
s.removeQueuedLocked(id)
if session, exists := s.iperfSessions[id]; exists {
delete(s.iperfSessions, id)
stopIperf3Process(session)
}
if s.activeID == id {
s.activeID = ""
s.promoteNextLocked(time.Now().UTC())
}
return request, exists
}
@@ -902,6 +953,7 @@ func (s *Server) takeIperf3Session(id string) (*iperfSession, bool) {
}
if s.activeID == id {
s.activeID = ""
s.promoteNextLocked(time.Now().UTC())
}
return session, exists
}
@@ -919,6 +971,7 @@ func (s *Server) expireIperf3Session(id string, ttl time.Duration) {
}
if s.activeID == id {
s.activeID = ""
s.promoteNextLocked(time.Now().UTC())
}
s.mu.Unlock()
if exists {
@@ -926,6 +979,53 @@ func (s *Server) expireIperf3Session(id string, ttl time.Duration) {
}
}
func (s *Server) queueStatusLocked(id string) (queueStatusResponse, bool) {
if s.activeID == id {
return queueStatusResponse{ID: id, Status: "ready", QueuePosition: 0}, true
}
if _, exists := s.sessions[id]; !exists {
return queueStatusResponse{}, false
}
position := queuePositionLocked(s.queue, id)
if position > 0 {
return queueStatusResponse{ID: id, Status: "queued", QueuePosition: position}, true
}
return queueStatusResponse{ID: id, Status: "ready", QueuePosition: 0}, true
}
func (s *Server) promoteNextLocked(now time.Time) {
for len(s.queue) > 0 {
id := s.queue[0]
s.queue = s.queue[1:]
request, exists := s.sessions[id]
if !exists {
continue
}
request.CreatedAt = now
s.sessions[id] = request
s.activeID = id
return
}
}
func (s *Server) removeQueuedLocked(id string) {
for index, queuedID := range s.queue {
if queuedID == id {
s.queue = append(s.queue[:index], s.queue[index+1:]...)
return
}
}
}
func queuePositionLocked(queue []string, id string) int {
for index, queuedID := range queue {
if queuedID == id {
return index + 1
}
}
return 0
}
func (s *Server) hasActiveTest() bool {
s.mu.Lock()
defer s.mu.Unlock()
@@ -935,11 +1035,13 @@ func (s *Server) hasActiveTest() bool {
func (s *Server) expireActiveLocked(now time.Time) {
if s.activeID == "" {
s.promoteNextLocked(now)
return
}
request, exists := s.sessions[s.activeID]
if !exists {
s.activeID = ""
s.promoteNextLocked(now)
return
}
if now.Sub(request.CreatedAt) > request.Duration*2+60*time.Second {
@@ -949,6 +1051,7 @@ func (s *Server) expireActiveLocked(now time.Time) {
}
delete(s.sessions, s.activeID)
s.activeID = ""
s.promoteNextLocked(now)
}
}
@@ -1068,6 +1171,15 @@ func completeSessionID(path string) (string, bool) {
return id, id != ""
}
func queueSessionID(path string) (string, bool) {
if !strings.HasPrefix(path, "/api/tests/") || !strings.HasSuffix(path, "/queue") {
return "", false
}
id := strings.TrimSuffix(strings.TrimPrefix(path, "/api/tests/"), "/queue")
id = strings.Trim(id, "/")
return id, id != ""
}
func completeIperf3SessionID(path string) (string, bool) {
if !strings.HasPrefix(path, "/api/iperf3/sessions/") || !strings.HasSuffix(path, "/complete") {
return "", false
internal/web/server_test.go
+102 -10
View File
@@ -279,6 +279,9 @@ func TestCreateTestReturnsSessionIDAndEventsURL(t *testing.T) {
if payload.ID == "" {
t.Fatal("payload.ID is empty")
}
if payload.Status != "ready" || payload.QueuePosition != 0 || payload.QueueURL != "/api/tests/"+payload.ID+"/queue" {
t.Fatalf("queue fields = %q/%d/%q, want ready/0/queue URL", payload.Status, payload.QueuePosition, payload.QueueURL)
}
if payload.EventsURL != "/api/tests/"+payload.ID+"/events" {
t.Fatalf("EventsURL = %q, want /api/tests/%s/events", payload.EventsURL, payload.ID)
}
@@ -293,7 +296,7 @@ func TestCreateTestReturnsSessionIDAndEventsURL(t *testing.T) {
}
}
func TestCreateTestReturnsBusyWhenAnotherSessionIsActive(t *testing.T) {
func TestCreateTestQueuesWhenAnotherSessionIsActive(t *testing.T) {
server := New(Options{Store: &memoryStore{}, Runner: noopRunner})
firstReq := httptest.NewRequest(http.MethodPost, "/api/tests", jsonBody(`{"target":"example.com","durationSeconds":1}`))
@@ -306,11 +309,86 @@ func TestCreateTestReturnsBusyWhenAnotherSessionIsActive(t *testing.T) {
secondReq := httptest.NewRequest(http.MethodPost, "/api/tests", jsonBody(`{"target":"example.org","durationSeconds":1}`))
secondRes := httptest.NewRecorder()
server.Handler().ServeHTTP(secondRes, secondReq)
if secondRes.Code != http.StatusConflict {
t.Fatalf("second status = %d, want 409; body=%s", secondRes.Code, secondRes.Body.String())
if secondRes.Code != http.StatusAccepted {
t.Fatalf("second status = %d, want 202 queued; body=%s", secondRes.Code, secondRes.Body.String())
}
if !strings.Contains(secondRes.Body.String(), "正在测速") {
t.Fatalf("busy response = %q, want Chinese wait message", secondRes.Body.String())
var queued createTestResponse
if err := json.Unmarshal(secondRes.Body.Bytes(), &queued); err != nil {
t.Fatalf("decode queued response: %v", err)
}
if queued.Status != "queued" || queued.QueuePosition != 1 || queued.QueueURL != "/api/tests/"+queued.ID+"/queue" {
t.Fatalf("queued response = %#v, want queued at position 1", queued)
}
}
func TestQueuedTestPromotesAfterActiveSessionCompletes(t *testing.T) {
server := New(Options{Store: &memoryStore{}, Runner: noopRunner})
firstReq := httptest.NewRequest(http.MethodPost, "/api/tests", jsonBody(`{"target":"example.com","durationSeconds":1}`))
firstRes := httptest.NewRecorder()
server.Handler().ServeHTTP(firstRes, firstReq)
if firstRes.Code != http.StatusAccepted {
t.Fatalf("first status = %d, want 202; body=%s", firstRes.Code, firstRes.Body.String())
}
var first createTestResponse
if err := json.Unmarshal(firstRes.Body.Bytes(), &first); err != nil {
t.Fatalf("decode first response: %v", err)
}
secondReq := httptest.NewRequest(http.MethodPost, "/api/tests", jsonBody(`{"target":"example.org","durationSeconds":1}`))
secondRes := httptest.NewRecorder()
server.Handler().ServeHTTP(secondRes, secondReq)
if secondRes.Code != http.StatusAccepted {
t.Fatalf("second status = %d, want 202; body=%s", secondRes.Code, secondRes.Body.String())
}
var second createTestResponse
if err := json.Unmarshal(secondRes.Body.Bytes(), &second); err != nil {
t.Fatalf("decode second response: %v", err)
}
if second.Status != "queued" || second.QueuePosition != 1 {
t.Fatalf("second = %#v, want queued position 1", second)
}
queueReq := httptest.NewRequest(http.MethodGet, second.QueueURL, nil)
queueRes := httptest.NewRecorder()
server.Handler().ServeHTTP(queueRes, queueReq)
if queueRes.Code != http.StatusOK {
t.Fatalf("queue status = %d, want 200; body=%s", queueRes.Code, queueRes.Body.String())
}
var queued queueStatusResponse
if err := json.Unmarshal(queueRes.Body.Bytes(), &queued); err != nil {
t.Fatalf("decode queue status: %v", err)
}
if queued.Status != "queued" || queued.QueuePosition != 1 {
t.Fatalf("queue status = %#v, want queued position 1", queued)
}
completeReq := httptest.NewRequest(http.MethodPost, first.CompleteURL, jsonBody(`{"samples":[]}`))
completeRes := httptest.NewRecorder()
server.Handler().ServeHTTP(completeRes, completeReq)
if completeRes.Code != http.StatusOK {
t.Fatalf("complete status = %d, want 200; body=%s", completeRes.Code, completeRes.Body.String())
}
readyReq := httptest.NewRequest(http.MethodGet, second.QueueURL, nil)
readyRes := httptest.NewRecorder()
server.Handler().ServeHTTP(readyRes, readyReq)
if readyRes.Code != http.StatusOK {
t.Fatalf("ready status = %d, want 200; body=%s", readyRes.Code, readyRes.Body.String())
}
var ready queueStatusResponse
if err := json.Unmarshal(readyRes.Body.Bytes(), &ready); err != nil {
t.Fatalf("decode ready status: %v", err)
}
if ready.Status != "ready" || ready.QueuePosition != 0 {
t.Fatalf("ready status = %#v, want ready position 0", ready)
}
downloadReq := httptest.NewRequest(http.MethodGet, second.DownloadURL+"&bytes=16", nil)
downloadRes := httptest.NewRecorder()
server.Handler().ServeHTTP(downloadRes, downloadReq)
if downloadRes.Code != http.StatusOK || downloadRes.Body.Len() != 16 {
t.Fatalf("download status/body = %d/%d, want 200/16; body=%s", downloadRes.Code, downloadRes.Body.Len(), downloadRes.Body.String())
}
}
@@ -336,8 +414,15 @@ func TestCreateIperf3SessionStartsOneOffServerAndUsesGlobalLock(t *testing.T) {
secondReq := httptest.NewRequest(http.MethodPost, "/api/tests", jsonBody(`{"target":"example.org","durationSeconds":1}`))
secondRes := httptest.NewRecorder()
server.Handler().ServeHTTP(secondRes, secondReq)
if secondRes.Code != http.StatusConflict {
t.Fatalf("second status = %d, want 409 while iperf3 session is active; body=%s", secondRes.Code, secondRes.Body.String())
if secondRes.Code != http.StatusAccepted {
t.Fatalf("second status = %d, want 202 queued while iperf3 session is active; body=%s", secondRes.Code, secondRes.Body.String())
}
var queued createTestResponse
if err := json.Unmarshal(secondRes.Body.Bytes(), &queued); err != nil {
t.Fatalf("decode queued response: %v", err)
}
if queued.Status != "queued" || queued.QueuePosition != 1 {
t.Fatalf("queued response = %#v, want queued position 1", queued)
}
}
@@ -389,7 +474,7 @@ func TestCompleteIperf3SessionPersistsRecordAndReleasesLock(t *testing.T) {
}
}
func TestBrowserSessionExpiresAfterDownloadAndUploadWindows(t *testing.T) {
func TestBrowserSessionQueuesDuringDownloadAndUploadWindows(t *testing.T) {
server := New(Options{Store: &memoryStore{}, Runner: noopRunner})
firstReq := httptest.NewRequest(http.MethodPost, "/api/tests", jsonBody(`{"target":"example.com","durationSeconds":30}`))
@@ -408,8 +493,15 @@ func TestBrowserSessionExpiresAfterDownloadAndUploadWindows(t *testing.T) {
secondReq := httptest.NewRequest(http.MethodPost, "/api/tests", jsonBody(`{"target":"example.org","durationSeconds":1}`))
secondRes := httptest.NewRecorder()
server.Handler().ServeHTTP(secondRes, secondReq)
if secondRes.Code != http.StatusConflict {
t.Fatalf("second status = %d, want 409 while 30s download + 30s upload window is active; body=%s", secondRes.Code, secondRes.Body.String())
if secondRes.Code != http.StatusAccepted {
t.Fatalf("second status = %d, want 202 queued while 30s download + 30s upload window is active; body=%s", secondRes.Code, secondRes.Body.String())
}
var queued createTestResponse
if err := json.Unmarshal(secondRes.Body.Bytes(), &queued); err != nil {
t.Fatalf("decode queued response: %v", err)
}
if queued.Status != "queued" || queued.QueuePosition != 1 {
t.Fatalf("queued response = %#v, want queued position 1", queued)
}
}
web/assets_test.go
+13
View File
@@ -70,6 +70,19 @@ func TestStaticAssetsIncludeCopyableClientCommand(t *testing.T) {
}
}
func TestStaticAssetsIncludeQueueWaitingBehavior(t *testing.T) {
app, err := readAsset("app.js")
if err != nil {
t.Fatalf("read app.js: %v", err)
}
if !contains(app, "waitForBrowserQueue") ||
!contains(app, "queuePosition") ||
!contains(app, "当前排第") ||
!contains(app, "排队中") {
t.Fatalf("app.js missing queue waiting behavior")
}
}
func contains(text, pattern string) bool {
return strings.Contains(text, pattern)
}
web/static/app.js
+38
View File
@@ -126,6 +126,7 @@ async function startTest(event) {
if (!response.ok) {
throw new Error(body.error || "创建测试失败");
}
await waitForBrowserQueue(body);
await runBrowserSpeedTest(body, payload);
} catch (error) {
setServerState("error", "错误");
@@ -134,6 +135,32 @@ async function startTest(event) {
}
}
async function waitForBrowserQueue(session) {
let position = Number(session.queuePosition || 0);
if (session.status !== "queued" && position <= 0) {
return;
}
if (!session.queueUrl) {
throw new Error("服务端没有返回队列状态地址");
}
setServerState("busy", "排队中");
while (true) {
setNotice(queueNotice(position));
await sleep(5000);
const response = await fetch(`${session.queueUrl}?cache=${Date.now()}`, { cache: "no-store" });
const body = await response.json();
if (!response.ok) {
throw new Error(body.error || "读取队列状态失败");
}
position = Number(body.queuePosition || 0);
if (body.status === "ready" || position <= 0) {
setServerState("running", "运行中");
return;
}
}
}
async function runBrowserSpeedTest(session, request) {
const startedAt = new Date();
const durationMs = Math.max(1000, (request.durationSeconds || 30) * 1000);
@@ -540,6 +567,17 @@ function setNotice(message, isError = false) {
els.notice.className = isError ? "notice error" : "notice";
}
function queueNotice(position) {
if (position > 0) {
return `正在排队，当前排第 ${position} 位，等待其他用户完成...`;
}
return "正在排队，等待其他用户完成...";
}
function sleep(ms) {
return new Promise(resolve => window.setTimeout(resolve, ms));
}
function closeSource() {
if (state.source) {
state.source.close();