diff --git a/conn.go b/conn.go index 13e4e53..4155529 100644 --- a/conn.go +++ b/conn.go @@ -34,7 +34,6 @@ type upTrack interface { } type downConnection interface { - Close() error } type downTrack interface { diff --git a/disk.go b/disk.go index 8a31ce6..f1c1b00 100644 --- a/disk.go +++ b/disk.go @@ -20,7 +20,7 @@ type diskClient struct { id string mu sync.Mutex - down []*diskConn + down map[string]*diskConn closed bool } @@ -52,7 +52,7 @@ func (client *diskClient) Close() error { return nil } -func (client *diskClient) pushConn(conn upConnection, tracks []upTrack, label string) error { +func (client *diskClient) pushConn(id string, conn upConnection, tracks []upTrack, label string) error { client.mu.Lock() defer client.mu.Unlock() @@ -60,18 +60,32 @@ func (client *diskClient) pushConn(conn upConnection, tracks []upTrack, label st return errors.New("disk client is closed") } + old := client.down[id] + if old != nil { + old.Close() + delete(client.down, id) + } + + if conn == nil { + return nil + } + directory := filepath.Join(recordingsDir, client.group.name) err := os.MkdirAll(directory, 0700) if err != nil { return err } + if client.down == nil { + client.down = make(map[string]*diskConn) + } + down, err := newDiskConn(directory, label, conn, tracks) if err != nil { return err } - client.down = append(client.down, down) + client.down[conn.Id()] = down return nil } diff --git a/group.go b/group.go index 728f2ac..12e991e 100644 --- a/group.go +++ b/group.go @@ -25,7 +25,7 @@ type client interface { Group() *group Id() string Username() string - pushConn(conn upConnection, tracks []upTrack, label string) error + pushConn(id string, conn upConnection, tracks []upTrack, label string) error pushClient(id, username string, add bool) error } @@ -53,11 +53,8 @@ type group struct { history []chatHistoryEntry } -type delConnAction struct { - id string -} - -type addConnAction struct { +type pushConnAction struct { + id string conn upConnection tracks []upTrack } diff --git a/rtpconn.go b/rtpconn.go index 04524f6..78e96b0 100644 --- a/rtpconn.go +++ b/rtpconn.go @@ -111,7 +111,6 @@ type rtpDownConnection struct { remote upConnection tracks []*rtpDownTrack iceCandidates []*webrtc.ICECandidateInit - close func() error } func newDownConn(id string, remote upConnection) (*rtpDownConnection, error) { @@ -133,10 +132,6 @@ func newDownConn(id string, remote upConnection) (*rtpDownConnection, error) { return conn, nil } -func (down *rtpDownConnection) Close() error { - return down.close() -} - func (down *rtpDownConnection) addICECandidate(candidate *webrtc.ICECandidateInit) error { if down.pc.RemoteDescription() != nil { return down.pc.AddICECandidate(*candidate) @@ -279,7 +274,6 @@ type rtpUpConnection struct { iceCandidates []*webrtc.ICECandidateInit mu sync.Mutex - closed bool tracks []*rtpUpTrack local []downConnection } @@ -303,9 +297,6 @@ func (up *rtpUpConnection) Label() string { func (up *rtpUpConnection) addLocal(local downConnection) error { up.mu.Lock() defer up.mu.Unlock() - if up.closed { - return ErrConnectionClosed - } for _, t := range up.local { if t == local { return nil @@ -335,21 +326,6 @@ func (up *rtpUpConnection) getLocal() []downConnection { return local } -func (up *rtpUpConnection) Close() error { - up.mu.Lock() - defer up.mu.Unlock() - - go func(local []downConnection) { - for _, l := range local { - l.Close() - } - }(up.local) - - up.local = nil - up.closed = true - return up.pc.Close() -} - func (up *rtpUpConnection) addICECandidate(candidate *webrtc.ICECandidateInit) error { if up.pc.RemoteDescription() != nil { return up.pc.AddICECandidate(*candidate) @@ -468,7 +444,7 @@ func newUpConn(c client, id string) (*rtpUpConnection, error) { } clients := c.Group().getClients(c) for _, cc := range clients { - cc.pushConn(conn, tracks, conn.label) + cc.pushConn(conn.id, conn, tracks, conn.label) } go rtcpUpSender(conn) } diff --git a/webclient.go b/webclient.go index dde26a6..30d2c24 100644 --- a/webclient.go +++ b/webclient.go @@ -261,7 +261,14 @@ func delUpConn(c *webClient, id string) bool { } conn.mu.Unlock() - conn.Close() + go func(clients []client) { + for _, c := range clients { + c.pushConn(conn.id, nil, nil, "") + } + }(c.Group().getClients(c)) + + conn.pc.Close() + return true } @@ -320,10 +327,6 @@ func addDownConn(c *webClient, id string, remote upConnection) (*rtpDownConnecti } c.down[id] = conn - conn.close = func() error { - return c.action(delConnAction{conn.id}) - } - return conn, nil } @@ -571,12 +574,12 @@ func addDownConnTracks(c *webClient, remote upConnection, tracks []upTrack) (*rt return down, nil } -func (c *webClient) pushConn(conn upConnection, tracks []upTrack, label string) error { - err := c.action(addConnAction{conn, tracks}) +func (c *webClient) pushConn(id string, conn upConnection, tracks []upTrack, label string) error { + err := c.action(pushConnAction{id, conn, tracks}) if err != nil { return err } - if label != "" { + if conn != nil && label != "" { err := c.action(addLabelAction{conn.Id(), conn.Label()}) if err != nil { return err @@ -714,7 +717,17 @@ func clientLoop(c *webClient, conn *websocket.Conn) error { } case a := <-c.actionCh: switch a := a.(type) { - case addConnAction: + case pushConnAction: + found := delDownConn(c, a.id) + if a.conn == nil { + if found { + c.write(clientMessage{ + Type: "close", + Id: a.id, + }) + } + continue + } down, err := addDownConnTracks( c, a.conn, a.tracks, ) @@ -736,14 +749,6 @@ func clientLoop(c *webClient, conn *websocket.Conn) error { continue } } - case delConnAction: - found := delDownConn(c, a.id) - if found { - c.write(clientMessage{ - Type: "close", - Id: a.id, - }) - } case addLabelAction: c.write(clientMessage{ Type: "label", @@ -757,7 +762,7 @@ func clientLoop(c *webClient, conn *websocket.Conn) error { for i, t := range tracks { ts[i] = t } - go a.c.pushConn(u, ts, u.label) + go a.c.pushConn(u.id, u, ts, u.label) } case connectionFailedAction: found := delUpConn(c, a.id)