upd

main.go

@@ -5,6 +5,9 @@ import (
 	"flag"
 	"log"
 	"net/http"
+	"os"
+	"os/signal"
+	"time"
 
 	"golang.org/x/crypto/blake2b"
 
@@ -14,7 +17,7 @@ import (
 	"github.com/gorilla/websocket"
 )
 
-func sender(queue chan []byte, secret string) {
+func sender(queue chan []byte, secret string, cooldown uint) {
 	log.Println("start sender")
 
 	h := make(http.Header, 1)
@@ -33,13 +36,13 @@ func sender(queue chan []byte, secret string) {
 		for {
 			data := <-queue
 
-			log.Printf("sending %db\n", len(data))
-
 			if err := ws.WriteMessage(websocket.BinaryMessage, data); err != nil {
 				log.Println("WriteMessage() failed")
 
 				break
 			}
+
+			time.Sleep(time.Duration(cooldown) * time.Millisecond)
 		}
 
 		ws.Close()
@@ -94,16 +97,18 @@ func processor(inQueue chan []byte, outQueue chan []byte) {
 
 func main() {
 	var deviceName string
-	var width uint
-	var height uint
 	var bufferSize uint
+	var processorsCount uint
+	var sendersCount uint
+	var sendCooldown uint
 	var ident string
 	var secret string
 
 	flag.StringVar(&deviceName, "device", "/dev/video0", "v4l2 device to capture from")
-	flag.UintVar(&width, "width", 1280, "frame width")
-	flag.UintVar(&height, "height", 720, "frame height")
 	flag.UintVar(&bufferSize, "bufferSize", 1024, "sending queue max capacity")
+	flag.UintVar(&processorsCount, "processorsCount", 1, "count of processor goroutines")
+	flag.UintVar(&sendersCount, "sendersCount", 1, "count of sender goroutines")
+	flag.UintVar(&sendCooldown, "sendCooldown", 100, "send cooldown (in ms)")
 	flag.StringVar(&ident, "ident", "", "yebi.su ident")
 	flag.StringVar(&secret, "secret", "", "yebi.su secret")
 
@@ -121,21 +126,40 @@ func main() {
 
 	log.Printf("open v4l2 device: %s\n", deviceName)
 
-	device, err := dev.Open(
-		deviceName,
-		dev.WithPixFormat(v4l2.PixFormat{Width: uint32(width), Height: uint32(height), PixelFormat: v4l2.PixelFmtYUYV, Field: v4l2.FieldNone}),
-	)
+	device, err := dev.Open(deviceName)
 	if err != nil {
 		log.Panicln("failed to open device")
 	}
 
 	defer device.Close()
 
+	frameSizes, err := v4l2.GetFormatFrameSizes(device.Fd(), v4l2.PixelFmtYUYV)
+	if err != nil {
+		log.Panicln("failed to get supported frame sizes")
+	}
+
+	if err := device.SetPixFormat(v4l2.PixFormat{
+		Width:       frameSizes[0].Size.MinWidth,
+		Height:      frameSizes[0].Size.MinHeight,
+		PixelFormat: v4l2.PixelFmtYUYV,
+		Field:       v4l2.FieldNone,
+	}); err != nil {
+		log.Panicln("failed to set device pixel format")
+	}
+
 	processorQueue := make(chan []byte, bufferSize)
 	senderQueue := make(chan []byte, bufferSize)
 
-	go sender(senderQueue, secret)
-	go processor(processorQueue, senderQueue)
+	defer close(processorQueue)
+	defer close(senderQueue)
+
+	for i := 0; i < int(sendersCount); i++ {
+		go sender(senderQueue, secret, sendCooldown)
+	}
+
+	for i := 0; i < int(processorsCount); i++ {
+		go processor(processorQueue, senderQueue)
+	}
 
 	log.Println("start recording")
 
@@ -149,8 +173,19 @@ func main() {
 
 	log.Println("start loop")
 
+	interrupt := make(chan os.Signal, 1)
+	signal.Notify(interrupt, os.Interrupt)
+
 	for {
-		frame := <-device.GetOutput()
-		processorQueue <- frame
+		select {
+		case frame := <-device.GetOutput():
+			if len(frame) != 0 {
+				processorQueue <- frame
+			}
+		case <-interrupt:
+			log.Println("interrupted")
+
+			return
+		}
 	}
 }