Skip to content
This repository was archived by the owner on Sep 17, 2019. It is now read-only.

Commit fe25d90

Browse files
senseabsenseab
committed
websocket done; next is policies
1 parent 9c44422 commit fe25d90

2 files changed

Lines changed: 34 additions & 15 deletions

File tree

‎controllers/client.go‎

Lines changed: 22 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -76,6 +76,12 @@ func (c *ClientController) WebSocket() {
7676
c.ServeJSON()
7777
return
7878
}
79+
ticker := time.NewTicker(5 * time.Second)
80+
81+
ws.SetReadDeadline(time.Now().Add(10 * time.Second))
82+
ws.SetPongHandler(func(string) error {
83+
ws.SetReadDeadline(time.Now().Add(10 * time.Second))
84+
})
7985
var c chan models.Signal
8086
c, ok := models.SignalChannels[HostsId]
8187
if !ok {
@@ -87,28 +93,29 @@ func (c *ClientController) WebSocket() {
8793
select {
8894
case s := <-models.SignalChannels[HostsId]:
8995
ws.WriteJSON(s)
90-
_, bConfirm, _ := ws.ReadMessage()
96+
_, bConfirm, err := ws.ReadMessage()
97+
if websocket.IsCloseError(err,
98+
websocket.CloseGoingAway) {
99+
beego.Info("Host", name, "is offline.")
100+
break
101+
} else if err != nil {
102+
beego.Warn("Error on reading:", err.Error())
103+
break
104+
}
91105
if string(bConfirm) == ClientWebSocketReplyDone {
92106
models.DeleteSignal(HostId, s["id"])
93107
}
94-
case <-time.After(5 * time.Second):
95-
ws.WriteControl(websocket.PingMessage,
96-
[]byte("Are you alive?"),
97-
time.Now().Add(10*time.Second))
98-
mType, _, err := ws.ReadMessage()
99-
if err != nil || mType != websocket.PongMessage {
100-
beego.Info("Host", name, "become offline.")
101-
return
102-
}
103-
default:
104-
mType, _, _ := ws.ReadMessage()
105-
if mType == websocket.CloseMessage {
106-
beego.Info("Host", name, "become offline.")
108+
case <-ticker.C:
109+
err := ws.WriteMessage(websocket.PingMessage, []byte{})
110+
if err != nil {
107111
return
108112
}
109113
}
110114
}
111-
defer ws.Close()
115+
defer func() {
116+
ticker.Stop()
117+
ws.Close()
118+
}()
112119
}
113120
}
114121

‎models/signals.go‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import (
55
"fmt"
66
"moduleab_server/common"
77
"moduleab_server/models"
8+
"sync"
89
"time"
910

1011
"github.com/pborman/uuid"
@@ -26,6 +27,7 @@ var (
2627

2728
var (
2829
SignalChannels map[string]chan Signal
30+
lock *sync.Mutex
2931
)
3032

3133
func init() {
@@ -35,6 +37,8 @@ func init() {
3537
type Signal map[string]interface{}
3638

3739
func AddSignal(hostId string, signal Signal) (string, error) {
40+
lock.Lock()
41+
defer lock.Unlock()
3842
keyName := fmt.Sprintf("%s%s", common.DefaultRedisKey, hostId)
3943
newId := uuid.New()
4044
signal["id"] = newId
@@ -56,6 +60,8 @@ func AddSignal(hostId string, signal Signal) (string, error) {
5660
}
5761

5862
func GetSignals(hostId string) []Signal {
63+
lock.Lock()
64+
defer lock.Unlock()
5965
keyName := fmt.Sprintf("%s%s", common.DefaultRedisKey, hostId)
6066
v := common.DefaultRedisClient.Get(keyName)
6167
n, ok := v.([]Signal)
@@ -66,6 +72,8 @@ func GetSignals(hostId string) []Signal {
6672
}
6773

6874
func GetSignal(hostId, id string) (Signal, error) {
75+
lock.Lock()
76+
defer lock.Unlock()
6977
signals := models.GetSignals(hostId)
7078
for _, v := range signals {
7179
if v["id"] == id {
@@ -76,11 +84,15 @@ func GetSignal(hostId, id string) (Signal, error) {
7684
}
7785

7886
func TruncateSignals(hostId string) {
87+
lock.Lock()
88+
defer lock.Unlock()
7989
keyName := fmt.Sprintf("%s%s", common.DefaultRedisKey, hostId)
8090
common.DefaultRedisClient.Delete(keyName)
8191
}
8292

8393
func DeleteSignal(hostId string, signalId string) error {
94+
lock.Lock()
95+
defer lock.Unlock()
8496
keyName := fmt.Sprintf("%s%s", common.DefaultRedisKey, hostId)
8597
if common.DefaultRedisClient.IsExist(keyName) {
8698
v := common.DefaultRedisClient.Get(keyName)

0 commit comments

Comments
 (0)