debug needed

lib/audiodriver/wavfifo.go

@@ -0,0 +1,131 @@
+package audiodriver
+
+import (
+	"context"
+	"encoding/binary"
+	"errors"
+	"io"
+	"os"
+	"time"
+
+	"github.com/pion/mediadevices/pkg/io/audio"
+	"github.com/pion/mediadevices/pkg/prop"
+	"github.com/pion/mediadevices/pkg/wave"
+	"github.com/sirupsen/logrus"
+)
+
+const (
+	DataChunkIDSize   = 4
+	DataChunkSizeSize = 4
+)
+
+// BufferSize for pcm bytes
+const BufferSize = 512
+const BitsPerByte = 8
+
+type WavFIFODriver struct {
+	FIFOFile   string
+	f          *os.File
+	waveHeader *WavHeader
+	closed     <-chan struct{}
+	cancel     func()
+}
+
+func New(fifoname string) *WavFIFODriver {
+	return &WavFIFODriver{
+		FIFOFile: fifoname,
+	}
+}
+
+func (w *WavFIFODriver) Open() error {
+	ctx, cancel := context.WithCancel(context.Background())
+	w.closed = ctx.Done()
+	w.cancel = cancel
+
+	f, err := os.Open(w.FIFOFile)
+	if err != nil {
+		return err
+	}
+
+	w.f = f
+	w.waveHeader = &WavHeader{}
+	return w.waveHeader.Parse(f)
+}
+
+func (w *WavFIFODriver) Close() error {
+	defer w.cancel()
+	return w.f.Close()
+}
+
+func (w *WavFIFODriver) Properties() []prop.Media {
+	return []prop.Media{
+		{
+			Audio: prop.Audio{
+				SampleRate:    int(w.waveHeader.SampleRate),
+				ChannelCount:  int(w.waveHeader.NumChannels),
+				Latency:       time.Millisecond * time.Duration(BufferSize) / time.Duration(w.waveHeader.SampleRate) / time.Duration(w.waveHeader.NumChannels),
+				IsFloat:       false, // just 8bit or 16bit with qemu
+				IsBigEndian:   false, // qemu should be little endian
+				IsInterleaved: true,
+			},
+		},
+	}
+}
+
+func (w *WavFIFODriver) AudioRecord(p prop.Media) (audio.Reader, error) {
+	offset := FmtHeaderOffset + FmtHeaderIDSize + FmtHeaderChunkSizeSize + int64(w.waveHeader.Size) + DataChunkIDSize + DataChunkSizeSize
+	if _, err := w.f.Seek(
+		offset,
+		io.SeekStart,
+	); err != nil {
+		return nil, err
+	}
+
+	var pcm <-chan [BufferSize]byte
+	go func() {
+		samples := make(chan [BufferSize]byte, BufferSize)
+		defer close(samples)
+
+		pcm = samples
+		for {
+			var b [BufferSize]byte
+			if _, err := w.f.Read(b[:]); err != nil {
+				if errors.Is(err, io.EOF) {
+					return
+				}
+				logrus.Errorf("read audio stream error: %v", err)
+				continue
+			}
+			samples <- b
+		}
+	}()
+
+	reader := func() (wave.Audio, func(), error) {
+		select {
+		case <-w.closed:
+			return nil, func() {}, io.EOF
+		case pcmData, ok := <-pcm:
+			if !ok {
+				return nil, func() {}, io.ErrClosedPipe
+			}
+
+			a := wave.NewInt16Interleaved(wave.ChunkInfo{
+				Len:          BufferSize / int(w.waveHeader.BitsPerSample/BitsPerByte),
+				Channels:     p.ChannelCount,
+				SamplingRate: p.SampleRate,
+			})
+			copy(a.Data, bytesTo16BitSamples(pcmData[:]))
+			return a, func() {}, nil
+		}
+	}
+	return audio.ReaderFunc(reader), nil
+}
+
+func bytesTo16BitSamples(b []byte) []int16 {
+	samples := make([]int16, 0)
+	for i := 0; i < len(b); i += 2 {
+		sample := binary.LittleEndian.Uint16(b[i : i+1])
+		samples = append(samples, int16(sample))
+	}
+	return samples
+}

