ace/servers/qemuserver/server.go

package qemuserver

import (
	"fmt"
	"net/url"

	"git.sense-t.eu.org/ACE/ace/lib/audiodriver"
	"git.sense-t.eu.org/ACE/ace/lib/qemuconnection"
	"github.com/digitalocean/go-qemu/qemu"
	"github.com/digitalocean/go-qemu/qmp"
	"github.com/pion/mediadevices/pkg/driver"
	"github.com/pion/mediadevices/pkg/driver/vncdriver"
	"github.com/sirupsen/logrus"
)

type Server struct {
	options      *Options
	QmpConnector struct {
		RX chan *qemuconnection.Event
		TX chan qemu.Status
	}
}

var DefaultServer *Server

func NewServer(o *Options) *Server {
	server := &Server{
		options: o,
	}
	server.QmpConnector.RX = make(chan *qemuconnection.Event)
	server.QmpConnector.TX = make(chan qemu.Status)
	return server
}

func (s *Server) Run() error {
	logrus.Debug("qemu server running")
	defer logrus.Debug("qemu server exit")
	u, err := url.Parse(s.options.QmpAddress)
	if err != nil {
		return err
	}
	var address string
	if u.Scheme == "unix" {
		address = u.Path
	} else {
		address = u.Host
	}
	logrus.Debugf("trying to connect qmp with %s://%s", u.Scheme, address)
	qmpConnection, err := qmp.NewSocketMonitor(u.Scheme, address, s.options.Timeout)
	if err != nil {
		return err
	}
	defer qmpConnection.Disconnect()

	if err := qmpConnection.Connect(); err != nil {
		return err
	}
	logrus.Debug("qmp connected")

	qemu, err := qemu.NewDomain(qmpConnection, s.options.Name)
	if err != nil {
		return err
	}
	defer qemu.Close()

	go s.startCapture(qemu)
	logrus.Debug("qemu capture start")

	for ev := range s.QmpConnector.RX {
		if ev.Type == qemuconnection.QueryStatusEvent {
			status, err := qemu.Status()
			if err != nil {
				logrus.Error("get qemu status error: ", err)
				continue
			}
			s.QmpConnector.TX <- status
			continue
		}
		for _, cmd := range ev.ToQemuCommand() {
			_, err := qemu.Run(cmd)
			if err != nil {
				logrus.Error("run command error: ", err)
			}

		}
	}
	return nil
}

func (s *Server) startCapture(qemu *qemu.Domain) {
	if err := s.options.MakeFIFO(); err != nil {
		logrus.Fatal("failed to make pipe file: ", err)
	}

	if err := driver.GetManager().Register(
		audiodriver.New(s.options.AudioPipe),
		driver.Info{
			Label:      "audioFifo",
			DeviceType: driver.Microphone,
			Priority:   driver.PriorityNormal,
		},
	); err != nil {
		logrus.Fatal("audio initialize failed: ", err)
	}
	if err := driver.GetManager().Register(
		vncdriver.NewVnc(s.options.VNCAddress),
		driver.Info{
			Label:      "vnc",
			DeviceType: driver.Camera,
			Priority:   driver.PriorityNormal,
		},
	); err != nil {
		logrus.Fatal("video initialize failed: ", err)
	}

	if _, err := qemu.Run(qmp.Command{
		Execute: "human-monitor-command",
		Args: map[string]string{
			"command-line": fmt.Sprintf(
				"wavcapture %s %s",
				s.options.AudioPipe,
				s.options.AudioDevice,
			),
		},
	}); err != nil {
		logrus.Fatal("run audio command failed: ", err)
	}
	logrus.Debug("audio capture set")
}

func Setup(o *Options) {
	DefaultServer = NewServer(o)
	go func() {
		if err := DefaultServer.Run(); err != nil {
			logrus.Fatal("cannot run qemuserver with error: ", err)
		}
	}()
}

func SendEvent(b []byte) error {
	ev, err := qemuconnection.ParseEvent(b)
	if err != nil {
		return err
	}
	DefaultServer.QmpConnector.RX <- ev
	return nil
}

func GetStatus() qemu.Status {
	DefaultServer.QmpConnector.RX <- &qemuconnection.Event{
		Type: qemuconnection.QueryStatusEvent,
	}
	return <-DefaultServer.QmpConnector.TX
}