large refactor, fixes #2

server/internal/webrtc/data.go

@@ -1,22 +0,0 @@
-package webrtc
-
-type dataHeader struct {
-	Event  uint8
-	Length uint16
-}
-
-type dataMouseMove struct {
-	dataHeader
-	X int16
-	Y int16
-}
-
-type dataMouseKey struct {
-	dataHeader
-	Key uint8
-}
-
-type dataKeyboardKey struct {
-	dataHeader
-	Key uint16
-}

server/internal/webrtc/handle.go

@@ -0,0 +1,138 @@
+package webrtc
+
+import (
+	"bytes"
+	"encoding/binary"
+	"strconv"
+
+	"github.com/pion/webrtc/v2"
+	"n.eko.moe/neko/internal/hid"
+)
+
+const OP_MOVE = 0x01
+const OP_SCROLL = 0x02
+const OP_KEY_DOWN = 0x03
+const OP_KEY_UP = 0x04
+const OP_KEY_CLK = 0x05
+
+type PayloadHeader struct {
+	Event  uint8
+	Length uint16
+}
+
+type PayloadMove struct {
+	PayloadHeader
+	X uint16
+	Y uint16
+}
+
+type PayloadScroll struct {
+	PayloadHeader
+	X int16
+	Y int16
+}
+
+type PayloadKey struct {
+	PayloadHeader
+	Key uint16
+}
+
+func (m *WebRTCManager) handle(id string, msg webrtc.DataChannelMessage) error {
+	if !m.sessions.IsHost(id) {
+		return nil
+	}
+
+	buffer := bytes.NewBuffer(msg.Data)
+	header := &PayloadHeader{}
+	hbytes := make([]byte, 3)
+
+	if _, err := buffer.Read(hbytes); err != nil {
+		return err
+	}
+
+	if err := binary.Read(bytes.NewBuffer(hbytes), binary.LittleEndian, header); err != nil {
+		return err
+	}
+
+	buffer = bytes.NewBuffer(msg.Data)
+
+	switch header.Event {
+	case OP_MOVE:
+		payload := &PayloadMove{}
+		if err := binary.Read(buffer, binary.LittleEndian, payload); err != nil {
+			return err
+		}
+
+		hid.Move(int(payload.X), int(payload.Y))
+		break
+	case OP_SCROLL:
+		payload := &PayloadScroll{}
+		if err := binary.Read(buffer, binary.LittleEndian, payload); err != nil {
+			return err
+		}
+
+		m.logger.
+			Debug().
+			Str("x", strconv.Itoa(int(payload.X))).
+			Str("y", strconv.Itoa(int(payload.Y))).
+			Msg("scroll")
+
+		hid.Scroll(int(payload.X), int(payload.Y))
+		break
+	case OP_KEY_DOWN:
+		payload := &PayloadKey{}
+		if err := binary.Read(buffer, binary.LittleEndian, payload); err != nil {
+			return err
+		}
+
+		if payload.Key < 8 {
+			button, err := hid.ButtonDown(int(payload.Key))
+			if err != nil {
+				m.logger.Warn().Err(err).Msg("key down failed")
+				return nil
+			}
+
+			m.logger.Debug().Msgf("button down %s(%d)", button.Name, payload.Key)
+		} else {
+			key, err := hid.KeyDown(int(payload.Key))
+			if err != nil {
+				m.logger.Warn().Err(err).Msg("key down failed")
+				return nil
+			}
+
+			m.logger.Debug().Msgf("key down %s(%d)", key.Name, payload.Key)
+		}
+
+		break
+	case OP_KEY_UP:
+		payload := &PayloadKey{}
+		err := binary.Read(buffer, binary.LittleEndian, payload)
+		if err != nil {
+			return err
+		}
+
+		if payload.Key < 8 {
+			button, err := hid.ButtonUp(int(payload.Key))
+			if err != nil {
+				m.logger.Warn().Err(err).Msg("button up failed")
+				return nil
+			}
+
+			m.logger.Debug().Msgf("button up %s(%d)", button.Name, payload.Key)
+		} else {
+			key, err := hid.KeyUp(int(payload.Key))
+			if err != nil {
+				m.logger.Warn().Err(err).Msg("keyup failed")
+				return nil
+			}
+
+			m.logger.Debug().Msgf("key up %s(%d)", key.Name, payload.Key)
+		}
+		break
+	case OP_KEY_CLK:
+		// unused
+		break
+	}
+
+	return nil
+}

server/internal/webrtc/logger.go