lib/audiodriver/wavheader.go

@@ -0,0 +1,85 @@
+package audiodriver
+
+import (
+	"bytes"
+	"encoding/binary"
+	"errors"
+	"io"
+)
+
+// Skip riff header and `fmt ` just 16 bytes
+const (
+	FmtHeaderOffset        int64 = 0x0c
+	FmtHeaderIDSize        int64 = 4
+	FmtHeaderChunkSizeSize int64 = 4
+)
+
+type WavHeader struct {
+	ID            [4]byte
+	Size          uint32
+	AudioFormat   uint16
+	NumChannels   uint16
+	SampleRate    uint32
+	ByteRate      uint32
+	BlockAlign    uint16
+	BitsPerSample uint16
+}
+
+func (w *WavHeader) Parse(f io.ReadSeeker) error {
+	if _, err := f.Seek(FmtHeaderOffset, io.SeekStart); err != nil {
+		return err
+	}
+
+	var id [4]byte
+	if _, err := f.Read(id[:]); err != nil {
+		return err
+	}
+	if bytes.Equal(id[:], []byte{'f', 'm', 't', '0'}[:]) {
+		return errors.New("bad header")
+	}
+	w.ID = id
+
+	var size [4]byte
+	if _, err := f.Read(size[:]); err != nil {
+		return err
+	}
+	w.Size = binary.LittleEndian.Uint32(size[:])
+
+	var af [2]byte
+	if _, err := f.Read(af[:]); err != nil {
+		return err
+	}
+	w.AudioFormat = binary.LittleEndian.Uint16(af[:])
+
+	var nc [2]byte
+	if _, err := f.Read(nc[:]); err != nil {
+		return err
+	}
+	w.NumChannels = binary.LittleEndian.Uint16(nc[:])
+
+	var sr [4]byte
+	if _, err := f.Read(sr[:]); err != nil {
+		return err
+	}
+	w.SampleRate = binary.LittleEndian.Uint32(sr[:])
+
+	var br [4]byte
+	if _, err := f.Read(br[:]); err != nil {
+		return err
+	}
+	w.ByteRate = binary.LittleEndian.Uint32(br[:])
+
+	var ba [2]byte
+	if _, err := f.Read(ba[:]); err != nil {
+		return err
+	}
+	w.BlockAlign = binary.LittleEndian.Uint16(ba[:])
+
+	var bps [2]byte
+	if _, err := f.Read(ba[:]); err != nil {
+		return err
+	}
+	w.BitsPerSample = binary.LittleEndian.Uint16(bps[:])
+
+	return nil
+}

lib/qemuconnection/events.go

@@ -0,0 +1,90 @@
+package qemuconnection
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"github.com/digitalocean/go-qemu/qmp"
+)
+
+const (
+	MouseMoveEvent = iota
+	MouseButtonEvent
+	KeyboardEvent
+	ControlEvent
+	QueryStatusEvent
+)
+
+const (
+	ControlEventRestart = iota
+)
+
+type Event struct {
+	Type int            `json:"type"`
+	Args map[string]any `json:"args"`
+}
+
+type CommandLine struct {
+	Command string `json:"command-line"`
+}
+
+func ParseEvent(b []byte) (*Event, error) {
+	var event *Event
+	if err := json.Unmarshal(b, event); err != nil {
+		return nil, err
+	}
+	return event, nil
+}
+
+func (e *Event) ToQemuCommand() []qmp.Command {
+	switch e.Type {
+	case MouseMoveEvent:
+		return []qmp.Command{
+			makeHMCommand("mouse_move %d %d %d", e.Args["dx"], e.Args["dy"], e.Args["dz"]),
+		}
+	case MouseButtonEvent:
+		return []qmp.Command{
+			makeHMCommand("mouse_button %d", e.Args["button"]),
+		}
+	case KeyboardEvent:
+		return []qmp.Command{
+			makeHMCommand("sendkey %s", e.Args["key"]),
+		}
+	case ControlEvent:
+		t, ok := e.Args["cmd"].(int)
+		if ok {
+			return makeControlCommand(t)
+		}
+	case QueryStatusEvent:
+		return []qmp.Command{
+			{
+				Execute: "query-status",
+			},
+		}
+	}
+	return make([]qmp.Command, 0)
+}
+
+func makeControlCommand(t int) []qmp.Command {
+	switch t {
+	case ControlEventRestart:
+		return []qmp.Command{
+			makeHMCommand("system_reset"),
+			makeHMCommand("cont"),
+		}
+	}
+	return make([]qmp.Command, 0)
+}
+
+func makeHMCommand(cmdTemplate string, args ...any) qmp.Command {
+	template := qmp.Command{
+		Execute: "human-monitor-command",
+	}
+	command := CommandLine{
+		Command: fmt.Sprintf(cmdTemplate, args...),
+	}
+
+	template.Args = command
+
+	return template
+}

