add gb_test func:TestGbPublishWithApi

Author: duiniuluantanqin

trunk/3rdparty/srs-bench/gb28181/gb28181.go

@@ -88,11 +88,11 @@ func Parse(ctx context.Context) interface{} {
 		os.Exit(0)
 	}
 
-	// showHelp := c.sipConfig.String() == ""
-	// if showHelp {
-	// 	fl.Usage()
-	// 	os.Exit(-1)
-	// }
+	showHelp := c.sipConfig.String() == "" && c.psConfig.addr == ""
+	if showHelp {
+		fl.Usage()
+		os.Exit(-1)
+	}
 
 	summaryDesc := ""
 	if c.sipConfig.addr != "" {

trunk/3rdparty/srs-bench/gb28181/gb_test.go

@@ -1,6 +1,6 @@
 // The MIT License (MIT)
 //
-// Copyright (c) 2022 Winlin
+// # Copyright (c) 2022 Winlin
 //
 // Permission is hereby granted, free of charge, to any person obtaining a copy of
 // this software and associated documentation files (the "Software"), to deal in
@@ -23,11 +23,18 @@ package gb28181
 import (
 	"context"
 	"fmt"
+	"io"
+	"math/rand"
+	"strconv"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/ossrs/srs-bench/srs"
+
 	"github.com/ghettovoice/gosip/sip"
 	"github.com/ossrs/go-oryx-lib/errors"
 	"github.com/ossrs/go-oryx-lib/logger"
-	"testing"
-	"time"
 )
 
 func TestGbPublishRegularly(t *testing.T) {
@@ -523,3 +530,72 @@ func TestGbPublishUnregister(t *testing.T) {
 		t.Errorf("err %+v", err)
 	}
 }
+
+func TestGbPublishWithApi(t *testing.T) {
+	ctx := logger.WithContext(context.Background())
+	ctx, cancel := context.WithTimeout(ctx, time.Duration(*srsTimeout)*time.Millisecond)
+	defer cancel()
+
+	psConfig := PSConfig{
+		video: *srsPublishVideo,
+		fps:   *srsPublishVideoFps,
+		audio: *srsPublishAudio,
+	}
+
+	var r0, r1, r2 error
+	r0 = func() error {
+		// generate a random ssrc, and use it as the stream name.
+		ssrc := rand.Intn(900000000) + 1000000000
+		stream := strconv.FormatUint(uint64(ssrc), 10)
+
+		port, err := apiGbPublishRequest(ctx, stream, strconv.Itoa(ssrc))
+		if err != nil {
+			return errors.Wrapf(err, "apiGbPublishRequest stream=%v", stream)
+		}
+
+		ingester := NewPSIngester(&IngesterConfig{
+			psConfig:    psConfig,
+			ssrc:        uint32(ssrc),
+			clockRate:   90000,
+			payloadType: 96,
+		})
+		defer ingester.Close()
+
+		var wg sync.WaitGroup
+		defer wg.Wait()
+
+		// Check the stream is publishing.
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			for ctx.Err() == nil {
+				stat := srs.NewStatApi(ctx).Streams().FilterByStreamSuffix(stream)
+				logger.Tf(ctx, "Check publishing, stream=%v", stat.Stream())
+				if stat.Stream() != nil {
+					return
+				}
+				time.Sleep(1 * time.Second)
+			}
+			r1 = ctx.Err()
+		}()
+
+		// Ingest the stream.
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			ingester.conf.serverAddr = "tcp://localhost:" + strconv.Itoa(port)
+			if err := ingester.Ingest(ctx); err != nil {
+				if errors.Cause(err) == io.EOF {
+					logger.Tf(ctx, "EOF, video=%v, audio=%v", psConfig.video, psConfig.audio)
+					return
+				}
+				r2 = err
+			}
+		}()
+		return err
+	}()
+
+	if err := filterTestError(ctx.Err(), r0, r1, r2); err != nil {
+		t.Errorf("err %+v", err)
+	}
+}

trunk/3rdparty/srs-bench/gb28181/util.go