@@ -0,0 +1,32 @@
+package webrtc
+
+import (
+	"github.com/pion/logging"
+	"github.com/rs/zerolog"
+)
+
+type logger struct {
+	logger zerolog.Logger
+}
+
+func (l logger) Trace(msg string)                          { l.logger.Trace().Msg(msg) }
+func (l logger) Tracef(format string, args ...interface{}) { l.logger.Trace().Msgf(format, args...) }
+func (l logger) Debug(msg string)                          { l.logger.Debug().Msg(msg) }
+func (l logger) Debugf(format string, args ...interface{}) { l.logger.Debug().Msgf(format, args...) }
+func (l logger) Info(msg string)                           { l.logger.Info().Msg(msg) }
+func (l logger) Infof(format string, args ...interface{})  { l.logger.Info().Msgf(format, args...) }
+func (l logger) Warn(msg string)                           { l.logger.Warn().Msg(msg) }
+func (l logger) Warnf(format string, args ...interface{})  { l.logger.Warn().Msgf(format, args...) }
+func (l logger) Error(msg string)                          { l.logger.Error().Msg(msg) }
+func (l logger) Errorf(format string, args ...interface{}) { l.logger.Error().Msgf(format, args...) }
+
+type loggerFactory struct {
+	logger zerolog.Logger
+}
+
+func (l loggerFactory) NewLogger(subsystem string) logging.LeveledLogger {
+	l.logger.Debug().Msgf("creating logger for %s", subsystem)
+	return logger{
+		logger: l.logger.With().Str("subsystem", subsystem).Logger(),
+	}
+}

server/internal/webrtc/manager.go

@@ -1,61 +1,42 @@
 package webrtc
 
 import (
-	"math/rand"
-	"net/http"
+	"fmt"
 	"time"
 
-	"github.com/gorilla/websocket"
 	"github.com/pion/webrtc/v2"
+	"github.com/pkg/errors"
 	"github.com/rs/zerolog"
 	"github.com/rs/zerolog/log"
 
+	"n.eko.moe/neko/internal/config"
+	"n.eko.moe/neko/internal/event"
 	"n.eko.moe/neko/internal/gst"
+	"n.eko.moe/neko/internal/hid"
+	"n.eko.moe/neko/internal/message"
+	"n.eko.moe/neko/internal/session"
 )
 
