add a benchmark tool

cmd/radiobench/radiobench.go

+package main
+
+import (
+	"flag"
+	"fmt"
+	"io/ioutil"
+	"log"
+	"net/http"
+	"os"
+	"strings"
+	"sync"
+	"time"
+)
+
+var (
+	numConns int
+
+	stats     = &Stats{}
+	retryTime = 2 * time.Second
+)
+
+type Stats struct {
+	HttpStatus map[int]int
+	HttpErrors int
+	Errors     int
+	lock       sync.Mutex
+}
+
+func (s *Stats) HttpError(resp *http.Response) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	cur, ok := s.HttpStatus[resp.StatusCode]
+	if !ok {
+		cur = 0
+	}
+	s.HttpStatus[resp.StatusCode] = cur + 1
+	s.HttpErrors++
+}
+
+func (s *Stats) Error() {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	s.Errors++
+}
+
+func (s *Stats) Dump() {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+	log.Printf("errs=%d http_errs=%d http_status=%v", s.Errors, s.HttpErrors, s.HttpStatus)
+}
+
+func init() {
+	flag.IntVar(&numConns, "n", 3, "number of parallel connections")
+	flag.IntVar(&numConns, "num-clients", 3, "number of parallel connections")
+}
+
+func readstream(id int, streamUrl string) error {
+	resp, err := http.Get(streamUrl)
+	if err != nil {
+		stats.Error()
+		return err
+	}
+	if resp.StatusCode != 200 {
+		stats.HttpError(resp)
+		resp.Body.Close()
+		return fmt.Errorf("http status %s", resp.Status)
+	}
+
+	if resp.Header.Get("Content-Type") == "audio/x-mpegurl" {
+		data, err := ioutil.ReadAll(resp.Body)
+		resp.Body.Close()
+		if err != nil {
+			stats.Error()
+			return err
+		}
+
+		streamUrl = strings.TrimSpace(string(data))
+		resp, err = http.Get(streamUrl)
+		if err != nil {
+			stats.Error()
+			return err
+		}
+		if resp.StatusCode != 200 {
+			stats.HttpError(resp)
+			resp.Body.Close()
+			return fmt.Errorf("http status %s", resp.Status)
+		}
+	}
+
+	log.Printf("worker(%d): connected to %s", id, streamUrl)
+
+	defer resp.Body.Close()
+
+	// Just read data and discard it.
+	buf := make([]byte, 16384)
+	for {
+		n, err := resp.Body.Read(buf)
+		if err != nil {
+			stats.Error()
+			break
+		}
+		if n == 0 {
+			break
+		}
+	}
+	return fmt.Errorf("connection lost")
+}
+
+func worker(id int, streamUrl string) {
+	for {
+		err := readstream(id, streamUrl)
+		log.Printf("worker(%d): %v", id, err)
+		time.Sleep(retryTime)
+	}
+}
+
+func dumpStats() {
+	t := time.NewTicker(10 * time.Second)
+	for {
+		<-t.C
+		stats.Dump()
+	}
+}
+
+func main() {
+	flag.Parse()
+	if flag.NArg() != 1 {
+		fmt.Printf("Usage: radiobench [<OPTIONS>] <STREAM_URL>\n")
+		os.Exit(1)
+	}
+
+	streamUrl := flag.Arg(0)
+
+	go dumpStats()
+
+	var wg sync.WaitGroup
+	for i := 0; i < numConns; i++ {
+		wg.Add(1)
+		go func(id int) {
+			worker(id, streamUrl)
+			wg.Done()
+		}(i)
+	}
+	wg.Wait()
+}