Browse Source

Move client audio in/out handling to audio package

wip
Trevor Slocum 3 years ago
parent
commit
d2aecadbc4
  1. 3
      go.mod
  2. 7
      go.sum
  3. 2
      pkg/agent/channel.go
  4. 79
      pkg/agent/client.go
  5. 2
      pkg/agent/message.go
  6. 2
      pkg/agent/user.go
  7. 56
      pkg/audio/in.go
  8. 88
      pkg/audio/out.go
  9. 13
      pkg/web/public/assets/js/harmony.js
  10. 171
      pkg/web/web.go

3
go.mod

@ -7,7 +7,8 @@ require (
github.com/gorilla/mux v1.7.3
github.com/gorilla/websocket v1.4.1
github.com/lucas-clemente/quic-go v0.14.1 // indirect
github.com/omeid/go-resources v0.0.0-20190324090249-46f4269d8abd
github.com/omeid/go-resources v0.0.0-20191215004320-084151b0b77f
github.com/pion/dtls/v2 v2.0.0-rc.4 // indirect
github.com/pion/ice v0.7.7 // indirect
github.com/pion/rtp v1.1.4
github.com/pion/webrtc/v2 v2.1.18

7
go.sum

@ -36,8 +36,8 @@ github.com/marten-seemann/qtls v0.2.3 h1:0yWJ43C62LsZt08vuQJDK1uC1czUc3FJeCLPoNA
github.com/marten-seemann/qtls v0.2.3/go.mod h1:xzjG7avBwGGbdZ8dTGxlBnLArsVKLvwmjgmPuiQEcYk=
github.com/marten-seemann/qtls v0.4.1 h1:YlT8QP3WCCvvok7MGEZkMldXbyqgr8oFg5/n8Gtbkks=
github.com/marten-seemann/qtls v0.4.1/go.mod h1:pxVXcHHw1pNIt8Qo0pwSYQEoZ8yYOOPXTCZLQQunvRc=
github.com/omeid/go-resources v0.0.0-20190324090249-46f4269d8abd h1:VxcHM9xpZ4BHxQPYWAavsxPciBZITxmnGNyIO7hsUfk=
github.com/omeid/go-resources v0.0.0-20190324090249-46f4269d8abd/go.mod h1:SIESmZeFlCKsQZcd2NEiX8spNNmCWB1V/RbM/eBKDfo=
github.com/omeid/go-resources v0.0.0-20191215004320-084151b0b77f h1:SGNncsU0Zwb3E69609zN2mK/Wd1nDMmVkGn29GJAr48=
github.com/omeid/go-resources v0.0.0-20191215004320-084151b0b77f/go.mod h1:z/Qlkuf0oJETCM1bXoga55K0aHbyfwGzYrJVELbU7ck=
github.com/onsi/ginkgo v1.6.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
github.com/onsi/ginkgo v1.7.0 h1:WSHQ+IS43OoUrWtD1/bbclrwK8TTH5hzp+umCiuxHgs=
github.com/onsi/ginkgo v1.7.0/go.mod h1:lLunBs/Ym6LB5Z9jYTR76FiuTmxDTDusOGeTQH+WWjE=
@ -47,6 +47,8 @@ github.com/pion/datachannel v1.4.13 h1:ezTn3AtUtXvKemRRjRdUgao/T8bH4ZJwrpOqU8Iz3
github.com/pion/datachannel v1.4.13/go.mod h1:+rBUwEDonA63KXx994DP/ofyyGVAm6AIMvOqQZxjWRU=
github.com/pion/dtls/v2 v2.0.0-rc.3 h1:u9utI+EDJOjOWfrkGQsD8WNssPcTwfYIanFB6oI8K+4=
github.com/pion/dtls/v2 v2.0.0-rc.3/go.mod h1:x0XH+cN5z+l/+/4nYL8r4sB8g6+0d1Zp2Pfkcoz8BKY=
github.com/pion/dtls/v2 v2.0.0-rc.4 h1:STK55fhtsnSqnHH74MrQoIC0t1JeC8CpI/FOnZp9aXc=
github.com/pion/dtls/v2 v2.0.0-rc.4/go.mod h1:k7HAs0qpJSz+Pelkbc5ZDNtenQpUvXgjg/yq4ZC6CdU=
github.com/pion/ice v0.7.6 h1:EARj1MBq5NYaMtXVhYkK03i0RS/meejNHvZS++K5tSY=
github.com/pion/ice v0.7.6/go.mod h1:4xCajahEEvc5w0AM+Ujx/Rr2EczON/fKndi3jLyDdh4=
github.com/pion/ice v0.7.7 h1:POqtOIISKHwaCd2XqgNyQrylgFl9IZJWZmL9cQQsqkk=
@ -107,6 +109,7 @@ golang.org/x/crypto v0.0.0-20190228161510-8dd112bcdc25/go.mod h1:djNgcEr1/C05ACk
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
golang.org/x/crypto v0.0.0-20190829043050-9756ffdc2472/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
golang.org/x/crypto v0.0.0-20191029031824-8986dd9e96cf/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
golang.org/x/crypto v0.0.0-20191205180655-e7c4368fe9dd/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
golang.org/x/crypto v0.0.0-20191206172530-e9b2fee46413 h1:ULYEB3JvPRE/IfO+9uO7vKV/xzVTO7XPAwm8xbf4w2g=
golang.org/x/crypto v0.0.0-20191206172530-e9b2fee46413/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=

2
pkg/web/channel.go → pkg/agent/channel.go

@ -1,4 +1,4 @@
package web
package agent
import (
"sync"

79
pkg/web/client.go → pkg/agent/client.go

@ -1,16 +1,14 @@
package web
package agent
import (
"encoding/json"
"log"
"strconv"
"sync"
"time"
"git.sr.ht/~tslocum/harmony/pkg/audio"
"github.com/gorilla/websocket"
"github.com/pion/rtp"
"github.com/pion/webrtc/v2"
"github.com/pkg/errors"
)
type Client struct {
@ -26,18 +24,8 @@ type Client struct {
In chan *Message
Out chan *Message
AudioTracks map[int]*webrtc.Track
VoiceIn chan []int16
VoiceInActive time.Time
VoiceInNotify time.Time
VoiceInTransmitting bool
VoiceInLock *sync.Mutex
VoiceOut []chan *rtp.Packet
VoiceOutClient []int
VoiceOutActive []time.Time
VoiceOutLock *sync.Mutex
AudioIn *audio.In
AudioOut *audio.Out
Channel *Channel
@ -51,10 +39,8 @@ func NewClient(conn *websocket.Conn) *Client {
PeerConnLock: new(sync.Mutex),
In: make(chan *Message, 10),
Out: make(chan *Message, 10),
AudioTracks: make(map[int]*webrtc.Track),
VoiceIn: make(chan []int16, 10),
VoiceInLock: new(sync.Mutex),
VoiceOutLock: new(sync.Mutex),
AudioIn: audio.NewIn(),
AudioOut: audio.NewOut(),
Terminated: make(chan bool)}
go c.handleRead()
@ -137,55 +123,19 @@ func (c *Client) Close() {
c.Out <- nil
// TODO Place behind debug/verbose flag
log.Printf("%+v", errors.New("Closing client #"+strconv.Itoa(c.ID)))
//log.Printf("%+v", errors.New("Closing client #"+strconv.Itoa(c.ID)))
go func() {
c.Terminated <- true
}()
}
func (c *Client) InitAudio() {
c.VoiceOutLock.Lock()
defer c.VoiceOutLock.Unlock()
if len(c.VoiceOut) > 0 {
return
}
for i := 0; i < 3; i++ {
c.VoiceOut = append(c.VoiceOut, make(chan *rtp.Packet, 10))
c.VoiceOutClient = append(c.VoiceOutClient, 0)
c.VoiceOutActive = append(c.VoiceOutActive, time.Time{})
}
}
func (c *Client) WriteAudio(p *rtp.Packet, source int) {
c.VoiceOutLock.Lock()
for i := range c.VoiceOut {
if c.VoiceOutClient[i] == 0 || c.VoiceOutClient[i] == source || time.Since(c.VoiceOutActive[i]) >= 50*time.Millisecond {
select {
case c.VoiceOut[i] <- p:
default:
log.Printf("client %d warning: filled voice out buffer when writing from %d", c.ID, source)
continue
}
c.VoiceOutActive[i] = time.Now()
c.VoiceOutClient[i] = source
c.VoiceOutLock.Unlock()
return
}
}
c.VoiceOutLock.Unlock()
}
func (c *Client) CloseAudio() {
c.ClosePeerConns()
c.PeerConns = make(map[int]*webrtc.PeerConnection)
c.VoiceOut = nil
c.VoiceOutClient = nil
c.VoiceOutActive = nil
c.AudioOut.Reset()
}
func (c *Client) ClosePeerConns() {
@ -195,22 +145,11 @@ func (c *Client) ClosePeerConns() {
}
func (c *Client) ClosePeerConn(id int) {
c.VoiceOutLock.Lock()
defer c.VoiceOutLock.Unlock()
pc := c.PeerConns[id]
if pc == nil {
return
}
select {
case c.VoiceOut[id] <- nil:
default:
log.Println("failed to close channel for client " + strconv.Itoa(c.ID))
}
pc.Close()
pc = nil
c.AudioTracks[id] = nil
}

2
pkg/web/message.go → pkg/agent/message.go

@ -1,4 +1,4 @@
package web
package agent
import "fmt"

2
pkg/web/user.go → pkg/agent/user.go

@ -1,4 +1,4 @@
package web
package agent
import (
"regexp"

56
pkg/audio/in.go

@ -0,0 +1,56 @@
package audio
import (
"sync"
"time"
)
const (
TransmitRepeat = 5 * time.Second
TransmitExpire = 100 * time.Millisecond
)
type In struct {
In chan []int16
Active time.Time
Notify time.Time
Transmitting bool
*sync.Mutex
}
func NewIn() *In {
in := In{
In: make(chan []int16, 10),
Mutex: new(sync.Mutex),
}
return &in
}
func (in *In) StartTransmit() bool {
in.Lock()
defer in.Unlock()
in.Active = time.Now()
if in.Transmitting && time.Since(in.Notify) < TransmitRepeat {
return false
}
in.Transmitting = true
in.Notify = time.Now()
return true
}
func (in *In) ExpireTransmit() bool {
in.Lock()
defer in.Unlock()
if !in.Transmitting || time.Since(in.Active) < TransmitExpire {
return false
}
in.Transmitting = false
in.Active = time.Time{}
in.Notify = time.Time{}
return true
}

88
pkg/audio/out.go

@ -0,0 +1,88 @@
package audio
import (
"log"
"sync"
"time"
"github.com/pion/rtp"
"github.com/pion/webrtc/v2"
"github.com/pion/webrtc/v2/pkg/media"
)
type Out struct {
Out []chan *rtp.Packet
Tracks []*webrtc.Track
Client []int
Active []time.Time
*sync.Mutex
}
func NewOut() *Out {
out := Out{
Mutex: new(sync.Mutex),
}
return &out
}
func (o *Out) AddTrack(track *webrtc.Track) {
o.Lock()
defer o.Unlock()
o.Tracks = append(o.Tracks, track)
o.Out = append(o.Out, make(chan *rtp.Packet, 10))
o.Client = append(o.Client, 0)
o.Active = append(o.Active, time.Time{})
go o.handleTrack(len(o.Tracks) - 1)
}
func (o *Out) handleTrack(i int) {
track := o.Tracks[i]
var err error
for p := range o.Out[i] {
if p == nil {
return
}
err = track.WriteSample(media.Sample{Data: p.Payload, Samples: Samples})
if err != nil {
panic(err)
}
}
}
func (o *Out) Write(p *rtp.Packet, source int) {
o.Lock()
defer o.Unlock()
for i := range o.Out {
if o.Client[i] == 0 || o.Client[i] == source || time.Since(o.Active[i]) >= 50*time.Millisecond {
select {
case o.Out[i] <- p:
default:
log.Printf("warning: filled voice out buffer when writing from %d", source)
continue
}
o.Active[i] = time.Now()
o.Client[i] = source
return
}
}
}
func (o *Out) Reset() {
o.Lock()
defer o.Unlock()
for i := range o.Out {
o.Out[i] <- nil
}
o.Tracks = nil
o.Out = nil
o.Client = nil
o.Active = nil
}

13
pkg/web/public/assets/js/harmony.js

@ -14,7 +14,7 @@ var numConnections = 3;
var peerConnections = [];
var RTCConstraints = {
audio: {
channelCount: {exact: 2},
channelCount: 2,
autoGainControl: false,
echoCancellation: false,
noiseSuppression: false,
@ -333,6 +333,10 @@ function Connect() {
var pc = peerConnections[p.PC];
pc.setRemoteDescription(new RTCSessionDescription({type: 'answer', sdp: p.M}));
if (peerConnections.length < 3) {
peerConnections.push(createPeerConnection(peerConnections.length));
}
} else if (p.T == MessageConnect) {
if (p.N === undefined) {
return;
@ -529,10 +533,7 @@ function JoinVoice(channelID) {
voice = true;
if (peerConnections.length == 0) {
var i;
for (i = 0; i < numConnections; i++) {
peerConnections.push(createPeerConnection(i));
}
peerConnections.push(createPeerConnection(0));
}
socket.send(JSON.stringify({T: MessageJoin, C: parseInt(channelID)}));
@ -616,7 +617,7 @@ function updateChannelList() {
continue;
}
$("#joinvoice" + c.ID).click(function (e) {
$("#joinvoice" + c.ID).on("click touchstart", function (e) {
if (voiceChannel == parseInt($(this).attr('id').substring(9))) {
QuitVoice();
return false;

171
pkg/web/web.go

@ -12,12 +12,12 @@ import (
"sync"
"time"
"git.sr.ht/~tslocum/harmony/pkg/agent"
"git.sr.ht/~tslocum/harmony/pkg/audio"
"github.com/gorilla/mux"
"github.com/gorilla/websocket"
"github.com/pion/rtp"
"github.com/pion/webrtc/v2"
"github.com/pion/webrtc/v2/pkg/media"
"github.com/pkg/errors"
"gitlab.com/golang-commonmark/markdown"
)
@ -32,7 +32,7 @@ var peerConnectionConfig = webrtc.Configuration{
var assets http.FileSystem
var incomingClients = make(chan *Client, 10)
var incomingClients = make(chan *agent.Client, 10)
var markdownRenderer = markdown.New(markdown.Typographer(false), markdown.Breaks(true), markdown.Quotes([]string{`"`, `"`, `'`, `'`}))
@ -46,10 +46,10 @@ var upgrader = websocket.Upgrader{
}
type WebInterface struct {
Clients map[int]*Client
Clients map[int]*agent.Client
ClientsLock *sync.Mutex
Channels map[int]*Channel
Channels map[int]*agent.Channel
ChannelsLock *sync.Mutex
}
@ -58,20 +58,17 @@ func NewWebInterface(address string, path string) *WebInterface {
panic("failed to load web assets")
}
w := WebInterface{Clients: make(map[int]*Client), ClientsLock: new(sync.Mutex), Channels: make(map[int]*Channel), ChannelsLock: new(sync.Mutex)}
w := WebInterface{Clients: make(map[int]*agent.Client), ClientsLock: new(sync.Mutex), Channels: make(map[int]*agent.Channel), ChannelsLock: new(sync.Mutex)}
w.createChannels()
r := mux.NewRouter()
go w.handleIncomingClients()
go w.handleExpireTransmit()
r := mux.NewRouter()
r.HandleFunc(path+"w", w.webSocketHandler)
r.PathPrefix(path).Handler(http.StripPrefix(path, http.FileServer(assets)))
go w.handleIncomingClients()
go w.handleExpireTransmit()
addressSplit := strings.Split(address, ",")
for _, add := range addressSplit {
add := add // Capture
@ -86,32 +83,32 @@ func NewWebInterface(address string, path string) *WebInterface {
return &w
}
func (w *WebInterface) nextChannelID() int {
id := 0
for cid := range w.Channels {
if cid > id {
id = cid
}
}
return id + 1
}
func (w *WebInterface) createChannels() {
w.ChannelsLock.Lock()
defer w.ChannelsLock.Unlock()
ch := NewChannel(w.nextChannelID(), ChannelAll)
ch := agent.NewChannel(w.nextChannelID(), agent.ChannelAll)
ch.Name = "lobby"
ch.Topic = "harmony demo server"
w.Channels[ch.ID] = ch
ch = NewChannel(w.nextChannelID(), ChannelAll)
ch = agent.NewChannel(w.nextChannelID(), agent.ChannelAll)
ch.Name = "alt"
ch.Topic = "alt demo channel"
w.Channels[ch.ID] = ch
}
func (w *WebInterface) nextChannelID() int {
id := 0
for cid := range w.Channels {
if cid > id {
id = cid
}
}
return id + 1
}
func (w *WebInterface) handleIncomingClients() {
for c := range incomingClients {
c := c // Capture
@ -121,7 +118,7 @@ func (w *WebInterface) handleIncomingClients() {
c.ID = id
w.Clients[id] = c
go func(c *Client) {
go func(c *agent.Client) {
time.Sleep(500 * time.Millisecond)
c.Connected = true
@ -130,7 +127,7 @@ func (w *WebInterface) handleIncomingClients() {
w.ClientsLock.Lock()
for _, wc := range w.Clients {
wc.Out <- &Message{T: MessageConnect, N: c.Name, M: []byte(c.Name)}
wc.Out <- &agent.Message{T: agent.MessageConnect, N: c.Name, M: []byte(c.Name)}
}
w.ClientsLock.Unlock()
}(c)
@ -148,50 +145,46 @@ func (w *WebInterface) handleExpireTransmit() {
for range t.C {
w.ClientsLock.Lock()
for _, wc := range w.Clients {
wc.VoiceInLock.Lock()
if wc.VoiceInTransmitting && time.Since(wc.VoiceInActive) >= 100*time.Millisecond {
wc.VoiceInTransmitting = false
for _, wcc := range w.Clients {
if len(wcc.AudioTracks) > 0 {
wcc.Out <- &Message{T: MessageTransmitStop, S: wc.ID}
if wc.AudioIn.ExpireTransmit() {
for _, wcc := range wc.Channel.Clients {
if len(wcc.AudioOut.Tracks) > 0 {
wcc.Out <- &agent.Message{T: agent.MessageTransmitStop, S: wc.ID}
}
}
}
wc.VoiceInLock.Unlock()
}
w.ClientsLock.Unlock()
}
}
func (w *WebInterface) handleRead(c *Client) {
func (w *WebInterface) handleRead(c *agent.Client) {
for msg := range c.In {
if msg == nil {
return
}
if msg.T != MessagePing {
if msg.T != agent.MessagePing {
log.Printf("%d -> %s %d", msg.S, msg.T, len(msg.M))
}
switch msg.T {
case MessageBinary:
case agent.MessageBinary:
// TODO Binary message
continue
case MessagePing:
c.Out <- &Message{T: MessagePong, M: msg.M}
case MessageCall:
case agent.MessagePing:
c.Out <- &agent.Message{T: agent.MessagePong, M: msg.M}
case agent.MessageCall:
answer, err := w.answerRTC(c, msg.PC, msg.M)
if err != nil {
log.Printf("failed to answer call: %s", err)
continue
}
c.Out <- &Message{T: MessageAnswer, PC: msg.PC, M: answer}
case MessageChat:
c.Out <- &agent.Message{T: agent.MessageAnswer, PC: msg.PC, M: answer}
case agent.MessageChat:
if bytes.HasPrefix(bytes.ToLower(msg.M), []byte("/nick ")) {
go func(mm []byte) {
c.In <- &Message{S: c.ID, T: MessageNick, M: mm}
c.In <- &agent.Message{S: c.ID, T: agent.MessageNick, M: mm}
}(msg.M[6:])
continue
@ -210,18 +203,18 @@ func (w *WebInterface) handleRead(c *Client) {
w.ClientsLock.Lock()
for _, wc := range w.Clients {
wc.Out <- &Message{S: c.ID, N: c.Name, T: MessageChat, M: msg.M}
wc.Out <- &agent.Message{S: c.ID, N: c.Name, T: agent.MessageChat, M: msg.M}
}
w.ClientsLock.Unlock()
case MessageNick:
case agent.MessageNick:
w.ClientsLock.Lock()
oldNick := c.Name
c.Name = Nickname(string(msg.M))
c.Name = agent.Nickname(string(msg.M))
if c.Connected {
msg := &Message{S: c.ID, N: oldNick, T: MessageNick, M: []byte(c.Name)}
msg := &agent.Message{S: c.ID, N: oldNick, T: agent.MessageNick, M: []byte(c.Name)}
for _, wc := range w.Clients {
wc.Out <- msg
}
@ -230,8 +223,8 @@ func (w *WebInterface) handleRead(c *Client) {
w.ClientsLock.Unlock()
w.updateUserList()
case MessageJoin, MessageQuit:
if msg.T == MessageJoin {
case agent.MessageJoin, agent.MessageQuit:
if msg.T == agent.MessageJoin {
w.joinChannel(c, w.Channels[msg.C])
} else { // MessageQuit
w.quitChannel(c)
@ -240,10 +233,10 @@ func (w *WebInterface) handleRead(c *Client) {
}
w.updateUserList()
case MessageConnect, MessageDisconnect:
case agent.MessageConnect, agent.MessageDisconnect:
w.ClientsLock.Lock()
if msg.T == MessageDisconnect {
if msg.T == agent.MessageDisconnect {
w.quitChannel(c)
c.Close()
@ -264,7 +257,7 @@ func (w *WebInterface) handleRead(c *Client) {
}
}
func (w *WebInterface) joinChannel(c *Client, ch *Channel) {
func (w *WebInterface) joinChannel(c *agent.Client, ch *agent.Channel) {
if ch == nil || (c.Channel != nil && c.Channel.ID == ch.ID) {
return
}
@ -278,11 +271,11 @@ func (w *WebInterface) joinChannel(c *Client, ch *Channel) {
c.Channel = ch
for _, wc := range ch.Clients {
if len(wc.AudioTracks) == 0 && wc.ID != c.ID {
if len(wc.AudioOut.Tracks) == 0 && wc.ID != c.ID {
continue
}
wc.Out <- &Message{T: MessageJoin, N: c.Name, C: ch.ID}
wc.Out <- &agent.Message{T: agent.MessageJoin, N: c.Name, C: ch.ID}
}
ch.Unlock()
@ -291,7 +284,7 @@ func (w *WebInterface) joinChannel(c *Client, ch *Channel) {
w.updateUserList()
}
func (w *WebInterface) quitChannel(c *Client) {
func (w *WebInterface) quitChannel(c *agent.Client) {
if c.Channel == nil {
return
}
@ -302,11 +295,11 @@ func (w *WebInterface) quitChannel(c *Client) {
ch.Lock()
for _, wc := range ch.Clients {
if len(wc.AudioTracks) == 0 && wc.ID != c.ID {
if len(wc.AudioOut.Tracks) == 0 && wc.ID != c.ID {
continue
}
wc.Out <- &Message{T: MessageQuit, N: c.Name, C: ch.ID}
wc.Out <- &agent.Message{T: agent.MessageQuit, N: c.Name, C: ch.ID}
}
delete(ch.Clients, c.ID)
@ -337,7 +330,7 @@ func (w *WebInterface) webSocketHandler(wr http.ResponseWriter, r *http.Request)
return
}
c := NewClient(conn)
c := agent.NewClient(conn)
incomingClients <- c
<-c.Terminated
@ -354,7 +347,7 @@ func (w *WebInterface) webSocketHandler(wr http.ResponseWriter, r *http.Request)
delete(w.Clients, id)
for _, wc := range w.Clients {
wc.Out <- &Message{T: MessageDisconnect, N: name, M: []byte(name)}
wc.Out <- &agent.Message{T: agent.MessageDisconnect, N: name, M: []byte(name)}
}
}
w.ClientsLock.Unlock()
@ -363,16 +356,16 @@ func (w *WebInterface) webSocketHandler(wr http.ResponseWriter, r *http.Request)
func (w *WebInterface) updateUserList() {
w.ClientsLock.Lock()
msg := &Message{T: MessageUsers}
msg := &agent.Message{T: agent.MessageUsers}
var userList UserList
var userList agent.UserList
for _, wc := range w.Clients {
c := 0
if wc.Channel != nil {
c = wc.Channel.ID
}
userList = append(userList, &User{ID: wc.ID, N: wc.Name, C: c})
userList = append(userList, &agent.User{ID: wc.ID, N: wc.Name, C: c})
}
sort.Sort(userList)
@ -390,13 +383,13 @@ func (w *WebInterface) updateUserList() {
w.ClientsLock.Unlock()
}
func (w *WebInterface) sendChannelList(c *Client) {
var channelList = make(ChannelList)
func (w *WebInterface) sendChannelList(c *agent.Client) {
var channelList = make(agent.ChannelList)
for _, ch := range w.Channels {
channelList[ch.ID] = &ChannelListing{ID: ch.ID, Type: ch.Type, Name: ch.Name, Topic: ch.Topic}
channelList[ch.ID] = &agent.ChannelListing{ID: ch.ID, Type: ch.Type, Name: ch.Name, Topic: ch.Topic}
}
msg := Message{T: MessageChannels}
msg := agent.Message{T: agent.MessageChannels}
var err error
msg.M, err = json.Marshal(channelList)
@ -407,7 +400,7 @@ func (w *WebInterface) sendChannelList(c *Client) {
c.Out <- &msg
}
func (w *WebInterface) answerRTC(c *Client, peerConnID int, offerSDP []byte) ([]byte, error) {
func (w *WebInterface) answerRTC(c *agent.Client, peerConnID int, offerSDP []byte) ([]byte, error) {
c.PeerConnLock.Lock()
defer c.PeerConnLock.Unlock()
@ -439,10 +432,6 @@ func (w *WebInterface) answerRTC(c *Client, peerConnID int, offerSDP []byte) ([]
panic(err)
}
if len(c.PeerConns) == 0 {
c.InitAudio()
}
c.PeerConns[peerConnID] = pc
err = pc.SetRemoteDescription(offer)
@ -464,7 +453,7 @@ func (w *WebInterface) answerRTC(c *Client, peerConnID int, offerSDP []byte) ([]
}
name := "harmony-audio-" + strconv.Itoa(peerConnID)
c.AudioTracks[peerConnID], err = pc.NewTrack(payloadType, 1000+uint32(peerConnID), name, name)
track, err := pc.NewTrack(payloadType, 1000+uint32(peerConnID), name, name)
if err != nil {
panic(err)
}
@ -473,27 +462,14 @@ func (w *WebInterface) answerRTC(c *Client, peerConnID int, offerSDP []byte) ([]
if peerConnID == 0 {
direction = webrtc.RTPTransceiverDirectionSendrecv
}
_, err = pc.AddTransceiverFromTrack(c.AudioTracks[peerConnID], webrtc.RtpTransceiverInit{Direction: direction})
_, err = pc.AddTransceiverFromTrack(track, webrtc.RtpTransceiverInit{Direction: direction})
if err != nil {
panic(err)
}
go func() {
for p := range c.VoiceOut[peerConnID] {
if p == nil {
return
}
err = c.AudioTracks[peerConnID].WriteSample(media.Sample{Data: p.Payload, Samples: audio.Samples})
if err != nil {
panic(err)
}
}
}()
c.AudioOut.AddTrack(track)
pc.OnTrack(func(remoteTrack *webrtc.Track, receiver *webrtc.RTPReceiver) {
log.Printf("client %d ontrack", c.ID)
var p *rtp.Packet
for {
p, err = remoteTrack.ReadRTP()
@ -509,28 +485,21 @@ func (w *WebInterface) answerRTC(c *Client, peerConnID int, offerSDP []byte) ([]
continue
}
c.VoiceInLock.Lock()
if !c.VoiceInTransmitting || time.Since(c.VoiceInNotify) >= 5*time.Second {
c.VoiceInTransmitting = true
c.VoiceInNotify = time.Now()
// TODO trim initial x ms transmitting to remove noise (configurable)
if c.AudioIn.StartTransmit() {
for _, wc := range c.Channel.Clients {
if len(wc.AudioTracks) > 0 {
wc.Out <- &Message{T: MessageTransmitStart, S: c.ID}
if len(wc.AudioOut.Tracks) > 0 {
wc.Out <- &agent.Message{T: agent.MessageTransmitStart, S: c.ID}
}
}
}
c.VoiceInActive = time.Now()
c.VoiceInLock.Unlock()
// TODO trim initial x ms transmitting to remove noise (configurable)
for ci, wc := range c.Channel.Clients {
if ci == c.ID {
continue
}
wc.WriteAudio(p, c.ID)
wc.AudioOut.Write(p, c.ID)
}
w.ClientsLock.Unlock()
}
@ -552,9 +521,9 @@ func (w *WebInterface) answerRTC(c *Client, peerConnID int, offerSDP []byte) ([]
}
})
pc.OnICEConnectionStateChange(func(connectionState webrtc.ICEConnectionState) {
/*pc.OnICEConnectionStateChange(func(connectionState webrtc.ICEConnectionState) {
log.Printf("%d ice state -> %s\n", c.ID, connectionState)
})
})*/
answerOptions := &webrtc.AnswerOptions{OfferAnswerOptions: webrtc.OfferAnswerOptions{VoiceActivityDetection: false}}

Loading…
Cancel
Save