byteDropper.go

package main

import (
	"bufio"
	"flag"
	"fmt"
	"io"
	"math/rand"
	"os"
	"time"
)

const TS_PACKET_LENGTH = 188
const IP_PACKET_LENGTH = 188 * 7

var pctDrop = flag.Float64("d", 0, "Fraction of packets to drop [0.0, 1.0)")
var delaySeconds = flag.Int("delay", 0, "Seconds to delay starting dropping")

//Example usage dropping 1% packets
//`cat <somefile.ts> | tsp -P regulate | ./byteDropper -d 0.01 -delay 180 | srt-live-transmit file://con <srt_url>`
func main() {
	flag.Parse()
	rand.Seed(time.Now().UnixNano())

	doDrop := true
	if *delaySeconds > 0 {
		doDrop = false
	}
	go func() {
		<-time.After(time.Duration(*delaySeconds) * time.Second)
		fmt.Fprintf(os.Stderr, "Beginning to drop packets.\n")
		doDrop = true
	}()

	in := os.Stdin

	packetIndex := int64(0)
	dropCount := int64(0)
	buf := make([]byte, IP_PACKET_LENGTH)
	buffer := bufio.NewReaderSize(in, IP_PACKET_LENGTH*10)
	oBuffer := bufio.NewWriterSize(os.Stdout, IP_PACKET_LENGTH*10)

	var err error
	var n int
	for {
		//This always reads a full packet
		_, err = io.ReadFull(buffer, buf)
		if err != nil {
			if err == io.EOF || err == io.ErrUnexpectedEOF {
				break
			}
			fmt.Fprintf(os.Stderr, "Error reading from Stdin: %v\n", err.Error())
			os.Exit(1)
		}
		packetIndex++

		if buf[0] != 0x47 {
			fmt.Fprintf(os.Stderr, "Error: No sync byte present in packet %d\n", packetIndex)
			os.Exit(1)
		}

		if doDrop && *pctDrop > 0.0 && rand.Float64() <= *pctDrop {
			// fmt.Fprintf(os.Stderr, "Dropping packet\n")
			dropCount++
			continue
		}

		n, err = oBuffer.Write(buf)
		if err != nil {
			if err == io.EOF || err == io.ErrUnexpectedEOF {
				break
			}
			fmt.Fprintf(os.Stderr, "Error writing to Stdout: %v\n", err.Error())
			os.Exit(1)
		}
		if n < len(buf) {
			fmt.Fprintf(os.Stderr, "Error, short write to Stdout: %d < %d\n", n, len(buf))
			os.Exit(1)
		}
	}
	fmt.Fprintf(os.Stderr, "Completed.\nProcessed: %d\tDropped: %d\tPercentage: %.4f\tTarget: %f\n", packetIndex, dropCount, float64(dropCount)/float64(packetIndex), *pctDrop)
}