yebisu-client/main.go

package main

import (
	"context"
	"flag"
	"log"
	"net/http"

	"golang.org/x/crypto/blake2b"

	dev "github.com/vladimirvivien/go4vl/device"
	"github.com/vladimirvivien/go4vl/v4l2"

	"github.com/gorilla/websocket"
)

func sender(queue chan []byte, done chan int, secret string) {
	log.Println("start sender")

	defer func() {
		log.Println("sender done")

		done <- 0
	}()

	h := make(http.Header, 1)
	h.Set("X-Secret", secret)

	ws, _, err := websocket.DefaultDialer.Dial("wss://yebi.su/ws", h)
	if err != nil {
		log.Println("dial error")

		return
	}

	defer ws.Close()

	log.Println("connected")

	for {
		data := <-queue

		log.Printf("sending %db\n", len(data))

		if err := ws.WriteMessage(websocket.BinaryMessage, data); err != nil {
			log.Println("WriteMessage() failed")

			break
		}
	}
}

func chunkBy[T any](items []T, chunkSize int) (chunks [][]T) {
	for chunkSize < len(items) {
		items, chunks = items[chunkSize:], append(chunks, items[0:chunkSize:chunkSize])
	}
	return chunks
}

func processor(inQueue chan []byte, outQueue chan []byte) {
	log.Println("start processor")

	for {
		data := <-inQueue

		newData := []byte{}
		for _, chunk := range chunkBy(data, 2) {
			newData = append(newData, chunk[0]^chunk[1])
		}

		data = newData

		newData = []byte{}
		for _, chunk := range chunkBy(data, 8) {
			var b byte = 0

			for j := 0; j < 8; j++ {
				if chunk[j]&1 != 0 {
					b |= (1 << (7 - j))
				}
			}

			newData = append(newData, b)
		}

		data = newData

		newData = []byte{}
		for _, chunk := range chunkBy(data, 32) {
			sum := blake2b.Sum256(chunk)

			newData = append(newData, sum[:]...)
		}

		outQueue <- newData
	}
}

func main() {
	var deviceName string
	var width uint
	var height uint
	var bufferSize 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.StringVar(&ident, "ident", "", "yebi.su ident")
	flag.StringVar(&secret, "secret", "", "yebi.su secret")

	flag.Parse()

	if ident == "" {
		log.Panicln("ident not specified")
	}

	if secret == "" {
		log.Panicln("secret not specified")
	}

	secret = ident + " " + secret

	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}),
	)
	if err != nil {
		log.Panicln("failed to open device")
	}

	defer device.Close()

	senderDone := make(chan int, 1)
	processorQueue := make(chan []byte, bufferSize)
	senderQueue := make(chan []byte, bufferSize)

	go sender(senderQueue, senderDone, secret)
	go processor(processorQueue, senderQueue)

	log.Println("start recording")

	ctx, stop := context.WithCancel(context.TODO())

	if err := device.Start(ctx); err != nil {
		log.Panicln("failed to start recording on device")
	}

	defer stop()

	log.Println("start loop")

	for {
		select {
		case <-senderDone:
			go sender(senderQueue, senderDone, secret)
		case frame := <-device.GetOutput():
			processorQueue <- frame
		}
	}
}