mirror of
https://github.com/jech/galene.git
synced 2024-11-09 02:05:59 +01:00
8a2357de21
When we moved the chat history to the client goroutine back in 0.6, we started sending the history whenever the group configuration changed. Only send it when first joining the group. Thanks to Rémi Nollet.
1940 lines
40 KiB
Go
1940 lines
40 KiB
Go
package rtpconn
|
|
|
|
import (
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"log"
|
|
"os"
|
|
"sync"
|
|
"time"
|
|
|
|
"github.com/gorilla/websocket"
|
|
"github.com/pion/webrtc/v3"
|
|
|
|
"github.com/jech/galene/conn"
|
|
"github.com/jech/galene/diskwriter"
|
|
"github.com/jech/galene/estimator"
|
|
"github.com/jech/galene/group"
|
|
"github.com/jech/galene/ice"
|
|
)
|
|
|
|
func errorToWSCloseMessage(id string, err error) (*clientMessage, []byte) {
|
|
var code int
|
|
var m *clientMessage
|
|
var text string
|
|
switch e := err.(type) {
|
|
case *websocket.CloseError:
|
|
code = websocket.CloseNormalClosure
|
|
case group.ProtocolError:
|
|
code = websocket.CloseProtocolError
|
|
m = &clientMessage{
|
|
Type: "usermessage",
|
|
Kind: "error",
|
|
Dest: id,
|
|
Privileged: true,
|
|
Value: e.Error(),
|
|
}
|
|
text = e.Error()
|
|
case group.UserError, group.KickError:
|
|
code = websocket.CloseNormalClosure
|
|
m = errorMessage(id, err)
|
|
text = e.Error()
|
|
default:
|
|
code = websocket.CloseInternalServerErr
|
|
}
|
|
return m, websocket.FormatCloseMessage(code, text)
|
|
}
|
|
|
|
func isWSNormalError(err error) bool {
|
|
return websocket.IsCloseError(err,
|
|
websocket.CloseNormalClosure,
|
|
websocket.CloseGoingAway)
|
|
}
|
|
|
|
type webClient struct {
|
|
group *group.Group
|
|
id string
|
|
username string
|
|
permissions []string
|
|
data map[string]interface{}
|
|
requested map[string][]string
|
|
done chan struct{}
|
|
writeCh chan interface{}
|
|
writerDone chan struct{}
|
|
actionCh chan struct{}
|
|
|
|
mu sync.Mutex
|
|
down map[string]*rtpDownConnection
|
|
up map[string]*rtpUpConnection
|
|
|
|
// action may be called with the group mutex taken, and therefore
|
|
// actions needs to use its own mutex.
|
|
actionMu sync.Mutex
|
|
actions []interface{}
|
|
}
|
|
|
|
func (c *webClient) Group() *group.Group {
|
|
return c.group
|
|
}
|
|
|
|
func (c *webClient) Id() string {
|
|
return c.id
|
|
}
|
|
|
|
func (c *webClient) Username() string {
|
|
return c.username
|
|
}
|
|
|
|
func (c *webClient) SetUsername(username string) {
|
|
c.username = username
|
|
}
|
|
|
|
func (c *webClient) Permissions() []string {
|
|
return c.permissions
|
|
}
|
|
|
|
func (c *webClient) Data() map[string]interface{} {
|
|
return c.data
|
|
}
|
|
|
|
func (c *webClient) SetPermissions(perms []string) {
|
|
c.permissions = perms
|
|
}
|
|
|
|
func (c *webClient) PushClient(group, kind, id, username string, perms []string, data map[string]interface{}) error {
|
|
return c.action(pushClientAction{
|
|
group, kind, id, username, perms, data,
|
|
})
|
|
}
|
|
|
|
type clientMessage struct {
|
|
Type string `json:"type"`
|
|
Version []string `json:"version"`
|
|
Kind string `json:"kind,omitempty"`
|
|
Id string `json:"id,omitempty"`
|
|
Replace string `json:"replace,omitempty"`
|
|
Source string `json:"source,omitempty"`
|
|
Dest string `json:"dest,omitempty"`
|
|
Username string `json:"username,omitempty"`
|
|
Password string `json:"password,omitempty"`
|
|
Token string `json:"token,omitempty"`
|
|
Privileged bool `json:"privileged,omitempty"`
|
|
Permissions []string `json:"permissions,omitempty"`
|
|
Status *group.Status `json:"status,omitempty"`
|
|
Data map[string]interface{} `json:"data,omitempty"`
|
|
Group string `json:"group,omitempty"`
|
|
Value interface{} `json:"value,omitempty"`
|
|
NoEcho bool `json:"noecho,omitempty"`
|
|
Time int64 `json:"time,omitempty"`
|
|
SDP string `json:"sdp,omitempty"`
|
|
Candidate *webrtc.ICECandidateInit `json:"candidate,omitempty"`
|
|
Label string `json:"label,omitempty"`
|
|
Request interface{} `json:"request,omitempty"`
|
|
RTCConfiguration *webrtc.Configuration `json:"rtcConfiguration,omitempty"`
|
|
}
|
|
|
|
type closeMessage struct {
|
|
data []byte
|
|
}
|
|
|
|
func getUpConn(c *webClient, id string) *rtpUpConnection {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
|
|
if c.up == nil {
|
|
return nil
|
|
}
|
|
return c.up[id]
|
|
}
|
|
|
|
func getUpConns(c *webClient) []*rtpUpConnection {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
up := make([]*rtpUpConnection, 0, len(c.up))
|
|
for _, u := range c.up {
|
|
up = append(up, u)
|
|
}
|
|
return up
|
|
}
|
|
|
|
func addUpConn(c *webClient, id, label string, offer string) (*rtpUpConnection, bool, error) {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
|
|
if c.up == nil {
|
|
c.up = make(map[string]*rtpUpConnection)
|
|
}
|
|
if c.down != nil && c.down[id] != nil {
|
|
return nil, false, errors.New("adding duplicate connection")
|
|
}
|
|
|
|
old := c.up[id]
|
|
if old != nil {
|
|
return old, false, nil
|
|
}
|
|
|
|
conn, err := newUpConn(c, id, label, offer)
|
|
if err != nil {
|
|
return nil, false, err
|
|
}
|
|
|
|
c.up[id] = conn
|
|
|
|
conn.pc.OnICECandidate(func(candidate *webrtc.ICECandidate) {
|
|
sendICE(c, id, candidate)
|
|
})
|
|
|
|
conn.pc.OnICEConnectionStateChange(func(state webrtc.ICEConnectionState) {
|
|
if state == webrtc.ICEConnectionStateFailed {
|
|
c.action(connectionFailedAction{id: id})
|
|
}
|
|
})
|
|
|
|
return conn, true, nil
|
|
}
|
|
|
|
var ErrUserMismatch = errors.New("user id mismatch")
|
|
|
|
// delUpConn deletes an up connection. If push is closed, the close is
|
|
// pushed to all corresponding down connections.
|
|
func delUpConn(c *webClient, id string, userId string, push bool) error {
|
|
c.mu.Lock()
|
|
if c.up == nil {
|
|
c.mu.Unlock()
|
|
return os.ErrNotExist
|
|
}
|
|
conn := c.up[id]
|
|
if conn == nil {
|
|
c.mu.Unlock()
|
|
return os.ErrNotExist
|
|
}
|
|
if userId != "" {
|
|
id, _ := conn.User()
|
|
if id != userId {
|
|
c.mu.Unlock()
|
|
return ErrUserMismatch
|
|
}
|
|
}
|
|
|
|
replace := conn.getReplace(false)
|
|
|
|
delete(c.up, id)
|
|
g := c.group
|
|
c.mu.Unlock()
|
|
|
|
conn.mu.Lock()
|
|
conn.closed = true
|
|
conn.mu.Unlock()
|
|
|
|
conn.pc.Close()
|
|
|
|
if push && g != nil {
|
|
for _, c := range g.GetClients(c) {
|
|
err := c.PushConn(g, id, nil, nil, replace)
|
|
if err != nil {
|
|
log.Printf("PushConn: %v", err)
|
|
}
|
|
}
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func getDownConn(c *webClient, id string) *rtpDownConnection {
|
|
if c.down == nil {
|
|
return nil
|
|
}
|
|
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
conn := c.down[id]
|
|
if conn == nil {
|
|
return nil
|
|
}
|
|
return conn
|
|
}
|
|
|
|
func getConn(c *webClient, id string) iceConnection {
|
|
up := getUpConn(c, id)
|
|
if up != nil {
|
|
return up
|
|
}
|
|
down := getDownConn(c, id)
|
|
if down != nil {
|
|
return down
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func addDownConn(c *webClient, remote conn.Up) (*rtpDownConnection, bool, error) {
|
|
id := remote.Id()
|
|
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
|
|
if c.up != nil && c.up[id] != nil {
|
|
return nil, false, errors.New("adding duplicate connection")
|
|
}
|
|
|
|
if c.down == nil {
|
|
c.down = make(map[string]*rtpDownConnection)
|
|
}
|
|
|
|
if down := c.down[id]; down != nil {
|
|
return down, false, nil
|
|
}
|
|
|
|
down, err := newDownConn(c, id, remote)
|
|
if err != nil {
|
|
return nil, false, err
|
|
}
|
|
|
|
down.pc.OnICECandidate(func(candidate *webrtc.ICECandidate) {
|
|
sendICE(c, down.id, candidate)
|
|
})
|
|
|
|
down.pc.OnICEConnectionStateChange(func(state webrtc.ICEConnectionState) {
|
|
if state == webrtc.ICEConnectionStateFailed {
|
|
c.action(connectionFailedAction{id: down.id})
|
|
}
|
|
})
|
|
|
|
err = remote.AddLocal(down)
|
|
if err != nil {
|
|
down.pc.Close()
|
|
return nil, false, err
|
|
}
|
|
|
|
c.down[down.id] = down
|
|
|
|
go rtcpDownSender(down)
|
|
|
|
return down, true, nil
|
|
}
|
|
|
|
func delDownConn(c *webClient, id string) error {
|
|
conn := delDownConnHelper(c, id)
|
|
if conn != nil {
|
|
conn.pc.Close()
|
|
return nil
|
|
}
|
|
return os.ErrNotExist
|
|
}
|
|
|
|
func delDownConnHelper(c *webClient, id string) *rtpDownConnection {
|
|
c.mu.Lock()
|
|
defer c.mu.Unlock()
|
|
|
|
if c.down == nil {
|
|
return nil
|
|
}
|
|
conn := c.down[id]
|
|
if conn == nil {
|
|
return nil
|
|
}
|
|
|
|
conn.remote.DelLocal(conn)
|
|
for _, track := range conn.tracks {
|
|
// we only insert the track after we get an answer, so
|
|
// ignore errors here.
|
|
track.remote.DelLocal(track)
|
|
}
|
|
delete(c.down, id)
|
|
return conn
|
|
}
|
|
|
|
var errUnexpectedTrackType = errors.New("unexpected track type, this shouldn't happen")
|
|
|
|
func addDownTrackUnlocked(conn *rtpDownConnection, remoteTrack *rtpUpTrack) error {
|
|
for _, t := range conn.tracks {
|
|
tt, ok := t.remote.(*rtpUpTrack)
|
|
if !ok {
|
|
return errUnexpectedTrackType
|
|
}
|
|
if tt == remoteTrack {
|
|
return os.ErrExist
|
|
}
|
|
}
|
|
|
|
id := remoteTrack.track.ID()
|
|
if id == "" {
|
|
log.Println("Got track with empty id")
|
|
id = remoteTrack.track.RID()
|
|
}
|
|
if id == "" {
|
|
id = remoteTrack.track.Kind().String()
|
|
}
|
|
msid := remoteTrack.track.StreamID()
|
|
if msid == "" || msid == "-" {
|
|
log.Println("Got track with empty msid")
|
|
msid = remoteTrack.conn.Label()
|
|
}
|
|
if msid == "" {
|
|
msid = "dummy"
|
|
}
|
|
|
|
local, err := webrtc.NewTrackLocalStaticRTP(
|
|
remoteTrack.Codec(), id, msid,
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
transceiver, err := conn.pc.AddTransceiverFromTrack(local,
|
|
webrtc.RTPTransceiverInit{
|
|
Direction: webrtc.RTPTransceiverDirectionSendonly,
|
|
},
|
|
)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
codec := local.Codec()
|
|
ptype, err := group.CodecPayloadType(local.Codec())
|
|
if err != nil {
|
|
log.Printf("Couldn't determine ptype for codec %v: %v",
|
|
codec.MimeType, err)
|
|
} else {
|
|
err := transceiver.SetCodecPreferences(
|
|
[]webrtc.RTPCodecParameters{
|
|
{
|
|
RTPCodecCapability: codec,
|
|
PayloadType: ptype,
|
|
},
|
|
},
|
|
)
|
|
if err != nil {
|
|
log.Printf("Couldn't set ptype for codec %v: %v",
|
|
codec.MimeType, err)
|
|
}
|
|
}
|
|
|
|
parms := transceiver.Sender().GetParameters()
|
|
if len(parms.Encodings) != 1 {
|
|
return errors.New("got multiple encodings")
|
|
}
|
|
|
|
track := &rtpDownTrack{
|
|
track: local,
|
|
sender: transceiver.Sender(),
|
|
ssrc: parms.Encodings[0].SSRC,
|
|
conn: conn,
|
|
remote: remoteTrack,
|
|
maxBitrate: new(bitrate),
|
|
maxREMBBitrate: new(bitrate),
|
|
stats: new(receiverStats),
|
|
rate: estimator.New(time.Second),
|
|
atomics: &downTrackAtomics{},
|
|
}
|
|
|
|
conn.tracks = append(conn.tracks, track)
|
|
|
|
go rtcpDownListener(track)
|
|
|
|
return nil
|
|
}
|
|
|
|
func delDownTrackUnlocked(conn *rtpDownConnection, track *rtpDownTrack) error {
|
|
for i := range conn.tracks {
|
|
if conn.tracks[i] == track {
|
|
track.remote.DelLocal(track)
|
|
conn.tracks =
|
|
append(conn.tracks[:i], conn.tracks[i+1:]...)
|
|
return conn.pc.RemoveTrack(track.sender)
|
|
}
|
|
}
|
|
return os.ErrNotExist
|
|
}
|
|
|
|
func replaceTracks(conn *rtpDownConnection, remote []conn.UpTrack, limitSid bool) (bool, error) {
|
|
conn.mu.Lock()
|
|
defer conn.mu.Unlock()
|
|
|
|
var add []*rtpUpTrack
|
|
var del []*rtpDownTrack
|
|
|
|
outer:
|
|
for _, rtrack := range remote {
|
|
rt, ok := rtrack.(*rtpUpTrack)
|
|
if !ok {
|
|
return false, errUnexpectedTrackType
|
|
}
|
|
for _, track := range conn.tracks {
|
|
rt2, ok := track.remote.(*rtpUpTrack)
|
|
if !ok {
|
|
return false, errUnexpectedTrackType
|
|
}
|
|
if rt == rt2 {
|
|
continue outer
|
|
}
|
|
}
|
|
add = append(add, rt)
|
|
}
|
|
|
|
outer2:
|
|
for _, track := range conn.tracks {
|
|
rt, ok := track.remote.(*rtpUpTrack)
|
|
if !ok {
|
|
return false, errUnexpectedTrackType
|
|
}
|
|
for _, rtrack := range remote {
|
|
rt2, ok := rtrack.(*rtpUpTrack)
|
|
if !ok {
|
|
return false, errUnexpectedTrackType
|
|
}
|
|
if rt == rt2 {
|
|
continue outer2
|
|
}
|
|
}
|
|
del = append(del, track)
|
|
}
|
|
|
|
defer func() {
|
|
for _, t := range conn.tracks {
|
|
layer := t.getLayerInfo()
|
|
layer.limitSid = limitSid
|
|
if limitSid {
|
|
layer.wantedSid = 0
|
|
}
|
|
t.setLayerInfo(layer)
|
|
}
|
|
}()
|
|
|
|
if len(del) == 0 && len(add) == 0 {
|
|
return false, nil
|
|
}
|
|
|
|
for _, t := range del {
|
|
err := delDownTrackUnlocked(conn, t)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
}
|
|
|
|
for _, rt := range add {
|
|
err := addDownTrackUnlocked(conn, rt)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
}
|
|
|
|
return true, nil
|
|
}
|
|
|
|
func negotiate(c *webClient, down *rtpDownConnection, restartIce bool, replace string) error {
|
|
if down.pc.SignalingState() == webrtc.SignalingStateHaveLocalOffer {
|
|
// avoid sending multiple offers back-to-back
|
|
if restartIce {
|
|
down.negotiationNeeded = negotiationRestartIce
|
|
} else if down.negotiationNeeded == negotiationUnneeded {
|
|
down.negotiationNeeded = negotiationNeeded
|
|
}
|
|
return nil
|
|
}
|
|
|
|
down.negotiationNeeded = negotiationUnneeded
|
|
|
|
options := webrtc.OfferOptions{ICERestart: restartIce}
|
|
offer, err := down.pc.CreateOffer(&options)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = down.pc.SetLocalDescription(offer)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
source, username := down.remote.User()
|
|
|
|
return c.write(clientMessage{
|
|
Type: "offer",
|
|
Id: down.id,
|
|
Label: down.remote.Label(),
|
|
Replace: replace,
|
|
Source: source,
|
|
Username: username,
|
|
SDP: down.pc.LocalDescription().SDP,
|
|
})
|
|
}
|
|
|
|
func sendICE(c *webClient, id string, candidate *webrtc.ICECandidate) error {
|
|
if candidate == nil {
|
|
return nil
|
|
}
|
|
cand := candidate.ToJSON()
|
|
return c.write(clientMessage{
|
|
Type: "ice",
|
|
Id: id,
|
|
Candidate: &cand,
|
|
})
|
|
}
|
|
|
|
func gotOffer(c *webClient, id, label string, sdp string, replace string) error {
|
|
up, _, err := addUpConn(c, id, label, sdp)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if replace != "" {
|
|
up.replace = replace
|
|
delUpConn(c, replace, c.Id(), false)
|
|
}
|
|
|
|
err = up.pc.SetRemoteDescription(webrtc.SessionDescription{
|
|
Type: webrtc.SDPTypeOffer,
|
|
SDP: sdp,
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
answer, err := up.pc.CreateAnswer(nil)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = up.pc.SetLocalDescription(answer)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = up.flushICECandidates()
|
|
if err != nil {
|
|
log.Printf("ICE: %v", err)
|
|
}
|
|
|
|
return c.write(clientMessage{
|
|
Type: "answer",
|
|
Id: id,
|
|
SDP: up.pc.LocalDescription().SDP,
|
|
})
|
|
}
|
|
|
|
var ErrUnknownId = errors.New("unknown id")
|
|
|
|
func gotAnswer(c *webClient, id string, sdp string) error {
|
|
down := getDownConn(c, id)
|
|
if down == nil {
|
|
return ErrUnknownId
|
|
}
|
|
|
|
err := down.pc.SetRemoteDescription(webrtc.SessionDescription{
|
|
Type: webrtc.SDPTypeAnswer,
|
|
SDP: sdp,
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
err = down.flushICECandidates()
|
|
if err != nil {
|
|
log.Printf("ICE: %v", err)
|
|
}
|
|
|
|
add := func() {
|
|
down.pc.OnConnectionStateChange(nil)
|
|
for _, t := range down.tracks {
|
|
err := t.remote.AddLocal(t)
|
|
if err != nil && err != os.ErrClosed {
|
|
log.Printf("Add track: %v", err)
|
|
}
|
|
}
|
|
}
|
|
down.pc.OnConnectionStateChange(func(state webrtc.PeerConnectionState) {
|
|
if state == webrtc.PeerConnectionStateConnected {
|
|
add()
|
|
}
|
|
})
|
|
if down.pc.ConnectionState() == webrtc.PeerConnectionStateConnected {
|
|
add()
|
|
}
|
|
|
|
return nil
|
|
}
|
|
|
|
func gotICE(c *webClient, candidate *webrtc.ICECandidateInit, id string) error {
|
|
conn := getConn(c, id)
|
|
if conn == nil {
|
|
return errors.New("unknown id in ICE")
|
|
}
|
|
return conn.addICECandidate(candidate)
|
|
}
|
|
|
|
var errBadType = errors.New("bad type")
|
|
|
|
func toStringArray(r interface{}) ([]string, error) {
|
|
if r == nil {
|
|
return nil, nil
|
|
}
|
|
rr, ok := r.([]interface{})
|
|
if !ok {
|
|
return nil, errBadType
|
|
}
|
|
if rr == nil {
|
|
return nil, nil
|
|
}
|
|
|
|
rrr := make([]string, len(rr))
|
|
for i, s := range rr {
|
|
rrr[i], ok = s.(string)
|
|
if !ok {
|
|
return nil, errBadType
|
|
}
|
|
}
|
|
return rrr, nil
|
|
}
|
|
|
|
func parseRequested(r interface{}) (map[string][]string, error) {
|
|
if r == nil {
|
|
return nil, nil
|
|
}
|
|
rr, ok := r.(map[string]interface{})
|
|
if !ok {
|
|
return nil, errBadType
|
|
}
|
|
if rr == nil {
|
|
return nil, nil
|
|
}
|
|
rrr := make(map[string][]string)
|
|
for k, v := range rr {
|
|
vv, err := toStringArray(v)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
rrr[k] = vv
|
|
}
|
|
return rrr, nil
|
|
}
|
|
|
|
func (c *webClient) setRequested(requested map[string][]string) error {
|
|
if c.group == nil {
|
|
return errors.New("attempted to request with no group joined")
|
|
}
|
|
c.requested = requested
|
|
|
|
requestConns(c, c.group, "")
|
|
return nil
|
|
}
|
|
|
|
func (c *webClient) setRequestedStream(down *rtpDownConnection, requested []string) error {
|
|
var remoteClient group.Client
|
|
remote, ok := down.remote.(*rtpUpConnection)
|
|
if ok {
|
|
remoteClient = remote.client
|
|
}
|
|
down.requested = requested
|
|
return remoteClient.RequestConns(c, c.group, remote.id)
|
|
}
|
|
|
|
func (c *webClient) RequestConns(target group.Client, g *group.Group, id string) error {
|
|
return c.action(requestConnsAction{g, target, id})
|
|
}
|
|
|
|
func requestConns(target group.Client, g *group.Group, id string) {
|
|
clients := g.GetClients(target)
|
|
for _, c := range clients {
|
|
c.RequestConns(target, g, id)
|
|
}
|
|
}
|
|
|
|
func requestedTracks(c *webClient, override []string, up conn.Up, tracks []conn.UpTrack) ([]conn.UpTrack, bool) {
|
|
r := override
|
|
if r == nil {
|
|
var ok bool
|
|
r, ok = c.requested[up.Label()]
|
|
if !ok {
|
|
r, ok = c.requested[""]
|
|
}
|
|
if !ok || len(r) == 0 {
|
|
return nil, false
|
|
}
|
|
}
|
|
|
|
var audio, video, videoLow bool
|
|
for _, s := range r {
|
|
switch s {
|
|
case "audio":
|
|
audio = true
|
|
case "video":
|
|
video = true
|
|
case "video-low":
|
|
videoLow = true
|
|
default:
|
|
log.Printf("client requested unknown value %v", s)
|
|
}
|
|
}
|
|
|
|
find := func(kind webrtc.RTPCodecType, labels ...string) conn.UpTrack {
|
|
for _, l := range labels {
|
|
for _, t := range tracks {
|
|
if t.Kind() != kind {
|
|
continue
|
|
}
|
|
if t.Label() == l {
|
|
return t
|
|
}
|
|
}
|
|
}
|
|
for _, t := range tracks {
|
|
if t.Kind() != kind {
|
|
continue
|
|
}
|
|
return t
|
|
}
|
|
return nil
|
|
}
|
|
|
|
var ts []conn.UpTrack
|
|
limitSid := false
|
|
if audio {
|
|
t := find(webrtc.RTPCodecTypeAudio)
|
|
if t != nil {
|
|
ts = append(ts, t)
|
|
}
|
|
}
|
|
if video {
|
|
t := find(
|
|
webrtc.RTPCodecTypeVideo, "h", "m", "video",
|
|
)
|
|
if t != nil {
|
|
ts = append(ts, t)
|
|
}
|
|
} else if videoLow {
|
|
t := find(
|
|
webrtc.RTPCodecTypeVideo, "l", "m", "video",
|
|
)
|
|
if t != nil {
|
|
ts = append(ts, t)
|
|
if t.Label() != "l" {
|
|
limitSid = true
|
|
}
|
|
}
|
|
}
|
|
|
|
return ts, limitSid
|
|
}
|
|
|
|
func (c *webClient) PushConn(g *group.Group, id string, up conn.Up, tracks []conn.UpTrack, replace string) error {
|
|
err := c.action(pushConnAction{g, id, up, tracks, replace})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func readMessage(conn *websocket.Conn, m *clientMessage) error {
|
|
err := conn.SetReadDeadline(time.Now().Add(15 * time.Second))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer conn.SetReadDeadline(time.Time{})
|
|
|
|
return conn.ReadJSON(&m)
|
|
}
|
|
|
|
func StartClient(conn *websocket.Conn) (err error) {
|
|
var m clientMessage
|
|
|
|
err = readMessage(conn, &m)
|
|
if err != nil {
|
|
conn.Close()
|
|
return
|
|
}
|
|
|
|
if m.Type != "handshake" {
|
|
conn.WriteMessage(websocket.CloseMessage,
|
|
websocket.FormatCloseMessage(
|
|
websocket.CloseProtocolError,
|
|
"you must handshake first",
|
|
),
|
|
)
|
|
conn.Close()
|
|
err = group.ProtocolError("client didn't handshake")
|
|
return
|
|
}
|
|
|
|
versionError := true
|
|
if m.Version != nil {
|
|
for _, v := range m.Version {
|
|
if v == "1" {
|
|
versionError = false
|
|
}
|
|
}
|
|
}
|
|
|
|
c := &webClient{
|
|
id: m.Id,
|
|
actionCh: make(chan struct{}, 1),
|
|
done: make(chan struct{}),
|
|
}
|
|
|
|
defer close(c.done)
|
|
|
|
c.writeCh = make(chan interface{}, 100)
|
|
c.writerDone = make(chan struct{})
|
|
go clientWriter(conn, c.writeCh, c.writerDone)
|
|
defer func() {
|
|
m, e := errorToWSCloseMessage(c.id, err)
|
|
if isWSNormalError(err) {
|
|
err = nil
|
|
} else if _, ok := err.(group.KickError); ok {
|
|
err = nil
|
|
}
|
|
if m != nil {
|
|
c.write(*m)
|
|
}
|
|
c.close(e)
|
|
}()
|
|
|
|
return clientLoop(c, conn, versionError)
|
|
}
|
|
|
|
type pushConnAction struct {
|
|
group *group.Group
|
|
id string
|
|
conn conn.Up
|
|
tracks []conn.UpTrack
|
|
replace string
|
|
}
|
|
|
|
type requestConnsAction struct {
|
|
group *group.Group
|
|
target group.Client
|
|
id string
|
|
}
|
|
|
|
type connectionFailedAction struct {
|
|
id string
|
|
}
|
|
|
|
type pushClientAction struct {
|
|
group string
|
|
kind string
|
|
id string
|
|
username string
|
|
permissions []string
|
|
data map[string]interface{}
|
|
}
|
|
|
|
type permissionsChangedAction struct{}
|
|
|
|
type joinedAction struct {
|
|
group string
|
|
kind string
|
|
}
|
|
|
|
type kickAction struct {
|
|
id string
|
|
username string
|
|
message string
|
|
}
|
|
|
|
var errEmptyId = group.ProtocolError("empty id")
|
|
|
|
func member(v string, l []string) bool {
|
|
for _, w := range l {
|
|
if v == w {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func remove(v string, l []string) []string {
|
|
for i, w := range l {
|
|
if v == w {
|
|
l = append(l[:i], l[i+1:]...)
|
|
return l
|
|
}
|
|
}
|
|
return l
|
|
}
|
|
|
|
func addnew(v string, l []string) []string {
|
|
if member(v, l) {
|
|
return l
|
|
}
|
|
l = append(l, v)
|
|
return l
|
|
}
|
|
|
|
func clientLoop(c *webClient, ws *websocket.Conn, versionError bool) error {
|
|
read := make(chan interface{}, 1)
|
|
go clientReader(ws, read, c.done)
|
|
|
|
defer leaveGroup(c)
|
|
|
|
readTime := time.Now()
|
|
|
|
ticker := time.NewTicker(10 * time.Second)
|
|
defer ticker.Stop()
|
|
|
|
err := c.write(clientMessage{
|
|
Type: "handshake",
|
|
Version: []string{"1"},
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if versionError {
|
|
c.write(clientMessage{
|
|
Type: "usermessage",
|
|
Kind: "warning",
|
|
Dest: c.id,
|
|
Privileged: true,
|
|
Value: "This client is using an unknown protocol version.\n" +
|
|
"Perhaps it needs upgrading?\n" +
|
|
"Trying to continue, things may break.",
|
|
})
|
|
}
|
|
|
|
for {
|
|
select {
|
|
case m, ok := <-read:
|
|
if !ok {
|
|
return errors.New("reader died")
|
|
}
|
|
switch m := m.(type) {
|
|
case clientMessage:
|
|
readTime = time.Now()
|
|
err := handleClientMessage(c, m)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
case error:
|
|
return m
|
|
}
|
|
case <-c.actionCh:
|
|
c.actionMu.Lock()
|
|
actions := c.actions
|
|
c.actions = nil
|
|
c.actionMu.Unlock()
|
|
for _, a := range actions {
|
|
err := handleAction(c, a)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
case <-ticker.C:
|
|
if time.Since(readTime) > 75*time.Second {
|
|
return errors.New("client is dead")
|
|
}
|
|
// Some reverse proxies timeout connexions at 60
|
|
// seconds, make sure we generate some activity
|
|
// after 55s at most.
|
|
if time.Since(readTime) > 45*time.Second {
|
|
err := c.write(clientMessage{
|
|
Type: "ping",
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
func pushDownConn(c *webClient, id string, up conn.Up, tracks []conn.UpTrack, replace string) error {
|
|
var requested []conn.UpTrack
|
|
limitSid := false
|
|
if up != nil {
|
|
var old *rtpDownConnection
|
|
if replace != "" {
|
|
old = getDownConn(c, replace)
|
|
} else {
|
|
old = getDownConn(c, up.Id())
|
|
}
|
|
var override []string
|
|
if old != nil {
|
|
override = old.requested
|
|
}
|
|
requested, limitSid = requestedTracks(c, override, up, tracks)
|
|
}
|
|
|
|
if replace != "" {
|
|
err := delDownConn(c, replace)
|
|
if err != nil {
|
|
log.Printf("Replace: %v", err)
|
|
}
|
|
}
|
|
|
|
// closes over replace, which will be modified below
|
|
defer func() {
|
|
if replace != "" {
|
|
closeDownConn(c, replace, "")
|
|
}
|
|
}()
|
|
|
|
if len(requested) == 0 {
|
|
closeDownConn(c, id, "")
|
|
return nil
|
|
}
|
|
|
|
down, _, err := addDownConn(c, up)
|
|
if err != nil {
|
|
if err == os.ErrClosed {
|
|
return nil
|
|
}
|
|
return err
|
|
}
|
|
done, err := replaceTracks(down, requested, limitSid)
|
|
if err != nil || !done {
|
|
return err
|
|
}
|
|
err = negotiate(c, down, false, replace)
|
|
if err != nil {
|
|
log.Printf("Negotiation failed: %v", err)
|
|
closeDownConn(c, down.id, err.Error())
|
|
return err
|
|
}
|
|
replace = ""
|
|
return nil
|
|
}
|
|
|
|
func handleAction(c *webClient, a interface{}) error {
|
|
switch a := a.(type) {
|
|
case pushConnAction:
|
|
if c.group == nil || c.group != a.group {
|
|
log.Printf("Got connectsions for wrong group")
|
|
return nil
|
|
}
|
|
return pushDownConn(c, a.id, a.conn, a.tracks, a.replace)
|
|
case requestConnsAction:
|
|
g := c.group
|
|
if g == nil || a.group != g {
|
|
log.Printf("Misdirected pushConns")
|
|
return nil
|
|
}
|
|
for _, u := range c.up {
|
|
if a.id != "" && a.id != u.id {
|
|
continue
|
|
}
|
|
tracks := u.getTracks()
|
|
replace := u.getReplace(false)
|
|
|
|
ts := make([]conn.UpTrack, len(tracks))
|
|
for i, t := range tracks {
|
|
ts[i] = t
|
|
}
|
|
err := a.target.PushConn(g, u.id, u, ts, replace)
|
|
if err != nil {
|
|
log.Printf("PushConn: %v", err)
|
|
}
|
|
}
|
|
case connectionFailedAction:
|
|
if down := getDownConn(c, a.id); down != nil {
|
|
err := negotiate(c, down, true, "")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
tracks := make(
|
|
[]conn.UpTrack, len(down.tracks),
|
|
)
|
|
for i, t := range down.tracks {
|
|
tracks[i] = t.remote
|
|
}
|
|
c.PushConn(
|
|
c.group,
|
|
down.remote.Id(), down.remote,
|
|
tracks, "",
|
|
)
|
|
} else if up := getUpConn(c, a.id); up != nil {
|
|
c.write(clientMessage{
|
|
Type: "renegotiate",
|
|
Id: a.id,
|
|
})
|
|
} else {
|
|
log.Printf("Attempting to renegotiate " +
|
|
"unknown connection")
|
|
}
|
|
|
|
case pushClientAction:
|
|
if a.group != c.group.Name() {
|
|
log.Printf("got client for wrong group")
|
|
return nil
|
|
}
|
|
perms := append([]string(nil), a.permissions...)
|
|
return c.write(clientMessage{
|
|
Type: "user",
|
|
Kind: a.kind,
|
|
Id: a.id,
|
|
Username: a.username,
|
|
Permissions: perms,
|
|
Data: a.data,
|
|
})
|
|
case joinedAction:
|
|
var status *group.Status
|
|
var data map[string]interface{}
|
|
var g *group.Group
|
|
if a.group != "" {
|
|
g = group.Get(a.group)
|
|
if g != nil {
|
|
s := g.Status(true, "")
|
|
status = &s
|
|
data = g.Data()
|
|
}
|
|
}
|
|
perms := append([]string(nil), c.permissions...)
|
|
err := c.write(clientMessage{
|
|
Type: "joined",
|
|
Kind: a.kind,
|
|
Group: a.group,
|
|
Username: c.username,
|
|
Permissions: perms,
|
|
Status: status,
|
|
Data: data,
|
|
RTCConfiguration: ice.ICEConfiguration(),
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if a.kind == "join" {
|
|
if g == nil {
|
|
log.Println("g is null when joining" +
|
|
"this shouldn't happen")
|
|
return nil
|
|
}
|
|
h := g.GetChatHistory()
|
|
for _, m := range h {
|
|
err := c.write(clientMessage{
|
|
Type: "chathistory",
|
|
Source: m.Id,
|
|
Username: m.User,
|
|
Time: m.Time,
|
|
Value: m.Value,
|
|
Kind: m.Kind,
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
}
|
|
case permissionsChangedAction:
|
|
g := c.Group()
|
|
if g == nil {
|
|
return errors.New("Permissions changed in no group")
|
|
}
|
|
perms := append([]string(nil), c.permissions...)
|
|
status := g.Status(true, "")
|
|
c.write(clientMessage{
|
|
Type: "joined",
|
|
Kind: "change",
|
|
Group: g.Name(),
|
|
Username: c.username,
|
|
Permissions: perms,
|
|
Status: &status,
|
|
RTCConfiguration: ice.ICEConfiguration(),
|
|
})
|
|
if !member("present", c.permissions) {
|
|
up := getUpConns(c)
|
|
for _, u := range up {
|
|
err := delUpConn(
|
|
c, u.id, c.id, true,
|
|
)
|
|
if err == nil {
|
|
failUpConnection(
|
|
c, u.id,
|
|
"permission denied",
|
|
)
|
|
}
|
|
}
|
|
}
|
|
id := c.Id()
|
|
user := c.Username()
|
|
d := c.Data()
|
|
clients := g.GetClients(nil)
|
|
go func(clients []group.Client) {
|
|
for _, cc := range clients {
|
|
cc.PushClient(
|
|
g.Name(), "change", id, user, perms, d,
|
|
)
|
|
}
|
|
}(clients)
|
|
case kickAction:
|
|
return group.KickError{
|
|
a.id, a.username, a.message,
|
|
}
|
|
default:
|
|
log.Printf("unexpected action %T", a)
|
|
return errors.New("unexpected action")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func failUpConnection(c *webClient, id string, message string) error {
|
|
if id != "" {
|
|
err := c.write(clientMessage{
|
|
Type: "abort",
|
|
Id: id,
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
if message != "" {
|
|
err := c.error(group.UserError(message))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func leaveGroup(c *webClient) {
|
|
if c.group == nil {
|
|
return
|
|
}
|
|
|
|
if c.up != nil {
|
|
for id := range c.up {
|
|
delUpConn(c, id, c.id, true)
|
|
}
|
|
}
|
|
if c.down != nil {
|
|
for id := range c.down {
|
|
delDownConn(c, id)
|
|
}
|
|
}
|
|
|
|
group.DelClient(c)
|
|
c.permissions = nil
|
|
c.data = nil
|
|
c.requested = make(map[string][]string)
|
|
c.group = nil
|
|
}
|
|
|
|
func closeDownConn(c *webClient, id string, message string) error {
|
|
err := delDownConn(c, id)
|
|
if err != nil && !os.IsNotExist(err) {
|
|
log.Printf("Close down connection: %v", err)
|
|
}
|
|
err = c.write(clientMessage{
|
|
Type: "close",
|
|
Id: id,
|
|
})
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if message != "" {
|
|
err := c.error(group.UserError(message))
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func setPermissions(g *group.Group, id string, perm string) error {
|
|
client := g.GetClient(id)
|
|
if client == nil {
|
|
return group.UserError("no such user")
|
|
}
|
|
|
|
c, ok := client.(*webClient)
|
|
if !ok {
|
|
return group.UserError("this is not a real user")
|
|
}
|
|
|
|
switch perm {
|
|
case "op":
|
|
c.permissions = addnew("op", c.permissions)
|
|
if g.Description().AllowRecording {
|
|
c.permissions = addnew("record", c.permissions)
|
|
}
|
|
case "unop":
|
|
c.permissions = remove("op", c.permissions)
|
|
c.permissions = remove("record", c.permissions)
|
|
case "present":
|
|
c.permissions = addnew("present", c.permissions)
|
|
case "unpresent":
|
|
c.permissions = remove("present", c.permissions)
|
|
default:
|
|
return group.UserError("unknown permission")
|
|
}
|
|
return c.action(permissionsChangedAction{})
|
|
}
|
|
|
|
func (c *webClient) Kick(id, user, message string) error {
|
|
return c.action(kickAction{id, user, message})
|
|
}
|
|
|
|
func (c *webClient) Joined(group, kind string) error {
|
|
return c.action(joinedAction{group, kind})
|
|
}
|
|
|
|
func kickClient(g *group.Group, id, user, dest string, message string) error {
|
|
client := g.GetClient(dest)
|
|
if client == nil {
|
|
return group.UserError("no such user")
|
|
}
|
|
|
|
return client.Kick(id, user, message)
|
|
}
|
|
|
|
func handleClientMessage(c *webClient, m clientMessage) error {
|
|
if m.Source != "" {
|
|
if m.Source != c.Id() {
|
|
return group.ProtocolError("spoofed client id")
|
|
}
|
|
}
|
|
|
|
if m.Type != "join" {
|
|
if m.Username != "" {
|
|
if m.Username != c.Username() {
|
|
return group.ProtocolError("spoofed username")
|
|
}
|
|
}
|
|
}
|
|
|
|
switch m.Type {
|
|
case "join":
|
|
if m.Kind == "leave" {
|
|
if c.group == nil || c.group.Name() != m.Group {
|
|
return group.ProtocolError("you are not joined")
|
|
}
|
|
leaveGroup(c)
|
|
}
|
|
|
|
if m.Kind != "join" {
|
|
return group.ProtocolError("unknown kind")
|
|
}
|
|
|
|
if c.group != nil {
|
|
return group.ProtocolError(
|
|
"cannot join multiple groups",
|
|
)
|
|
}
|
|
c.username = m.Username
|
|
c.data = m.Data
|
|
g, err := group.AddClient(m.Group, c,
|
|
group.ClientCredentials{
|
|
Username: m.Username,
|
|
Password: m.Password,
|
|
Token: m.Token,
|
|
},
|
|
)
|
|
if err != nil {
|
|
var s string
|
|
if os.IsNotExist(err) {
|
|
s = "group does not exist"
|
|
} else if err == group.ErrNotAuthorised {
|
|
s = "not authorised"
|
|
time.Sleep(200 * time.Millisecond)
|
|
} else if err == group.ErrAnonymousNotAuthorised {
|
|
s = "please choose a username"
|
|
} else if e, ok := err.(group.UserError); ok {
|
|
s = string(e)
|
|
} else {
|
|
s = "internal server error"
|
|
log.Printf("Join group: %v", err)
|
|
}
|
|
return c.write(clientMessage{
|
|
Type: "joined",
|
|
Kind: "fail",
|
|
Group: m.Group,
|
|
Username: c.username,
|
|
Value: s,
|
|
})
|
|
}
|
|
if redirect := g.Description().Redirect; redirect != "" {
|
|
// We normally redirect at the HTTP level, but the group
|
|
// description could have been edited in the meantime.
|
|
return c.write(clientMessage{
|
|
Type: "joined",
|
|
Kind: "redirect",
|
|
Group: m.Group,
|
|
Username: c.username,
|
|
Value: redirect,
|
|
})
|
|
}
|
|
c.group = g
|
|
case "request":
|
|
requested, err := parseRequested(m.Request)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
return c.setRequested(requested)
|
|
case "requestStream":
|
|
down := getDownConn(c, m.Id)
|
|
if down == nil {
|
|
return ErrUnknownId
|
|
}
|
|
requested, err := toStringArray(m.Request)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
c.setRequestedStream(down, requested)
|
|
case "offer":
|
|
if m.Id == "" {
|
|
return errEmptyId
|
|
}
|
|
if !member("present", c.permissions) {
|
|
if m.Replace != "" {
|
|
delUpConn(c, m.Replace, c.id, true)
|
|
}
|
|
c.write(clientMessage{
|
|
Type: "abort",
|
|
Id: m.Id,
|
|
})
|
|
return c.error(group.UserError("not authorised"))
|
|
}
|
|
err := gotOffer(c, m.Id, m.Label, m.SDP, m.Replace)
|
|
if err != nil {
|
|
log.Printf("gotOffer: %v", err)
|
|
return failUpConnection(c, m.Id, err.Error())
|
|
}
|
|
case "answer":
|
|
if m.Id == "" {
|
|
return errEmptyId
|
|
}
|
|
err := gotAnswer(c, m.Id, m.SDP)
|
|
if err != nil {
|
|
log.Printf("gotAnswer: %v", err)
|
|
message := ""
|
|
if err != ErrUnknownId {
|
|
message = err.Error()
|
|
}
|
|
return closeDownConn(c, m.Id, message)
|
|
}
|
|
down := getDownConn(c, m.Id)
|
|
if down == nil {
|
|
return ErrUnknownId
|
|
}
|
|
if down.negotiationNeeded > negotiationUnneeded {
|
|
err := negotiate(
|
|
c, down,
|
|
down.negotiationNeeded == negotiationRestartIce,
|
|
"",
|
|
)
|
|
if err != nil {
|
|
return closeDownConn(c, m.Id, err.Error())
|
|
}
|
|
}
|
|
case "renegotiate":
|
|
if m.Id == "" {
|
|
return errEmptyId
|
|
}
|
|
down := getDownConn(c, m.Id)
|
|
if down != nil {
|
|
err := negotiate(c, down, true, "")
|
|
if err != nil {
|
|
return closeDownConn(c, m.Id, err.Error())
|
|
}
|
|
} else {
|
|
log.Printf("Trying to renegotiate unknown connection")
|
|
}
|
|
case "close":
|
|
if m.Id == "" {
|
|
return errEmptyId
|
|
}
|
|
err := delUpConn(c, m.Id, c.id, true)
|
|
if err != nil {
|
|
log.Printf("Deleting up connection %v: %v",
|
|
m.Id, err)
|
|
return nil
|
|
}
|
|
case "abort":
|
|
if m.Id == "" {
|
|
return errEmptyId
|
|
}
|
|
return closeDownConn(c, m.Id, "")
|
|
case "ice":
|
|
if m.Id == "" {
|
|
return errEmptyId
|
|
}
|
|
if m.Candidate == nil {
|
|
return group.ProtocolError("null candidate")
|
|
}
|
|
err := gotICE(c, m.Candidate, m.Id)
|
|
if err != nil {
|
|
log.Printf("ICE: %v", err)
|
|
}
|
|
case "chat", "usermessage":
|
|
g := c.group
|
|
if g == nil {
|
|
return c.error(group.UserError("join a group first"))
|
|
}
|
|
|
|
tm := group.ToJSTime(time.Now())
|
|
|
|
if m.Type == "chat" {
|
|
if m.Dest == "" {
|
|
g.AddToChatHistory(
|
|
m.Source, m.Username, tm, m.Kind, m.Value,
|
|
)
|
|
}
|
|
}
|
|
mm := clientMessage{
|
|
Type: m.Type,
|
|
Source: m.Source,
|
|
Dest: m.Dest,
|
|
Username: m.Username,
|
|
Privileged: member("op", c.permissions),
|
|
Time: tm,
|
|
Kind: m.Kind,
|
|
NoEcho: m.NoEcho,
|
|
Value: m.Value,
|
|
}
|
|
if m.Dest == "" {
|
|
var except group.Client
|
|
if m.NoEcho {
|
|
except = c
|
|
}
|
|
err := broadcast(g.GetClients(except), mm)
|
|
if err != nil {
|
|
log.Printf("broadcast(chat): %v", err)
|
|
}
|
|
} else {
|
|
cc := g.GetClient(m.Dest)
|
|
if cc == nil {
|
|
return c.error(group.UserError("user unknown"))
|
|
}
|
|
ccc, ok := cc.(*webClient)
|
|
if !ok {
|
|
return c.error(group.UserError(
|
|
"this user doesn't chat",
|
|
))
|
|
}
|
|
ccc.write(mm)
|
|
}
|
|
case "groupaction":
|
|
g := c.group
|
|
if g == nil {
|
|
return c.error(group.UserError("join a group first"))
|
|
}
|
|
switch m.Kind {
|
|
case "clearchat":
|
|
g.ClearChatHistory()
|
|
m := clientMessage{
|
|
Type: "usermessage",
|
|
Kind: "clearchat",
|
|
Privileged: true,
|
|
}
|
|
err := broadcast(g.GetClients(nil), m)
|
|
if err != nil {
|
|
log.Printf("broadcast(clearchat): %v", err)
|
|
}
|
|
case "lock", "unlock":
|
|
if !member("op", c.permissions) {
|
|
return c.error(group.UserError("not authorised"))
|
|
}
|
|
message := ""
|
|
v, ok := m.Value.(string)
|
|
if ok {
|
|
message = v
|
|
}
|
|
g.SetLocked(m.Kind == "lock", message)
|
|
case "record":
|
|
if !member("record", c.permissions) {
|
|
return c.error(group.UserError("not authorised"))
|
|
}
|
|
for _, cc := range g.GetClients(c) {
|
|
_, ok := cc.(*diskwriter.Client)
|
|
if ok {
|
|
return c.error(group.UserError("already recording"))
|
|
}
|
|
}
|
|
disk := diskwriter.New(g)
|
|
_, err := group.AddClient(g.Name(), disk,
|
|
group.ClientCredentials{
|
|
System: true,
|
|
},
|
|
)
|
|
if err != nil {
|
|
disk.Close()
|
|
return c.error(err)
|
|
}
|
|
requestConns(disk, c.group, "")
|
|
case "unrecord":
|
|
if !member("record", c.permissions) {
|
|
return c.error(group.UserError("not authorised"))
|
|
}
|
|
for _, cc := range g.GetClients(c) {
|
|
disk, ok := cc.(*diskwriter.Client)
|
|
if ok {
|
|
disk.Close()
|
|
group.DelClient(disk)
|
|
}
|
|
}
|
|
case "subgroups":
|
|
if !member("op", c.permissions) {
|
|
return c.error(group.UserError("not authorised"))
|
|
}
|
|
s := ""
|
|
for _, sg := range group.GetSubGroups(g.Name()) {
|
|
plural := ""
|
|
if sg.Clients > 1 {
|
|
plural = "s"
|
|
}
|
|
s = s + fmt.Sprintf("%v (%v client%v)\n",
|
|
sg.Name, sg.Clients, plural)
|
|
}
|
|
c.write(clientMessage{
|
|
Type: "chat",
|
|
Dest: c.id,
|
|
Username: "Server",
|
|
Time: group.ToJSTime(time.Now()),
|
|
Value: s,
|
|
})
|
|
case "setdata":
|
|
if !member("op", c.permissions) {
|
|
return c.error(group.UserError("not authorised"))
|
|
}
|
|
data, ok := m.Value.(map[string]interface{})
|
|
if !ok {
|
|
return c.error(group.UserError(
|
|
"Bad value in setdata",
|
|
))
|
|
}
|
|
g.UpdateData(data)
|
|
default:
|
|
return group.ProtocolError("unknown group action")
|
|
}
|
|
case "useraction":
|
|
g := c.group
|
|
if g == nil {
|
|
return c.error(group.UserError("join a group first"))
|
|
}
|
|
switch m.Kind {
|
|
case "op", "unop", "present", "unpresent":
|
|
if !member("op", c.permissions) {
|
|
return c.error(group.UserError("not authorised"))
|
|
}
|
|
err := setPermissions(g, m.Dest, m.Kind)
|
|
if err != nil {
|
|
return c.error(err)
|
|
}
|
|
case "kick":
|
|
if !member("op", c.permissions) {
|
|
return c.error(group.UserError("not authorised"))
|
|
}
|
|
message := ""
|
|
v, ok := m.Value.(string)
|
|
if ok {
|
|
message = v
|
|
}
|
|
err := kickClient(g, m.Source, m.Username, m.Dest, message)
|
|
if err != nil {
|
|
return c.error(err)
|
|
}
|
|
case "setdata":
|
|
if m.Dest != c.Id() {
|
|
return c.error(group.UserError("not authorised"))
|
|
}
|
|
data, ok := m.Value.(map[string]interface{})
|
|
if !ok {
|
|
return c.error(group.UserError(
|
|
"Bad value in setdata",
|
|
))
|
|
}
|
|
if c.data == nil {
|
|
c.data = make(map[string]interface{})
|
|
}
|
|
for k, v := range data {
|
|
if v == nil {
|
|
delete(c.data, k)
|
|
} else {
|
|
c.data[k] = v
|
|
}
|
|
}
|
|
id := c.Id()
|
|
user := c.Username()
|
|
perms := c.Permissions()
|
|
data = c.Data()
|
|
go func(clients []group.Client) {
|
|
for _, cc := range clients {
|
|
cc.PushClient(
|
|
g.Name(), "change",
|
|
id, user, perms, data,
|
|
)
|
|
}
|
|
}(g.GetClients(nil))
|
|
default:
|
|
return group.ProtocolError("unknown user action")
|
|
}
|
|
case "pong":
|
|
// nothing
|
|
case "ping":
|
|
return c.write(clientMessage{
|
|
Type: "pong",
|
|
})
|
|
default:
|
|
log.Printf("unexpected message: %v", m.Type)
|
|
return group.ProtocolError("unexpected message")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func clientReader(conn *websocket.Conn, read chan<- interface{}, done <-chan struct{}) {
|
|
defer close(read)
|
|
for {
|
|
var m clientMessage
|
|
err := conn.ReadJSON(&m)
|
|
if err != nil {
|
|
select {
|
|
case read <- err:
|
|
return
|
|
case <-done:
|
|
return
|
|
}
|
|
}
|
|
select {
|
|
case read <- m:
|
|
case <-done:
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
func clientWriter(conn *websocket.Conn, ch <-chan interface{}, done chan<- struct{}) {
|
|
defer func() {
|
|
close(done)
|
|
conn.Close()
|
|
}()
|
|
|
|
for {
|
|
m, ok := <-ch
|
|
if !ok {
|
|
break
|
|
}
|
|
err := conn.SetWriteDeadline(
|
|
time.Now().Add(500 * time.Millisecond),
|
|
)
|
|
if err != nil {
|
|
return
|
|
}
|
|
switch m := m.(type) {
|
|
case clientMessage:
|
|
err := conn.WriteJSON(m)
|
|
if err != nil {
|
|
return
|
|
}
|
|
case []byte:
|
|
err := conn.WriteMessage(websocket.TextMessage, m)
|
|
if err != nil {
|
|
return
|
|
}
|
|
case closeMessage:
|
|
if m.data != nil {
|
|
conn.WriteMessage(
|
|
websocket.CloseMessage,
|
|
m.data,
|
|
)
|
|
}
|
|
return
|
|
default:
|
|
log.Printf("clientWiter: unexpected message %T", m)
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
func (c *webClient) Warn(oponly bool, message string) error {
|
|
if oponly && !member("op", c.permissions) {
|
|
return nil
|
|
}
|
|
|
|
return c.write(clientMessage{
|
|
Type: "usermessage",
|
|
Kind: "warning",
|
|
Dest: c.id,
|
|
Privileged: true,
|
|
Value: message,
|
|
})
|
|
}
|
|
|
|
var ErrClientDead = errors.New("client is dead")
|
|
|
|
func (c *webClient) action(a interface{}) error {
|
|
c.actionMu.Lock()
|
|
empty := len(c.actions) == 0
|
|
c.actions = append(c.actions, a)
|
|
c.actionMu.Unlock()
|
|
|
|
if empty {
|
|
select {
|
|
case c.actionCh <- struct{}{}:
|
|
return nil
|
|
case <-c.done:
|
|
return ErrClientDead
|
|
default:
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (c *webClient) write(m clientMessage) error {
|
|
select {
|
|
case c.writeCh <- m:
|
|
return nil
|
|
case <-c.writerDone:
|
|
return ErrClientDead
|
|
}
|
|
}
|
|
|
|
func broadcast(cs []group.Client, m clientMessage) error {
|
|
b, err := json.Marshal(m)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
for _, c := range cs {
|
|
cc, ok := c.(*webClient)
|
|
if !ok {
|
|
continue
|
|
}
|
|
select {
|
|
case cc.writeCh <- b:
|
|
case <-cc.writerDone:
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (c *webClient) close(data []byte) error {
|
|
select {
|
|
case c.writeCh <- closeMessage{data}:
|
|
return nil
|
|
case <-c.writerDone:
|
|
return ErrClientDead
|
|
}
|
|
}
|
|
|
|
func errorMessage(id string, err error) *clientMessage {
|
|
switch e := err.(type) {
|
|
case group.UserError:
|
|
return &clientMessage{
|
|
Type: "usermessage",
|
|
Kind: "error",
|
|
Dest: id,
|
|
Privileged: true,
|
|
Value: e.Error(),
|
|
}
|
|
case group.KickError:
|
|
message := e.Message
|
|
if message == "" {
|
|
message = "you have been kicked out"
|
|
}
|
|
return &clientMessage{
|
|
Type: "usermessage",
|
|
Kind: "kicked",
|
|
Id: e.Id,
|
|
Username: e.Username,
|
|
Dest: id,
|
|
Privileged: true,
|
|
Value: message,
|
|
}
|
|
default:
|
|
return nil
|
|
}
|
|
}
|
|
|
|
func (c *webClient) error(err error) error {
|
|
m := errorMessage(c.id, err)
|
|
if m == nil {
|
|
return err
|
|
}
|
|
return c.write(*m)
|
|
}
|