lib/qemuconnection/loginit.go

@@ -0,0 +1,7 @@
+package qemuconnection
+
+import "github.com/sirupsen/logrus"
+
+func init() {
+	logrus.Info("qemu client control events module loaded")
+}

lib/webrtcconnection/connection.go

@@ -0,0 +1,124 @@
+package webrtcconnection
+
+import (
+	"github.com/pion/mediadevices"
+	"github.com/pion/mediadevices/pkg/codec/opus"
+	"github.com/pion/mediadevices/pkg/codec/x264"
+	"github.com/pion/mediadevices/pkg/prop"
+	"github.com/pion/webrtc/v3"
+	"github.com/sirupsen/logrus"
+)
+
+const DefaultStreamID = "ace-server"
+
+type Connection struct {
+	option *Options
+}
+
+func New(o *Options) *Connection {
+	return &Connection{
+		option: o,
+	}
+}
+
+func (c *Connection) Regist(offer *webrtc.SessionDescription) (*webrtc.SessionDescription, error) {
+	logrus.Debug("received offer ", offer.Type.String())
+	codecSelector, err := setupCodec(c.option.Video.BPS, c.option.Audio.BPS)
+	if err != nil {
+		return nil, err
+	}
+
+	me := &webrtc.MediaEngine{}
+	codecSelector.Populate(me)
+	api := webrtc.NewAPI(webrtc.WithMediaEngine(me))
+
+	rtc, err := api.NewPeerConnection(webrtc.Configuration{
+		ICEServers: []webrtc.ICEServer{
+			{
+				URLs: c.option.STUNServers,
+			},
+		},
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	rtc.OnICEConnectionStateChange(func(connectionState webrtc.ICEConnectionState) {
+		logrus.Debug("connection state has changed: ", connectionState.String())
+
+		switch connectionState {
+		case webrtc.ICEConnectionStateFailed:
+			fallthrough
+		case webrtc.ICEConnectionStateClosed:
+			rtc.Close()
+		}
+	})
+
+	s, err := mediadevices.GetUserMedia(mediadevices.MediaStreamConstraints{
+		Video: func(mtc *mediadevices.MediaTrackConstraints) {
+			mtc.Height = prop.IntExact(c.option.Video.Height)
+			mtc.Width = prop.IntExact(c.option.Video.Width)
+		},
+		Audio: func(mtc *mediadevices.MediaTrackConstraints) {},
+		Codec: codecSelector,
+	})
+	if err != nil {
+		return nil, err
+	}
+
+	for _, track := range s.GetTracks() {
+		track.OnEnded(func(err error) {
+			logrus.Errorf("Track (ID: %s) ended with error: %v", track.ID(), err)
+		})
+		_, err := rtc.AddTransceiverFromTrack(track, webrtc.RTPTransceiverInit{
+			Direction: webrtc.RTPTransceiverDirectionSendonly,
+		})
+		if err != nil {
+			logrus.Error(err)
+		}
+	}
+
+	rtc.OnDataChannel(dataChannel)
+
+	if err := rtc.SetRemoteDescription(*offer); err != nil {
+		return nil, err
+	}
+	logrus.Debug("offer set")
+
+	answer, err := rtc.CreateAnswer(nil)
+	if err != nil {
+		return nil, err
+	}
+
+	gatherComplete := webrtc.GatheringCompletePromise(rtc)
+
+	if err := rtc.SetLocalDescription(answer); err != nil {
+		return nil, err
+	}
+	logrus.Debug("answer set")
+
+	<-gatherComplete
+
+	defer logrus.Debug("regist complete")
+	return rtc.LocalDescription(), nil
+}
+
+func setupCodec(videoBPS, audioBPS int) (*mediadevices.CodecSelector, error) {
+	x264Prarm, err := x264.NewParams()
+	if err != nil {
+		return nil, err
+	}
+	x264Prarm.BitRate = videoBPS
+
+	opusParam, err := opus.NewParams()
+	if err != nil {
+		return nil, err
+	}
+	opusParam.BitRate = audioBPS
+
+	codecSelector := mediadevices.NewCodecSelector(
+		mediadevices.WithAudioEncoders(&opusParam),
+		mediadevices.WithVideoEncoders(&x264Prarm),
+	)
+	return codecSelector, nil
+}

lib/webrtcconnection/datachannel.go

@@ -0,0 +1,50 @@
+package webrtcconnection
+
+import (
+	"encoding/json"
+	"time"
+
+	"git.sense-t.eu.org/ACE/ace/servers/qemuserver"
+	"github.com/pion/webrtc/v3"
+	"github.com/sirupsen/logrus"
+)
+
+func dataChannel(d *webrtc.DataChannel) {
+	d.OnOpen(func() {
+		for {
+			status := qemuserver.GetStatus().String()
+			currentTime := time.Now().UnixMilli()
+
+			b, err := json.Marshal(map[string]any{
+				"qemu_status": status,
+				"server_time": currentTime,
+			})
+			if err != nil {
+				logrus.Errorf(
+					"failed to parse to json on '%s-%d' with error: %v",
+					d.Label(), *d.ID(), err,
+				)
+			}
+
+			if err := d.Send(b); err != nil {
+				logrus.Errorf(
+					"failed to send qemu status to '%s-%d' with error: %v",
+					d.Label(), *d.ID(), err,
+				)
+			}
+			time.Sleep(time.Second)
+		}
+	})
+	d.OnMessage(func(msg webrtc.DataChannelMessage) {
+		logrus.Debugf("received %d bytes message from '%s-%d'", len(msg.Data), d.Label(), *d.ID())
+		if !msg.IsString {
+			return
+		}
+		if err := qemuserver.SendEvent(msg.Data); err != nil {
+			logrus.Errorf(
+				"cannot parse message from '%s-%d' to qemu controll event: %v",
+				d.Label(), *d.ID(), err,
+			)
+		}
+	})
+}

lib/webrtcconnection/loginit.go

@@ -0,0 +1,7 @@
+package webrtcconnection
+
+import "github.com/sirupsen/logrus"
+
+func init() {
+	logrus.Info("webrtc connection module initialized")
+}

lib/webrtcconnection/options.go

@@ -0,0 +1,27 @@
+package webrtcconnection
+
+type Options struct {
+	STUNServers []string `yaml:"stun_servers"`
+	Video       struct {
+		Height int `yaml:"height"`
+		Width  int `yaml:"width"`
+		BPS    int `yaml:"bps"`
+	} `yaml:"video"`
+	Audio struct {
+		BPS int `yaml:"bps"`
+	} `yaml:"audio"`
+}
+
+func ExampleOptions() *Options {
+	options := &Options{
+		STUNServers: []string{
+			"stun:stun.l.google.com:19302",
+			"stun:wetofu.me:3478",
+		},
+	}
+	options.Video.BPS = 500_000
+	options.Video.Height = 768
+	options.Video.Width = 1024
+	options.Audio.BPS = 96_000
+	return options
+}