forked from topfreegames/pitaya
-
Notifications
You must be signed in to change notification settings - Fork 0
/
room.go
177 lines (156 loc) · 4.56 KB
/
room.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
package services
import (
"context"
"encoding/gob"
"fmt"
"strconv"
"time"
"github.com/topfreegames/pitaya"
"github.com/topfreegames/pitaya/component"
"github.com/topfreegames/pitaya/config"
"github.com/topfreegames/pitaya/examples/demo/protos"
"github.com/topfreegames/pitaya/groups"
"github.com/topfreegames/pitaya/timer"
)
type (
// Room represents a component that contains a bundle of room related handler
// like Join/Message
Room struct {
component.Base
timer *timer.Timer
Stats *Stats
}
// UserMessage represents a message that user sent
UserMessage struct {
Name string `json:"name"`
Content string `json:"content"`
}
// Stats exports the room status
Stats struct {
outboundBytes int
inboundBytes int
}
// SendRPCMsg represents a rpc message
SendRPCMsg struct {
ServerID string `json:"serverId"`
Route string `json:"route"`
Msg string `json:"msg"`
}
// NewUser message will be received when new user join room
NewUser struct {
Content string `json:"content"`
}
// AllMembers contains all members uid
AllMembers struct {
Members []string `json:"members"`
}
// JoinResponse represents the result of joining room
JoinResponse struct {
Code int `json:"code"`
Result string `json:"result"`
}
)
// Outbound gets the outbound status
func (Stats *Stats) Outbound(ctx context.Context, in []byte) ([]byte, error) {
Stats.outboundBytes += len(in)
return in, nil
}
// Inbound gets the inbound status
func (Stats *Stats) Inbound(ctx context.Context, in []byte) ([]byte, error) {
Stats.inboundBytes += len(in)
return in, nil
}
// NewRoom returns a new room
func NewRoom() *Room {
return &Room{
Stats: &Stats{},
}
}
// Init runs on service initialization
func (r *Room) Init() {
gsi := groups.NewMemoryGroupService(config.NewConfig())
pitaya.InitGroups(gsi)
pitaya.GroupCreate(context.Background(), "room")
// It is necessary to register all structs that will be used in RPC calls
// This must be done both in the caller and callee servers
gob.Register(&UserMessage{})
}
// AfterInit component lifetime callback
func (r *Room) AfterInit() {
r.timer = pitaya.NewTimer(time.Minute, func() {
count, err := pitaya.GroupCountMembers(context.Background(), "room")
println("UserCount: Time=>", time.Now().String(), "Count=>", count, "Error=>", err)
println("OutboundBytes", r.Stats.outboundBytes)
println("InboundBytes", r.Stats.outboundBytes)
})
}
// Entry is the entrypoint
func (r *Room) Entry(ctx context.Context, msg []byte) (*JoinResponse, error) {
s := pitaya.GetSessionFromCtx(ctx)
err := s.Bind(ctx, strconv.Itoa(int(s.ID())))
if err != nil {
return nil, pitaya.Error(err, "RH-000", map[string]string{"failed": "bind"})
}
return &JoinResponse{Result: "ok"}, nil
}
// GetSessionData gets the session data
func (r *Room) GetSessionData(ctx context.Context) (*SessionData, error) {
s := pitaya.GetSessionFromCtx(ctx)
return &SessionData{
Data: s.GetData(),
}, nil
}
// SetSessionData sets the session data
func (r *Room) SetSessionData(ctx context.Context, data *SessionData) ([]byte, error) {
s := pitaya.GetSessionFromCtx(ctx)
err := s.SetData(data.Data)
if err != nil {
return nil, err
}
err = s.PushToFront(ctx)
if err != nil {
return nil, err
}
return []byte("success"), nil
}
// Join room
func (r *Room) Join(ctx context.Context) (*JoinResponse, error) {
s := pitaya.GetSessionFromCtx(ctx)
err := pitaya.GroupAddMember(ctx, "room", s.UID())
if err != nil {
return nil, err
}
members, err := pitaya.GroupMembers(ctx, "room")
if err != nil {
return nil, err
}
s.Push("onMembers", &AllMembers{Members: members})
pitaya.GroupBroadcast(ctx, "connector", "room", "onNewUser", &NewUser{Content: fmt.Sprintf("New user: %d", s.ID())})
err = s.OnClose(func() {
pitaya.GroupRemoveMember(ctx, "room", s.UID())
})
if err != nil {
return nil, err
}
return &JoinResponse{Result: "success"}, nil
}
// Message sync last message to all members
func (r *Room) Message(ctx context.Context, msg *UserMessage) {
err := pitaya.GroupBroadcast(ctx, "connector", "room", "onMessage", msg)
if err != nil {
fmt.Println("error broadcasting message", err)
}
}
// SendRPC sends rpc
func (r *Room) SendRPC(ctx context.Context, msg *protos.RPCMsg) (*protos.RPCRes, error) {
ret := &protos.RPCRes{}
err := pitaya.RPC(ctx, "connector.connectorremote.remotefunc", ret, msg)
if err != nil {
return nil, pitaya.Error(err, "RPC-000")
}
return ret, nil
}
// MessageRemote just echoes the given message
func (r *Room) MessageRemote(ctx context.Context, msg *UserMessage, b bool, s string) (*UserMessage, error) {
return msg, nil
}