Skip to content

Commit

Permalink
Feat: room can join multiple client
Browse files Browse the repository at this point in the history
  • Loading branch information
@zijiren233
zijiren233 committed Nov 30, 2023
1 parent 56b599d commit 8147854
Show file tree
Hide file tree
Showing 10 changed files with 208 additions and 889 deletions.
44 changes: 0 additions & 44 deletions internal/op/current.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,9 +3,6 @@ package op
import (
"sync"
"time"

"github.com/synctv-org/synctv/internal/model"
pb "github.com/synctv-org/synctv/proto/message"
)

type current struct {
Expand Down Expand Up @@ -89,47 +86,6 @@ func (c *current) SetSeekRate(seek, rate, timeDiff float64) Status {
return c.current.SetSeekRate(seek, rate, timeDiff)
}

func (c *Current) Proto() *pb.Current {
current := &pb.Current{
Status: &pb.Status{
Seek: c.Status.Seek,
Rate: c.Status.Rate,
Playing: c.Status.Playing,
},
}
current.Movie = &pb.MovieInfo{
Id: c.Movie.Movie.ID,
Base: &pb.BaseMovieInfo{
Url: c.Movie.Movie.Base.Url,
Name: c.Movie.Movie.Base.Name,
Live: c.Movie.Movie.Base.Live,
Proxy: c.Movie.Movie.Base.Proxy,
RtmpSource: c.Movie.Movie.Base.RtmpSource,
Type: c.Movie.Movie.Base.Type,
Headers: c.Movie.Movie.Base.Headers,
},
CreatedAt: c.Movie.Movie.CreatedAt.UnixMilli(),
Creator: GetUserName(c.Movie.Movie.CreatorID),
}
if c.Movie.Movie.Base.VendorInfo.Vendor != "" {
current.Movie.Base.VendorInfo = &pb.VendorInfo{
Vendor: string(c.Movie.Movie.Base.VendorInfo.Vendor),
Shared: c.Movie.Movie.Base.VendorInfo.Shared,
}
switch c.Movie.Movie.Base.VendorInfo.Vendor {
case model.StreamingVendorBilibili:
current.Movie.Base.VendorInfo.Bilibili = &pb.BilibiliVendorInfo{
Bvid: c.Movie.Movie.Base.VendorInfo.Bilibili.Bvid,
Cid: c.Movie.Movie.Base.VendorInfo.Bilibili.Cid,
Epid: c.Movie.Movie.Base.VendorInfo.Bilibili.Epid,
Quality: c.Movie.Movie.Base.VendorInfo.Bilibili.Quality,
VendorName: c.Movie.Movie.Base.VendorInfo.Bilibili.VendorName,
}
}
}
return current
}

func (c *Current) UpdateSeek() {
if c.Movie.Movie.Base.Live {
c.Status.lastUpdate = time.Now()
Expand Down
116 changes: 71 additions & 45 deletions internal/op/hub.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@ import (

type Hub struct {
id string
clients rwmap.RWMap[string, *Client]
clients rwmap.RWMap[string, *rwmap.RWMap[*Client, struct{}]]
broadcast chan *broadcastMessage
exit chan struct{}
closed uint32
Expand All @@ -26,29 +26,22 @@ type Hub struct {
}

type broadcastMessage struct {
data Message
sender string
sendToSelf bool
ignoreId []string
data Message
ignoreClient []*Client
ignoreId []string
}

type BroadcastConf func(*broadcastMessage)

func WithSender(sender string) BroadcastConf {
func WithIgnoreClient(cli ...*Client) BroadcastConf {
return func(bm *broadcastMessage) {
bm.sender = sender
}
}

func WithSendToSelf() BroadcastConf {
return func(bm *broadcastMessage) {
bm.sendToSelf = true
bm.ignoreClient = cli
}
}

func WithIgnoreId(id ...string) BroadcastConf {
return func(bm *broadcastMessage) {
bm.ignoreId = append(bm.ignoreId, id...)
bm.ignoreId = id
}
}

Expand All @@ -73,19 +66,21 @@ func (h *Hub) serve() error {
select {
case message := <-h.broadcast:
h.devMessage(message.data)
h.clients.Range(func(_ string, cli *Client) bool {
if !message.sendToSelf {
if cli.u.Username == message.sender {
h.clients.Range(func(id string, cli *rwmap.RWMap[*Client, struct{}]) bool {
cli.Range(func(c *Client, value struct{}) bool {
if utils.In(message.ignoreId, c.u.ID) {
return true
}
}
if utils.In(message.ignoreId, cli.u.Username) {
if utils.In(message.ignoreClient, c) {
return true
}
if err := c.Send(message.data); err != nil {
log.Debugf("hub: %s, write to client err: %s\nmessage: %+v", h.id, err, message)
c.Close()
}
return true
}
if err := cli.Send(message.data); err != nil {
log.Debugf("hub: %s, write to client err: %s\nmessage: %+v", h.id, err, message)
cli.Close()
}
})

return true
})
case <-h.exit:
Expand All @@ -98,17 +93,18 @@ func (h *Hub) serve() error {
func (h *Hub) ping() {
ticker := time.NewTicker(time.Second * 5)
defer ticker.Stop()
var pre int64 = 0
var (
pre int64 = 0
current int64
)
for {
select {
case <-ticker.C:
current := h.ClientNum()
current = h.PeopleNum()
if current != pre {
if err := h.Broadcast(&ElementMessage{
ElementMessage: &pb.ElementMessage{
Type: pb.ElementMessageType_CHANGE_PEOPLE,
PeopleNum: current,
},
Type: pb.ElementMessageType_CHANGE_PEOPLE,
PeopleNum: current,
}); err != nil {
continue
}
Expand Down Expand Up @@ -144,9 +140,13 @@ func (h *Hub) Close() error {
return ErrAlreadyClosed
}
close(h.exit)
h.clients.Range(func(_ string, client *Client) bool {
h.clients.Delete(client.u.ID)
client.Close()
h.clients.Range(func(id string, client *rwmap.RWMap[*Client, struct{}]) bool {
h.clients.Delete(id)
client.Range(func(key *Client, value struct{}) bool {
client.Delete(key)
key.Close()
return true
})
return true
})
h.wg.Wait()
Expand All @@ -173,46 +173,72 @@ func (h *Hub) Broadcast(data Message, conf ...BroadcastConf) error {
}
}

func (h *Hub) RegClient(cli *Client) (*Client, error) {
func (h *Hub) RegClient(cli *Client) error {
if h.Closed() {
return nil, ErrAlreadyClosed
return ErrAlreadyClosed
}
err := h.Start()
if err != nil {
return nil, err
return err
}
c, loaded := h.clients.LoadOrStore(cli.u.ID, cli)
c, _ := h.clients.LoadOrStore(cli.u.ID, &rwmap.RWMap[*Client, struct{}]{})
_, loaded := c.LoadOrStore(cli, struct{}{})
if loaded {
return nil, errors.New("client already registered")
return errors.New("client already exist")
}
return c, nil
return nil
}

func (h *Hub) UnRegClient(user *User) error {
func (h *Hub) UnRegClient(cli *Client) error {
if h.Closed() {
return ErrAlreadyClosed
}
if user == nil {
if cli == nil {
return errors.New("user is nil")
}
_, loaded := h.clients.LoadAndDelete(user.ID)
c, loaded := h.clients.Load(cli.u.ID)
if !loaded {
return errors.New("client not found")
}
_, loaded2 := c.LoadAndDelete(cli)
if !loaded2 {
return errors.New("client not found")
}
if c.Len() == 0 {
if h.clients.CompareAndDelete(cli.u.ID, c) {
c.Range(func(key *Client, value struct{}) bool {
c.Delete(key)
h.RegClient(key)
return true
})
}
}
return nil
}

func (h *Hub) ClientNum() int64 {
func (h *Hub) PeopleNum() int64 {
return h.clients.Len()
}

func (h *Hub) SendToUser(userID string, data Message) error {
func (h *Hub) SendToUser(userID string, data Message) (err error) {
if h.Closed() {
return ErrAlreadyClosed
}
cli, ok := h.clients.Load(userID)
if !ok {
return nil
}
return cli.Send(data)
cli.Range(func(key *Client, value struct{}) bool {
if err = key.Send(data); err != nil {
cli.CompareAndDelete(key, value)
log.Debugf("hub: %s, write to client err: %s\nmessage: %+v", h.id, err, data)
key.Close()
}
return true
})
return
}

func (h *Hub) LoadClient(userID string) (*rwmap.RWMap[*Client, struct{}], bool) {
return h.clients.Load(userID)
}
47 changes: 3 additions & 44 deletions internal/op/message.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,6 @@ package op
import (
"io"

json "github.com/json-iterator/go"

"github.com/gorilla/websocket"
pb "github.com/synctv-org/synctv/proto/message"
"google.golang.org/protobuf/proto"
Expand All @@ -14,62 +12,27 @@ type Message interface {
MessageType() int
String() string
Encode(w io.Writer) error
BeforeSend(sendTo *User) error
}

type ElementJsonMessage struct {
BeforeSendFunc func(sendTo *User) error
*pb.ElementMessage
}

func (em *ElementJsonMessage) MessageType() int {
return websocket.TextMessage
}

func (em *ElementJsonMessage) String() string {
return em.ElementMessage.String()
}

func (em *ElementJsonMessage) Encode(w io.Writer) error {
return json.NewEncoder(w).Encode(em)
}

func (em *ElementJsonMessage) BeforeSend(sendTo *User) error {
if em.BeforeSendFunc != nil {
return em.BeforeSendFunc(sendTo)
}
return nil
}

type ElementMessage struct {
BeforeSendFunc func(sendTo *User) error
*pb.ElementMessage
}
type ElementMessage pb.ElementMessage

func (em *ElementMessage) MessageType() int {
return websocket.BinaryMessage
}

func (em *ElementMessage) String() string {
return em.ElementMessage.String()
return (*pb.ElementMessage)(em).String()
}

func (em *ElementMessage) Encode(w io.Writer) error {
b, err := proto.Marshal(em)
b, err := proto.Marshal((*pb.ElementMessage)(em))
if err != nil {
return err
}
_, err = w.Write(b)
return err
}

func (em *ElementMessage) BeforeSend(sendTo *User) error {
if em.BeforeSendFunc != nil {
return em.BeforeSendFunc(sendTo)
}
return nil
}

type PingMessage struct{}

func (pm *PingMessage) MessageType() int {
Expand All @@ -83,7 +46,3 @@ func (pm *PingMessage) String() string {
func (pm *PingMessage) Encode(w io.Writer) error {
return nil
}

func (pm *PingMessage) BeforeSend(sendTo *User) error {
return nil
}
Loading

0 comments on commit 8147854

Please sign in to comment.