-func NewManager(password string) (*WebRTCManager, error) {
+func New(sessions *session.SessionManager, conf *config.WebRTC) *WebRTCManager {
+	logger := log.With().Str("module", "webrtc").Logger()
 	engine := webrtc.MediaEngine{}
+	engine.RegisterDefaultCodecs()
 
-	videoCodec := webrtc.NewRTPVP8Codec(webrtc.DefaultPayloadTypeVP8, 90000)
-	video, err := webrtc.NewTrack(webrtc.DefaultPayloadTypeVP8, rand.Uint32(), "stream", "stream", videoCodec)
-	if err != nil {
-		return nil, err
+	setings := webrtc.SettingEngine{
+		LoggerFactory: loggerFactory{
+			logger: logger,
+		},
 	}
-	engine.RegisterCodec(videoCodec)
-
-	audioCodec := webrtc.NewRTPOpusCodec(webrtc.DefaultPayloadTypeOpus, 48000)
-	audio, err := webrtc.NewTrack(webrtc.DefaultPayloadTypeOpus, rand.Uint32(), "stream", "stream", audioCodec)
-	if err != nil {
-		return nil, err
-	}
-	engine.RegisterCodec(audioCodec)
-
-	videoPipeline := gst.CreatePipeline(webrtc.VP8, []*webrtc.Track{video}, "ximagesrc show-pointer=true use-damage=false ! video/x-raw,framerate=30/1 ! videoconvert")
-	// ximagesrc xid=0 show-pointer=true ! videoconvert ! queue | videotestsrc
-
-	audioPipeline := gst.CreatePipeline(webrtc.Opus, []*webrtc.Track{audio}, "pulsesrc device=auto_null.monitor ! audioconvert")
-	// pulsesrc device=auto_null.monitor ! audioconvert | audiotestsrc
-	// gst-launch-1.0 -v pulsesrc device=auto_null.monitor ! audioconvert ! vorbisenc ! oggmux ! filesink location=alsasrc.ogg
 
 	return &WebRTCManager{
-		logger:        log.With().Str("service", "webrtc").Logger(),
-		engine:        engine,
-		api:           webrtc.NewAPI(webrtc.WithMediaEngine(engine)),
-		video:         video,
-		videoPipeline: videoPipeline,
-		audio:         audio,
-		audioPipeline: audioPipeline,
-		controller:    "",
-		password:      password,
-		sessions:      make(map[string]*session),
-		debounce:      make(map[int]time.Time),
-		cleanup:       time.NewTicker(500 * time.Second),
-		shutdown:      make(chan bool),
-		upgrader: websocket.Upgrader{
-			CheckOrigin: func(r *http.Request) bool {
-				return true
-			},
-		},
+		logger:   logger,
+		engine:   engine,
+		setings:  setings,
+		api:      webrtc.NewAPI(webrtc.WithMediaEngine(engine), webrtc.WithSettingEngine(setings)),
+		cleanup:  time.NewTicker(1 * time.Second),
+		shutdown: make(chan bool),
+		sessions: sessions,
+		conf:     conf,
 		config: webrtc.Configuration{
 			ICEServers: []webrtc.ICEServer{
 				{
@@ -64,49 +45,182 @@ func NewManager(password string) (*WebRTCManager, error) {
 			},
 			SDPSemantics: webrtc.SDPSemanticsUnifiedPlanWithFallback,
 		},
-	}, nil
+	}
 }
 
 type WebRTCManager struct {
 	logger        zerolog.Logger
-	upgrader      websocket.Upgrader
 	engine        webrtc.MediaEngine
-	api           *webrtc.API
+	setings       webrtc.SettingEngine
 	config        webrtc.Configuration
-	password      string
-	controller    string
-	sessions      map[string]*session
-	debounce      map[int]time.Time
-	shutdown      chan bool
-	cleanup       *time.Ticker
+	sessions      *session.SessionManager
+	api           *webrtc.API
 	video         *webrtc.Track
 	audio         *webrtc.Track
 	videoPipeline *gst.Pipeline
 	audioPipeline *gst.Pipeline
+	cleanup       *time.Ticker
+	conf          *config.WebRTC
+	shutdown      chan bool
 }
 
-func (manager *WebRTCManager) Start() error {
-	manager.videoPipeline.Start()
-	manager.audioPipeline.Start()
+func (m *WebRTCManager) Start() {
+
+	hid.Display(m.conf.Display)
+
+	switch m.conf.VideoCodec {
+	case "vp8":
+		if err := m.createVideoTrack(webrtc.DefaultPayloadTypeVP8); err != nil {
+			m.logger.Panic().Err(err).Msg("unable to start webrtc manager")
+		}
+	case "vp9":
+		if err := m.createVideoTrack(webrtc.DefaultPayloadTypeVP9); err != nil {
+			m.logger.Panic().Err(err).Msg("unable to start webrtc manager")
+		}
+	case "h264":
+		if err := m.createVideoTrack(webrtc.DefaultPayloadTypeH264); err != nil {
+			m.logger.Panic().Err(err).Msg("unable to start webrtc manager")
+		}
+	default:
+		m.logger.Panic().Err(errors.Errorf("unknown video codec %s", m.conf.AudioCodec)).Msg("unable to start webrtc manager")
+	}
+
+	switch m.conf.AudioCodec {
+	case "opus":
+		if err := m.createAudioTrack(webrtc.DefaultPayloadTypeOpus); err != nil {
+			m.logger.Panic().Err(err).Msg("unable to start webrtc manager")
+		}
+	case "g722":
+		if err := m.createAudioTrack(webrtc.DefaultPayloadTypeG722); err != nil {
+			m.logger.Panic().Err(err).Msg("unable to start webrtc manager")
+		}
+	case "pcmu":
+		if err := m.createAudioTrack(webrtc.DefaultPayloadTypePCMU); err != nil {
+			m.logger.Panic().Err(err).Msg("unable to start webrtc manager")
+		}
+	case "pcma":
+		if err := m.createAudioTrack(webrtc.DefaultPayloadTypePCMA); err != nil {
+			m.logger.Panic().Err(err).Msg("unable to start webrtc manager")
+		}
+	default:
+		m.logger.Panic().Err(errors.Errorf("unknown audio codec %s", m.conf.AudioCodec)).Msg("unable to start webrtc manager")
+	}
+
+	m.videoPipeline.Start()
+	m.audioPipeline.Start()
 
 	go func() {
+		defer func() {
+			m.logger.Info().Msg("shutdown")
+		}()
+
 		for {
 			select {
-			case <-manager.shutdown:
+			case <-m.shutdown:
 				return
-			case <-manager.cleanup.C:
-				manager.checkKeys()
+			case <-m.cleanup.C:
+				hid.Check(time.Second * 10)
 			}
 		}
 	}()
 
+	m.sessions.OnHostCleared(func(id string) {
+		hid.Reset()
+	})
+
+	m.sessions.OnCreated(func(id string, session *session.Session) {
+		m.logger.Debug().Str("id", id).Msg("session created")
+	})
+
+	m.sessions.OnDestroy(func(id string) {
+		m.logger.Debug().Str("id", id).Msg("session destroyed")
+	})
+
+	// TODO: log resolution, bit rate and codec parameters
+	m.logger.Info().
+		Str("video_display", m.conf.Display).
+		Str("video_codec", m.conf.VideoCodec).
+		Str("audio_device", m.conf.Device).
+		Str("audio_codec", m.conf.AudioCodec).
+		Msgf("webrtc streaming")
+}
+
+func (m *WebRTCManager) Shutdown() error {
+	m.logger.Info().Msgf("webrtc shutting down")
+
+	m.cleanup.Stop()
+	m.shutdown <- true
+	m.videoPipeline.Stop()
+	m.audioPipeline.Stop()
 	return nil
 }
 
-func (manager *WebRTCManager) Shutdown() error {
-	manager.cleanup.Stop()
-	manager.shutdown <- true
-	manager.videoPipeline.Stop()
-	manager.audioPipeline.Stop()
+func (m *WebRTCManager) CreatePeer(id string, sdp string) error {
+	session, ok := m.sessions.Get(id)
+	if !ok {
+		return fmt.Errorf("invalid session id %s", id)
+	}
+
+	peer, err := m.api.NewPeerConnection(m.config)
+	if err != nil {
+		return err
+	}
+
+	if _, err := peer.AddTransceiverFromTrack(m.video, webrtc.RtpTransceiverInit{
+		Direction: webrtc.RTPTransceiverDirectionSendonly,
+	}); err != nil {
+		return err
+	}
+
+	if _, err := peer.AddTransceiverFromTrack(m.audio, webrtc.RtpTransceiverInit{
+		Direction: webrtc.RTPTransceiverDirectionSendonly,
+	}); err != nil {
+		return err
+	}
+
+	peer.SetRemoteDescription(webrtc.SessionDescription{
+		SDP:  sdp,
+		Type: webrtc.SDPTypeOffer,
+	})
+
+	answer, err := peer.CreateAnswer(nil)
+	if err != nil {
+		return err
+	}
+
+	if err = peer.SetLocalDescription(answer); err != nil {
+		return err
+	}
+
+	if err := session.Send(message.SDP{
+		Message: message.Message{Event: event.SDP_REPLY},
+		SDP:     answer.SDP,
+	}); err != nil {
+		return err
+	}
+
+	peer.OnDataChannel(func(d *webrtc.DataChannel) {
+		d.OnMessage(func(msg webrtc.DataChannelMessage) {
+			if err = m.handle(id, msg); err != nil {
+				m.logger.Warn().Err(err).Msg("data handle failed")
+			}
+		})
+	})
+
+	peer.OnConnectionStateChange(func(connectionState webrtc.PeerConnectionState) {
+		switch connectionState {
+		case webrtc.PeerConnectionStateDisconnected:
+		case webrtc.PeerConnectionStateFailed:
+			m.logger.Info().Str("id", id).Msg("peer disconnected")
+			m.sessions.Destroy(id)
+			break
+		case webrtc.PeerConnectionStateConnected:
+			m.logger.Info().Str("id", id).Msg("peer connected")
+			break
+		}
+	})
+
+	m.sessions.SetPeer(id, peer)
+
 	return nil
 }

server/internal/webrtc/messages.go

@@ -1,15 +0,0 @@
-package webrtc
-
-type message struct {
-	Event string `json:"event"`
-}
-
-type messageIdentityProvide struct {
-	message
-	ID string `json:"id"`
-}
-
-type messageSDP struct {
-	message
-	SDP string `json:"sdp"`
-}

server/internal/webrtc/peer.go

@@ -1,266 +0,0 @@
-package webrtc
-
-import (
-	"bytes"
-	"encoding/binary"
-	"encoding/json"
-	"time"
-
-	"github.com/go-vgo/robotgo"
-	"github.com/pion/webrtc/v2"
-
-	"n.eko.moe/neko/internal/keys"
-)
-
-func (manager *WebRTCManager) createPeer(session *session, raw []byte) error {
-	payload := messageSDP{}
-	if err := json.Unmarshal(raw, &payload); err != nil {
-		return err
-	}
-
-	peer, err := manager.api.NewPeerConnection(manager.config)
-	if err != nil {
-		return err
-	}
-
-	_, err = peer.AddTrack(manager.video)
-	if err != nil {
-		return err
-	}
-
-	_, err = peer.AddTrack(manager.audio)
-	if err != nil {
-		return err
-	}
-
-	peer.SetRemoteDescription(webrtc.SessionDescription{
-		SDP:  payload.SDP,
-		Type: webrtc.SDPTypeOffer,
-	})
-
-	answer, err := peer.CreateAnswer(nil)
-	if err != nil {
-		return err
-	}
-
-	if err = peer.SetLocalDescription(answer); err != nil {
-		return err
-	}
-
-	session.send(messageSDP{
-		message{Event: "sdp/reply"},
-		answer.SDP,
-	})
-
-	session.peer = peer
-
-	peer.OnDataChannel(func(d *webrtc.DataChannel) {
-		d.OnMessage(func(msg webrtc.DataChannelMessage) {
-			if err = manager.onData(session, msg); err != nil {
-				manager.logger.Warn().Err(err).Msg("onData failed")
-			}
-		})
-	})
-
-	peer.OnConnectionStateChange(func(connectionState webrtc.PeerConnectionState) {
-		switch connectionState {
-		case webrtc.PeerConnectionStateDisconnected:
-		case webrtc.PeerConnectionStateFailed:
-			manager.destroy(session)
-			break
-		case webrtc.PeerConnectionStateConnected:
-			manager.logger.Info().Str("ID", session.id).Msg("Peer connected")
-			break
-		}
-	})
-
-	return nil
-}
-
-func (manager *WebRTCManager) onData(session *session, msg webrtc.DataChannelMessage) error {
-	if manager.controller != session.id {
-		return nil
-	}
-
-	header := &dataHeader{}
-	buffer := bytes.NewBuffer(msg.Data)
-	byt := make([]byte, 3)
-	_, err := buffer.Read(byt)
-	if err != nil {
-		return err
-	}
-
-	err = binary.Read(bytes.NewBuffer(byt), binary.LittleEndian, header)
-	if err != nil {
-		return err
-	}
-
-	buffer = bytes.NewBuffer(msg.Data)
-
-	switch header.Event {
-	case 0x01: // MOUSE_MOVE
-		payload := &dataMouseMove{}
-		if err := binary.Read(buffer, binary.LittleEndian, payload); err != nil {
-			return err
-		}
-
-		robotgo.Move(int(payload.X), int(payload.Y))
-		break
-	case 0x02: // MOUSE_UP
-		payload := &dataMouseKey{}
-		if err := binary.Read(buffer, binary.LittleEndian, payload); err != nil {
-			return err
-		}
-
-		code := int(payload.Key)
-		if key, ok := keys.Mouse[code]; ok {
-			if _, ok := manager.debounce[code]; !ok {
-				return nil
-			}
-			delete(manager.debounce, code)
-			robotgo.MouseToggle("up", key)
-			manager.logger.Debug().Msgf("MOUSE_UP key: %v (%v)", code, key)
-		} else {
-			manager.logger.Warn().Msgf("Unknown MOUSE_UP key: %v", code)
-		}
-		break
-	case 0x03: // MOUSE_DOWN
-		payload := &dataMouseKey{}
-		err := binary.Read(buffer, binary.LittleEndian, payload)
-		if err != nil {
-			return err
-		}
-
-		code := int(payload.Key)
-		if key, ok := keys.Mouse[code]; ok {
-			if _, ok := manager.debounce[code]; ok {
-				return nil
-			}
-
-			manager.debounce[code] = time.Now()
-			robotgo.MouseToggle("down", key)
-			manager.logger.Debug().Msgf("MOUSE_DOWN key: %v (%v)", code, key)
-		} else {
-			manager.logger.Warn().Msgf("Unknown MOUSE_DOWN key: %v", code)
-		}
-		break
-	case 0x04: // MOUSE_CLK
-		payload := &dataMouseKey{}
-		err := binary.Read(buffer, binary.LittleEndian, payload)
-		if err != nil {
-			return err
-		}
-
-		code := int(payload.Key)
-		if key, ok := keys.Mouse[code]; ok {
-			switch code {
-			case keys.MOUSE_WHEEL_DOWN:
-				robotgo.Scroll(0, -1)
-				break
-			case keys.MOUSE_WHEEL_UP:
-				robotgo.Scroll(0, 1)
-				break
-			case keys.MOUSE_WHEEL_LEFT:
-				robotgo.Scroll(-1, 0)
-				break
-			case keys.MOUSE_WHEEL_RIGH:
-				robotgo.Scroll(1, 0)
-				break
-			default:
-				robotgo.Click(key, false)
-			}
-
-			manager.logger.Debug().Msgf("MOUSE_CLK key: %v (%v)", code, key)
-		} else {
-			manager.logger.Warn().Msgf("Unknown MOUSE_CLK key: %v", code)
-		}
-		break
-	case 0x05: // KEY_DOWN
-		payload := &dataKeyboardKey{}
-		err := binary.Read(buffer, binary.LittleEndian, payload)
-		if err != nil {
-			return err
-		}
-
-		code := int(payload.Key)
-		if key, ok := keys.Keyboard[code]; ok {
-			if _, ok := manager.debounce[code]; ok {
-				return nil
-			}
-			manager.debounce[code] = time.Now()
-			robotgo.KeyToggle(key, "down")
-			manager.logger.Debug().Msgf("KEY_DOWN key: %v (%v)", code, key)
-		} else {
-			manager.logger.Warn().Msgf("Unknown KEY_DOWN key: %v", code)
-		}
-		break
-	case 0x06: // KEY_UP
-		payload := &dataKeyboardKey{}
-		if err := binary.Read(buffer, binary.LittleEndian, payload); err != nil {
-			return err
-		}
-
-		code := int(payload.Key)
-		if key, ok := keys.Keyboard[code]; ok {
-			if _, ok := manager.debounce[code]; !ok {
-				return nil
-			}
-			delete(manager.debounce, code)
-			robotgo.KeyToggle(key, "up")
-			manager.logger.Debug().Msgf("KEY_UP key: %v (%v)", code, key)
-		} else {
-			manager.logger.Warn().Msgf("Unknown KEY_UP key: %v", code)
-		}
-		break
-	case 0x07: // KEY_CLK
-		payload := &dataKeyboardKey{}
-		err := binary.Read(buffer, binary.LittleEndian, payload)
-		if err != nil {
-			return err
-		}
-
-		code := int(payload.Key)
-		if key, ok := keys.Keyboard[code]; ok {
-			robotgo.KeyTap(key)
-			manager.logger.Debug().Msgf("KEY_CLK key: %v (%v)", code, key)
-		} else {
-			manager.logger.Warn().Msgf("Unknown KEY_CLK key: %v", code)
-		}
-		break
-	}
-
-	return nil
-}
-
-func (manager *WebRTCManager) clearKeys() {
-	for code := range manager.debounce {
-		if key, ok := keys.Keyboard[code]; ok {
-			robotgo.MouseToggle(key, "up")
-		}
-
-		if key, ok := keys.Mouse[code]; ok {
-			robotgo.KeyToggle(key, "up")
-		}
-
-		delete(manager.debounce, code)
-	}
-}
-
-func (manager *WebRTCManager) checkKeys() {
-	t := time.Now()
-	for code, start := range manager.debounce {
-		if t.Sub(start) < (time.Second * 10) {
-			continue
-		}
-
-		if key, ok := keys.Keyboard[code]; ok {
-			robotgo.MouseToggle(key, "up")
-		}
-
-		if key, ok := keys.Mouse[code]; ok {
-			robotgo.KeyToggle(key, "up")
-		}
-
-		delete(manager.debounce, code)
-	}
-}

server/internal/webrtc/session.go

@@ -1,41 +0,0 @@
-package webrtc
-
-import (
-	"sync"
-
-	"github.com/gorilla/websocket"
-	"github.com/pion/webrtc/v2"
-)
-
-type session struct {
-	id     string
-	socket *websocket.Conn
-	peer   *webrtc.PeerConnection
-	mu     sync.Mutex
-}
-
-func (session *session) send(v interface{}) error {
-	session.mu.Lock()
-	defer session.mu.Unlock()
-
-	if session.socket != nil {
-		return session.socket.WriteJSON(v)
-	}
-
-	return nil
-}
-
-func (session *session) destroy() error {
-	if session.peer != nil && session.peer.ConnectionState() == webrtc.PeerConnectionStateConnected {
-		if err := session.peer.Close(); err != nil {
-			return err
-		}
-	}
-
-	if session.socket != nil {
-		if err := session.socket.Close(); err != nil {
-			return err
-		}
-	}
-	return nil
-}

server/internal/webrtc/tracks.go

@@ -0,0 +1,99 @@
+package webrtc
+
+import (
+	"fmt"
+	"math/rand"
+
+	"github.com/pion/webrtc/v2"
+	"github.com/pkg/errors"
+
+	"n.eko.moe/neko/internal/gst"
+)
+
+func (m *WebRTCManager) createVideoTrack(payloadType uint8) error {
+
+	clockrate := uint32(90000)
+	var codec *webrtc.RTPCodec
+	switch payloadType {
+	case webrtc.DefaultPayloadTypeVP8:
+		codec = webrtc.NewRTPVP8Codec(payloadType, clockrate)
+		break
+	case webrtc.DefaultPayloadTypeVP9:
+		codec = webrtc.NewRTPVP9Codec(payloadType, clockrate)
+		break
+	case webrtc.DefaultPayloadTypeH264:
+		codec = webrtc.NewRTPH264Codec(payloadType, clockrate)
+		break
+	default:
+		return errors.Errorf("unknown video codec %s", payloadType)
+	}
+
+	track, err := webrtc.NewTrack(payloadType, rand.Uint32(), "stream", "stream", codec)
+	if err != nil {
+		return err
+	}
+
+	var pipeline *gst.Pipeline
+	src := fmt.Sprintf("ximagesrc xid=%s show-pointer=true use-damage=false ! video/x-raw,framerate=30/1 ! videoconvert ! queue", m.conf.Display)
+	switch payloadType {
+	case webrtc.DefaultPayloadTypeVP8:
+		pipeline = gst.CreatePipeline(webrtc.VP8, []*webrtc.Track{track}, src)
+		break
+	case webrtc.DefaultPayloadTypeVP9:
+		pipeline = gst.CreatePipeline(webrtc.VP9, []*webrtc.Track{track}, src)
+		break
+	case webrtc.DefaultPayloadTypeH264:
+		pipeline = gst.CreatePipeline(webrtc.H264, []*webrtc.Track{track}, src)
+		break
+	}
+
+	m.video = track
+	m.videoPipeline = pipeline
+	return nil
+}
+
+func (m *WebRTCManager) createAudioTrack(payloadType uint8) error {
+	var codec *webrtc.RTPCodec
+	switch payloadType {
+	case webrtc.DefaultPayloadTypeOpus:
+		codec = webrtc.NewRTPOpusCodec(payloadType, 48000)
+		break
+	case webrtc.DefaultPayloadTypeG722:
+		codec = webrtc.NewRTPG722Codec(payloadType, 48000)
+		break
+	case webrtc.DefaultPayloadTypePCMU:
+		codec = webrtc.NewRTPPCMUCodec(payloadType, 8000)
+		break
+	case webrtc.DefaultPayloadTypePCMA:
+		codec = webrtc.NewRTPPCMACodec(payloadType, 8000)
+		break
+	default:
+		return errors.Errorf("unknown audio codec %s", payloadType)
+	}
+
+	track, err := webrtc.NewTrack(payloadType, rand.Uint32(), "stream", "stream", codec)
+	if err != nil {
+		return err
+	}
+
+	var pipeline *gst.Pipeline
+	src := fmt.Sprintf("pulsesrc device=%s ! audioconvert", m.conf.Device)
+	switch payloadType {
+	case webrtc.DefaultPayloadTypeOpus:
+		pipeline = gst.CreatePipeline(webrtc.Opus, []*webrtc.Track{track}, src)
+		break
+	case webrtc.DefaultPayloadTypeG722:
+		pipeline = gst.CreatePipeline(webrtc.G722, []*webrtc.Track{track}, src)
+		break
+	case webrtc.DefaultPayloadTypePCMU:
+		pipeline = gst.CreatePipeline(webrtc.PCMU, []*webrtc.Track{track}, src)
+		break
+	case webrtc.DefaultPayloadTypePCMA:
+		pipeline = gst.CreatePipeline(webrtc.PCMA, []*webrtc.Track{track}, src)
+		break
+	}
+
+	m.audio = track
+	m.audioPipeline = pipeline
+	return nil
+}

server/internal/webrtc/websocket.go

@@ -1,228 +0,0 @@
-package webrtc
-
-import (
-	"encoding/json"
-	"net/http"
-	"sync"
-	"time"
-
-	"github.com/gorilla/websocket"
-	"github.com/pkg/errors"
-
-	"n.eko.moe/neko/internal/nanoid"
-)
-
-const (
-	// Send pings to peer with this period. Must be less than pongWait.
-	pingPeriod = 60 * time.Second
-)
-
-func (manager *WebRTCManager) Upgrade(w http.ResponseWriter, r *http.Request) error {
-	manager.logger.
-		Info().
-		Msg("Attempting to upgrade ws")
-
-	socket, err := manager.upgrader.Upgrade(w, r, nil)
-	if err != nil {
-		manager.logger.Error().Err(err).Msg("Failed to upgrade websocket!")
-		return err
-	}
-
-	sessionID, ok := manager.authenticate(r)
-	if ok != true {
-		manager.logger.Warn().Msg("Authenticatetion failed")
-		if err = socket.Close(); err != nil {
-			return err
-		}
-		return nil
-	}
-
-	session := &session{
-		id:     sessionID,
-		socket: socket,
-		mu:     sync.Mutex{},
-	}
-
-	manager.logger.
-		Info().
-		Str("ID", sessionID).
-		Str("RemoteAddr", socket.RemoteAddr().String()).
-		Msg("Created Session")
-
-	manager.sessions[sessionID] = session
-
-	defer func() {
-		manager.destroy(session)
-	}()
-
-	if err = manager.onConnected(session); err != nil {
-		manager.logger.Error().Err(err).Msg("onConnected failed!")
-		return nil
-	}
-
-	manager.handleWS(session)
-
-	return nil
-}
-
-func (manager *WebRTCManager) authenticate(r *http.Request) (sessionID string, ok bool) {
-
-	passwords, ok := r.URL.Query()["password"]
-	if !ok || len(passwords[0]) < 1 {
-		return "", false
-	}
-
-	if passwords[0] != manager.password {
-		manager.logger.Warn().Str("Password", passwords[0]).Msg("Wrong password: ")
-		return "", false
-	}
-
-	id, err := nanoid.NewIDSize(32)
-	if err != nil {
-		return "", false
-	}
-	return id, true
-}
-
-func (manager *WebRTCManager) onConnected(session *session) error {
-	if err := session.send(messageIdentityProvide{
-		message: message{Event: "identity/provide"},
-		ID:      session.id,
-	}); err != nil {
-		return err
-	}
-	return nil
-}
-
-func (manager *WebRTCManager) onMessage(session *session, raw []byte) error {
-	message := message{}
-	if err := json.Unmarshal(raw, &message); err != nil {
-		return err
-	}
-
-	switch message.Event {
-	case "sdp/provide":
-		return errors.Wrap(manager.createPeer(session, raw), "sdp/provide failed")
-	case "control/release":
-		return errors.Wrap(manager.controlRelease(session), "control/release failed")
-	case "control/request":
-		return errors.Wrap(manager.controlRequest(session), "control/request failed")
-	default:
-		manager.logger.Warn().Msgf("Unknown client method %s", message.Event)
-	}
-
-	return nil
-}
-
-func (manager *WebRTCManager) handleWS(session *session) {
-	bytes := make(chan []byte)
-	cancel := make(chan struct{})
-	ticker := time.NewTicker(pingPeriod)
-
-	go func() {
-		defer func() {
-			ticker.Stop()
-			manager.logger.Info().Str("RemoteAddr", session.socket.RemoteAddr().String()).Msg("Handle WS ending")
-			manager.destroy(session)
-		}()
-
-		for {
-			_, raw, err := session.socket.ReadMessage()
-			if err != nil {
-				if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseAbnormalClosure) {
-					manager.logger.Warn().Err(err).Msg("ReadMessage error")
-				}
-				break
-			}
-			bytes <- raw
-		}
-	}()
-
-	for {
-		select {
-		case raw := <-bytes:
-			manager.logger.Info().
-				Str("ID", session.id).
-				Str("Message", string(raw)).
-				Msg("Reading from Websocket")
-			if err := manager.onMessage(session, raw); err != nil {
-				manager.logger.Error().Err(err).Msg("onClientMessage has failed")
-				return
-			}
-		case <-cancel:
-			return
-		case _ = <-ticker.C:
-			if err := session.socket.WriteMessage(websocket.PingMessage, nil); err != nil {
-				return
-			}
-		}
-	}
-}
-
-func (manager *WebRTCManager) destroy(session *session) {
-	if manager.controller == session.id {
-		manager.controller = ""
-		manager.clearKeys()
-		for id, sess := range manager.sessions {
-			if id != session.id {
-				if err := sess.send(message{Event: "control/released"}); err != nil {
-					manager.logger.Error().Err(err).Msg("session.send has failed")
-				}
-			}
-		}
-	}
-
-	if err := session.destroy(); err != nil {
-		manager.logger.Error().Err(err).Msg("session.destroy has failed")
-	}
-
-	delete(manager.sessions, session.id)
-}
-
-func (manager *WebRTCManager) controlRelease(session *session) error {
-	if manager.controller == session.id {
-		manager.controller = ""
-		manager.clearKeys()
-
-		if err := session.send(message{Event: "control/release"}); err != nil {
-			return err
-		}
-
-		for id, sess := range manager.sessions {
-			if id != session.id {
-				if err := sess.send(message{Event: "control/released"}); err != nil {
-					return err
-				}
-			}
-		}
-	}
-	return nil
-}
-
-func (manager *WebRTCManager) controlRequest(session *session) error {
-	if manager.controller == "" {
-		manager.controller = session.id
-
-		if err := session.send(message{Event: "control/give"}); err != nil {
-			return err
-		}
-
-		for id, sess := range manager.sessions {
-			if id != session.id {
-				if err := sess.send(message{Event: "control/given"}); err != nil {
-					return err
-				}
-			}
-		}
-	} else {
-		if err := session.send(message{Event: "control/locked"}); err != nil {
-			return err
-		}
-
-		controller, ok := manager.sessions[manager.controller]
-		if ok {
-			controller.send(message{Event: "control/requesting"})
-		}
-	}
-	return nil
-}