@@ -23,13 +23,10 @@ package gb28181
 import (
 	"bufio"
 	"context"
-	"encoding/json"
 	"flag"
 	"fmt"
 	"io"
-	"io/ioutil"
 	"net"
-	"net/http"
 	"net/url"
 	"os"
 	"path"
@@ -39,7 +36,7 @@ import (
 	"github.com/ghettovoice/gosip/sip"
 	"github.com/ossrs/go-oryx-lib/aac"
 	"github.com/ossrs/go-oryx-lib/errors"
-	"github.com/ossrs/go-oryx-lib/logger"
+	"github.com/ossrs/srs-bench/srs"
 	"github.com/yapingcat/gomedia/mpeg2"
 )
 
@@ -365,52 +362,23 @@ func (v *AACReader) NextADTSFrame() ([]byte, error) {
 	return adts, nil
 }
 
-// Request SRS GB API, the apiPath like "/gb/v1/publish",
-//
-// Return the response of answer  in string.
-func apiGbRequest(ctx context.Context, apiPath, host, stream, ssrc string) (string, error) {
-	api := host
-	if !strings.HasPrefix(apiPath, "/") {
-		api += "/"
-	}
-	api += apiPath
-
-	if !strings.HasSuffix(apiPath, "/") {
-		api += "/"
-	}
-
-	// Build JSON body.
-	reqBody := struct {
+func apiGbPublishRequest(ctx context.Context, stream, ssrc string) (int, error) {
+	req := struct {
 		ClientIP string `json:"clientip"`
 		Stream   string `json:"stream"`
 		SSRC     string `json:"ssrc"`
 	}{
-		host, stream, ssrc,
-	}
-
-	b, err := json.Marshal(reqBody)
-	if err != nil {
-		return "", errors.Wrapf(err, "Marshal body %v", reqBody)
+		"", stream, ssrc,
 	}
-	logger.If(ctx, "Request url api=%v with %v", api, string(b))
-	logger.Tf(ctx, "Request url api=%v with %v bytes", api, len(b))
 
-	req, err := http.NewRequest("POST", api, strings.NewReader(string(b)))
-	if err != nil {
-		return "", errors.Wrapf(err, "HTTP request %v", string(b))
-	}
+	res := struct {
+		Code int `json:"code"`
+		Port int `json:"port"`
+	}{}
 
-	res, err := http.DefaultClient.Do(req.WithContext(ctx))
-	if err != nil {
-		return "", errors.Wrapf(err, "Do HTTP request %v", string(b))
-	}
-
-	b2, err := ioutil.ReadAll(res.Body)
-	if err != nil {
-		return "", errors.Wrapf(err, "Read response for %v", string(b))
+	if err := srs.ApiRequest(ctx, "http://localhost:1985/gb/v1/publish/", req, &res); err != nil {
+		return 0, errors.Wrapf(err, "gb/v1/publish")
 	}
-	logger.If(ctx, "Response from %v is %v", api, string(b2))
-	logger.Tf(ctx, "Response from %v is %v bytes", api, len(b2))
 
-	return string(b2), nil
+	return res.Port, nil
 }

trunk/3rdparty/srs-bench/srs/api.go

@@ -87,6 +87,10 @@ func apiRequest(ctx context.Context, r string, req interface{}, res interface{})
 	return nil
 }
 
+func ApiRequest(ctx context.Context, r string, req interface{}, res interface{}) error {
+	return apiRequest(ctx, r, req, res)
+}
+
 // The SRS HTTP statistic API.
 type statAPI struct {
 	ctx     context.Context
@@ -98,6 +102,10 @@ func newStatAPI(ctx context.Context) *statAPI {
 	return &statAPI{ctx: ctx}
 }
 
+func NewStatApi(ctx context.Context) *statAPI {
+	return newStatAPI(ctx)
+}
+
 type statGeneral struct {
 	Code   int    `json:"code"`
 	Server string `json:"server"`
@@ -152,3 +160,7 @@ func (v *statAPI) FilterByStreamSuffix(suffix string) *statAPI {
 	}
 	return v
 }
+
+func (v *statAPI) Stream() *statStream {
+	return v.stream
+}