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
	}
	qemu *qemu.Domain
}

var DefaultServer *Server

func NewServer(o *Options) (*Server, error) {
	if err := o.MakeFIFO(); err != nil {
		return nil, err
	}

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

	u, err := url.Parse(o.QmpAddress)
	if err != nil {
		return nil, 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, o.Timeout)
	if err != nil {
		return nil, err
	}

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

	qemu, err := qemu.NewDomain(qmpConnection, o.Name)
	if err != nil {
		return nil, err
	}
	server.qemu = qemu

	if err := driver.GetManager().Register(
		audiodriver.New(o.AudioPipe),
		driver.Info{
			Label:      "audioFifo",
			DeviceType: driver.Microphone,
			Priority:   driver.PriorityNormal,
		},
	); err != nil {
		return nil, err
	}
	if err := driver.GetManager().Register(
		vncdriver.NewVnc(o.VNCAddress),
		driver.Info{
			Label:      "vnc",
			DeviceType: driver.Camera,
			Priority:   driver.PriorityNormal,
		},
	); err != nil {
		return nil, err
	}

	return server, nil
}

func (s *Server) Run() error {
	logrus.Debug("qemu server running")
	defer logrus.Debug("qemu server exit")
	defer s.qemu.Close()

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

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

func (s *Server) startCapture() {
	if _, err := s.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) error {
	DefaultServer, err := NewServer(o)
	if err != nil {
		return err
	}
	go func() {
		if err := DefaultServer.Run(); err != nil {
			logrus.Fatal("cannot run qemuserver with error: ", err)
		}
	}()
	return nil
}

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
}