iridium/neko
2
2
Fork 0
mirror of https://github.com/m1k1o/neko.git synced 2024-07-24 14:40:50 +12:00
Code Issues Packages Projects Releases Wiki Activity

server -> client signaling

Browse Source
This commit is contained in:
Craig
2020-02-12 23:13:33 +00:00
parent 1f8b4d44c7
commit d5bff32302
17 changed files with 181 additions and 286 deletions
Download Patch File Download Diff File Expand all files Collapse all files

26
server/internal/session/manager.go
View File

@ -131,32 +131,6 @@ func (manager *SessionManager) Brodcast(v interface{}, exclude interface{}) erro
return nil
}
func (manager *SessionManager) WriteVideoSample(sample types.Sample) error {
for _, session := range manager.members {
if !session.connected {
continue
}
if err := session.WriteVideoSample(sample); err != nil {
return err
}
}
return nil
}
func (manager *SessionManager) WriteAudioSample(sample types.Sample) error {
for _, session := range manager.members {
if !session.connected {
continue
}
if err := session.WriteAudioSample(sample); err != nil {
return err
}
}
return nil
}
func (manager *SessionManager) OnHost(listener func(id string)) {
manager.emmiter.On("host", func(payload ...interface{}) {
listener(payload[0].(string))

23
server/internal/session/session.go
View File

@ -64,8 +64,6 @@ func (session *Session) SetMuted(muted bool) {
func (session *Session) SetName(name string) error {
session.name = name
session.connected = true
session.manager.emmiter.Emit("connected", session.id, session)
return nil
}
@ -79,6 +77,14 @@ func (session *Session) SetPeer(peer types.Peer) error {
return nil
}
func (session *Session) SetConnected(connected bool) error {
session.connected = connected
if connected {
session.manager.emmiter.Emit("connected", session.id, session)
}
return nil
}
func (session *Session) Kick(reason string) error {
if session.socket == nil {
return nil
@ -107,18 +113,11 @@ func (session *Session) Write(v interface{}) error {
return session.socket.Send(v)
}
func (session *Session) WriteVideoSample(sample types.Sample) error {
if session.peer == nil || !session.connected {
func (session *Session) SignalAnwser(sdp string) error {
if session.peer == nil {
return nil
}
return session.peer.WriteVideoSample(sample)
}
func (session *Session) WriteAudioSample(sample types.Sample) error {
if session.peer == nil || !session.connected {
return nil
}
return session.peer.WriteAudioSample(sample)
return session.peer.SignalAnwser(sdp)
}
func (session *Session) destroy() error {

5
server/internal/types/event/events.go
View File

@ -9,11 +9,6 @@ const (
SIGNAL_PROVIDE = "signal/provide"
)
const (
IDENTITY_PROVIDE = "identity/provide"
IDENTITY_DETAILS = "identity/details"
)
const (
MEMBER_LIST = "member/list"
MEMBER_CONNECTED = "member/connected"

11
server/internal/types/message/messages.go
View File

@ -13,19 +13,16 @@ type Disconnect struct {
Message string `json:"message"`
}
type Identity struct {
type SignalProvide struct {
Event string `json:"event"`
ID string `json:"id"`
SDP string `json:"sdp"`
}
type IdentityDetails struct {
type SignalAnswer struct {
Event string `json:"event"`
Username string `json:"username"`
}
type Signal struct {
Event string `json:"event"`
SDP string `json:"sdp"`
SDP string `json:"sdp"`
}
type MembersList struct {

6
server/internal/types/session.go
View File

@ -16,14 +16,14 @@ type Session interface {
Member() *Member
SetMuted(muted bool)
SetName(name string) error
SetConnected(connected bool) error
SetSocket(socket WebScoket) error
SetPeer(peer Peer) error
Address() *string
Kick(message string) error
Write(v interface{}) error
Send(v interface{}) error
WriteAudioSample(sample Sample) error
WriteVideoSample(sample Sample) error
SignalAnwser(sdp string) error
}
type SessionManager interface {
@ -39,8 +39,6 @@ type SessionManager interface {
Destroy(id string) error
Clear() error
Brodcast(v interface{}, exclude interface{}) error
WriteAudioSample(sample Sample) error
WriteVideoSample(sample Sample) error
OnHost(listener func(id string))
OnHostCleared(listener func(id string))
OnDestroy(listener func(id string))

5
server/internal/types/webrtc.go
View File

@ -8,13 +8,12 @@ type Sample struct {
type WebRTCManager interface {
Start()
Shutdown() error
CreatePeer(id string, sdp string) (string, Peer, error)
CreatePeer(id string, session Session) (string, error)
ChangeScreenSize(width int, height int, rate int) error
}
type Peer interface {
WriteVideoSample(sample Sample) error
WriteAudioSample(sample Sample) error
SignalAnwser(sdp string) error
WriteData(v interface{}) error
Destroy() error
}

21
server/internal/webrtc/peer.go
View File

@ -4,32 +4,17 @@ import (
"sync"
"github.com/pion/webrtc/v2"
"github.com/pion/webrtc/v2/pkg/media"
"n.eko.moe/neko/internal/types"
)
type Peer struct {
id string
engine webrtc.MediaEngine
api *webrtc.API
video *webrtc.Track
audio *webrtc.Track
manager *WebRTCManager
connection *webrtc.PeerConnection
mu sync.Mutex
}
func (peer *Peer) WriteAudioSample(sample types.Sample) error {
if err := peer.audio.WriteSample(media.Sample(sample)); err != nil {
return err
}
return nil
}
func (peer *Peer) WriteVideoSample(sample types.Sample) error {
if err := peer.video.WriteSample(media.Sample(sample)); err != nil {
return err
}
return nil
func (peer *Peer) SignalAnwser(sdp string) error {
return peer.connection.SetRemoteDescription(webrtc.SessionDescription{SDP: sdp, Type: webrtc.SDPTypeAnswer})
}
func (peer *Peer) WriteData(v interface{}) error {

8
server/internal/webrtc/tracks.go
View File

@ -7,9 +7,9 @@ import (
"github.com/pion/webrtc/v2"
)
func (m *WebRTCManager) createVideoTrack(engine webrtc.MediaEngine) (*webrtc.Track, error) {
func (m *WebRTCManager) createVideoTrack() (*webrtc.Track, error) {
var codec *webrtc.RTPCodec
for _, videoCodec := range engine.GetCodecsByKind(webrtc.RTPCodecTypeVideo) {
for _, videoCodec := range m.engine.GetCodecsByKind(webrtc.RTPCodecTypeVideo) {
if videoCodec.Name == m.videoPipeline.CodecName {
codec = videoCodec
break
@ -23,9 +23,9 @@ func (m *WebRTCManager) createVideoTrack(engine webrtc.MediaEngine) (*webrtc.Tra
return webrtc.NewTrack(codec.PayloadType, rand.Uint32(), "stream", "stream", codec)
}
func (m *WebRTCManager) createAudioTrack(engine webrtc.MediaEngine) (*webrtc.Track, error) {
func (m *WebRTCManager) createAudioTrack() (*webrtc.Track, error) {
var codec *webrtc.RTPCodec
for _, videoCodec := range engine.GetCodecsByKind(webrtc.RTPCodecTypeAudio) {
for _, videoCodec := range m.engine.GetCodecsByKind(webrtc.RTPCodecTypeAudio) {
if videoCodec.Name == m.audioPipeline.CodecName {
codec = videoCodec
break

124
server/internal/webrtc/webrtc.go
View File

@ -2,10 +2,12 @@ package webrtc
import (
"fmt"
"io"
"strings"
"time"
"github.com/pion/webrtc/v2"
"github.com/pion/webrtc/v2/pkg/media"
"github.com/rs/zerolog"
"github.com/rs/zerolog/log"
@ -28,13 +30,22 @@ func New(sessions types.SessionManager, config *config.WebRTC) *WebRTCManager {
settings.SetEphemeralUDPPortRange(config.EphemeralMin, config.EphemeralMax)
settings.SetNAT1To1IPs(config.NAT1To1IPs, webrtc.ICECandidateTypeHost)
// Create MediaEngine based off sdp
engine := webrtc.MediaEngine{}
engine.RegisterDefaultCodecs()
// Create API with MediaEngine and SettingEngine
api := webrtc.NewAPI(webrtc.WithMediaEngine(engine), webrtc.WithSettingEngine(settings))
return &WebRTCManager{
logger: logger,
settings: settings,
cleanup: time.NewTicker(1 * time.Second),
shutdown: make(chan bool),
sessions: sessions,
engine: engine,
config: config,
api: api,
configuration: &webrtc.Configuration{
SDPSemantics: webrtc.SDPSemanticsUnifiedPlanWithFallback,
},
@ -44,18 +55,23 @@ func New(sessions types.SessionManager, config *config.WebRTC) *WebRTCManager {
type WebRTCManager struct {
logger zerolog.Logger
settings webrtc.SettingEngine
sessions types.SessionManager
engine webrtc.MediaEngine
api *webrtc.API
videoTrack *webrtc.Track
audioTrack *webrtc.Track
videoPipeline *gst.Pipeline
audioPipeline *gst.Pipeline
sessions types.SessionManager
cleanup *time.Ticker
config *config.WebRTC
shutdown chan bool
configuration *webrtc.Configuration
}
func (m *WebRTCManager) Start() {
// Set display and change to default resolution
xorg.Display(m.config.Display)
if !xorg.ValidScreenSize(m.config.ScreenWidth, m.config.ScreenHeight, m.config.ScreenRate) {
m.logger.Warn().Msgf("invalid screen option %dx%d@%d", m.config.ScreenWidth, m.config.ScreenHeight, m.config.ScreenRate)
} else {
@ -64,31 +80,39 @@ func (m *WebRTCManager) Start() {
}
}
// Create video track/pipeline
videoPipeline, err := gst.CreatePipeline(
m.config.VideoCodec,
m.config.Display,
m.config.VideoParams,
)
if err != nil {
m.logger.Panic().Err(err).Msg("unable to start webrtc manager")
}
m.videoPipeline = videoPipeline
video, err := m.createVideoTrack()
if err != nil {
m.logger.Panic().Err(err).Msg("unable to start webrtc manager")
}
m.videoTrack = video
// Create audio track/pipeline
audioPipeline, err := gst.CreatePipeline(
m.config.AudioCodec,
m.config.Device,
m.config.AudioParams,
)
if err != nil {
m.logger.Panic().Err(err).Msg("unable to start webrtc manager")
}
m.videoPipeline = videoPipeline
m.audioPipeline = audioPipeline
videoPipeline.Start()
audioPipeline.Start()
audio, err := m.createAudioTrack()
if err != nil {
m.logger.Panic().Err(err).Msg("unable to start webrtc manager")
}
m.audioTrack = audio
go func() {
defer func() {
@ -100,11 +124,11 @@ func (m *WebRTCManager) Start() {
case <-m.shutdown:
return
case sample := <-m.videoPipeline.Sample:
if err := m.sessions.WriteVideoSample(sample); err != nil {
if err := m.videoTrack.WriteSample(media.Sample(sample)); err != nil && err != io.ErrClosedPipe {
m.logger.Warn().Err(err).Msg("video pipeline failed to write")
}
case sample := <-m.audioPipeline.Sample:
if err := m.sessions.WriteAudioSample(sample); err != nil {
if err := m.audioTrack.WriteSample(media.Sample(sample)); err != nil && err != io.ErrClosedPipe {
m.logger.Warn().Err(err).Msg("audio pipeline failed to write")
}
case <-m.cleanup.C:
@ -125,6 +149,10 @@ func (m *WebRTCManager) Start() {
m.logger.Debug().Str("id", id).Msg("session destroyed")
})
// start pipelines
videoPipeline.Start()
audioPipeline.Start()
// TODO: log resolution, bit rate and codec parameters
m.logger.Info().
Str("video_display", m.config.Display).
@ -147,64 +175,28 @@ func (m *WebRTCManager) Shutdown() error {
return nil
}
func (m *WebRTCManager) CreatePeer(id string, sdp string) (string, types.Peer, error) {
// Create SessionDescription
description := webrtc.SessionDescription{
SDP: sdp,
Type: webrtc.SDPTypeOffer,
}
// Create MediaEngine based off sdp
engine := webrtc.MediaEngine{}
engine.PopulateFromSDP(description)
// Create API with MediaEngine and SettingEngine
api := webrtc.NewAPI(webrtc.WithMediaEngine(engine), webrtc.WithSettingEngine(m.settings))
func (m *WebRTCManager) CreatePeer(id string, session types.Session) (string, error) {
// Create new peer connection
connection, err := api.NewPeerConnection(*m.configuration)
connection, err := m.api.NewPeerConnection(*m.configuration)
if err != nil {
return "", nil, err
return "", err
}
// Create video track
video, err := m.createVideoTrack(engine)
if err != nil {
return "", nil, err
}
_, err = connection.AddTransceiverFromTrack(video, webrtc.RtpTransceiverInit{
if _, err = connection.AddTransceiverFromTrack(m.videoTrack, webrtc.RtpTransceiverInit{
Direction: webrtc.RTPTransceiverDirectionSendonly,
})
if err != nil {
return "", nil, err
}); err != nil {
return "", err
}
// Create audio track
audio, err := m.createAudioTrack(engine)
if err != nil {
return "", nil, err
}
_, err = connection.AddTransceiverFromTrack(audio, webrtc.RtpTransceiverInit{
if _, err = connection.AddTransceiverFromTrack(m.audioTrack, webrtc.RtpTransceiverInit{
Direction: webrtc.RTPTransceiverDirectionSendonly,
})
if err != nil {
return "", nil, err
}); err != nil {
return "", err
}
// Set remote description
connection.SetRemoteDescription(description)
answer, err := connection.CreateAnswer(nil)
description, err := connection.CreateOffer(nil)
if err != nil {
return "", nil, err
}
if err = connection.SetLocalDescription(answer); err != nil {
return "", nil, err
return "", err
}
connection.OnDataChannel(func(d *webrtc.DataChannel) {
@ -215,6 +207,7 @@ func (m *WebRTCManager) CreatePeer(id string, sdp string) (string, types.Peer, e
})
})
connection.SetLocalDescription(description)
connection.OnConnectionStateChange(func(state webrtc.PeerConnectionState) {
switch state {
case webrtc.PeerConnectionStateDisconnected:
@ -224,18 +217,23 @@ func (m *WebRTCManager) CreatePeer(id string, sdp string) (string, types.Peer, e
break
case webrtc.PeerConnectionStateConnected:
m.logger.Info().Str("id", id).Msg("peer connected")
if err = session.SetConnected(true); err != nil {
m.logger.Warn().Err(err).Msg("unable to set connected on peer")
m.sessions.Destroy(id)
}
break
}
})
return answer.SDP, &Peer{
if err := session.SetPeer(&Peer{
id: id,
api: api,
engine: engine,
video: video,
audio: audio,
manager: m,
connection: connection,
}, nil
}); err != nil {
return "", err
}
return description.SDP, nil
}
func (m *WebRTCManager) ChangeScreenSize(width int, height int, rate int) error {

13
server/internal/websocket/handler.go
View File

@ -57,18 +57,11 @@ func (h *MessageHandler) Message(id string, raw []byte) error {
switch header.Event {
// Signal Events
case event.SIGNAL_PROVIDE:
payload := &message.Signal{}
case event.SIGNAL_ANSWER:
payload := &message.SignalAnswer{}
return errors.Wrapf(
utils.Unmarshal(payload, raw, func() error {
return h.createPeer(id, session, payload)
}), "%s failed", header.Event)
// Identity Events
case event.IDENTITY_DETAILS:
payload := &message.IdentityDetails{}
return errors.Wrapf(
utils.Unmarshal(payload, raw, func() error {
return h.identityDetails(id, session, payload)
return h.signalAnswer(id, session, payload)
}), "%s failed", header.Event)
// Control Events

34
server/internal/websocket/identity.go
View File

@ -1,34 +0,0 @@
package websocket
import (
"n.eko.moe/neko/internal/types"
"n.eko.moe/neko/internal/types/event"
"n.eko.moe/neko/internal/types/message"
)
func (h *MessageHandler) identityDetails(id string, session types.Session, payload *message.IdentityDetails) error {
if err := session.SetName(payload.Username); err != nil {
return err
}
return nil
}
func (h *MessageHandler) createPeer(id string, session types.Session, payload *message.Signal) error {
sdp, peer, err := h.webrtc.CreatePeer(id, payload.SDP)
if err != nil {
return err
}
if err := session.SetPeer(peer); err != nil {
return err
}
if err := session.Send(message.Signal{
Event: event.SIGNAL_ANSWER,
SDP: sdp,
}); err != nil {
return err
}
return nil
}

8
server/internal/websocket/session.go
View File

@ -7,10 +7,8 @@ import (
)
func (h *MessageHandler) SessionCreated(id string, session types.Session) error {
if err := session.Send(message.Identity{
Event: event.IDENTITY_PROVIDE,
ID: id,
}); err != nil {
// send sdp and id over to client
if err := h.signalProvide(id, session); err != nil {
return err
}
@ -20,7 +18,7 @@ func (h *MessageHandler) SessionCreated(id string, session types.Session) error
}
if session.Admin() {
// send screen configurations
// send screen configurations if admin
if err := h.screenConfigurations(id, session); err != nil {
return err
}

36
server/internal/websocket/signal.go Normal file
View File

@ -0,0 +1,36 @@
package websocket
import (
"n.eko.moe/neko/internal/types"
"n.eko.moe/neko/internal/types/event"
"n.eko.moe/neko/internal/types/message"
)
func (h *MessageHandler) signalProvide(id string, session types.Session) error {
sdp, err := h.webrtc.CreatePeer(id, session)
if err != nil {
return err
}
if err := session.Send(message.SignalProvide{
Event: event.SIGNAL_PROVIDE,
ID: id,
SDP: sdp,
}); err != nil {
return err
}
return nil
}
func (h *MessageHandler) signalAnswer(id string, session types.Session, payload *message.SignalAnswer) error {
if err := session.SetName(payload.Username); err != nil {
return err
}
if err := session.SignalAnwser(payload.SDP); err != nil {
return err
}
return nil
}