juice/cliManager.go at master · miyabyte/juice · GitHub

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
package juice

import (
	"context"
	"github.com/google/uuid"
	"github.com/gorilla/websocket"
	"sync"
	"time"
)

var cm *cliManager
var onceCM sync.Once

type Observer interface {
	Online(cli *Client)
	Offline(cli *Client)
}

type cliManager struct {
	clients map[uint32]*Client
	Event
}

func GetCliManager() *cliManager {
	return cm
}

func NewCliManager(e Event) *cliManager {
	onceCM.Do(func() {
		cm = &cliManager{
			clients: make(map[uint32]*Client),
			Event:   e,
		}
	})
	return cm
}

func NewClient(conn *websocket.Conn) (cli *Client, err error) {
	if _UUID, err := uuid.NewUUID(); err != nil {
		return nil, err
	} else {
		ctx, cancel := context.WithCancel(context.Background())
		return &Client{
			conn:     conn,
			UUID:     _UUID.ID(),
			LastTime: time.Now(),
			Ctx:      ctx,
			Cancel:   cancel,
		}, nil
	}
}

func (cm *cliManager) GetClients() map[uint32]*Client {
	return cm.clients
}

func (cm *cliManager) GetClient(uuid uint32) (cli *Client, ok bool) {
	cli, ok = cm.clients[uuid]
	return
}

// up  observer mode
func (cm *cliManager) AddClient(cli *Client) *cliManager {
	if GetConfig().EnableAnalyzeUid {
		ucm := GetUserCliManager()
		//lock
		ucm.Lock()
		defer ucm.Unlock()

		ucm.AddClient(cli)
	}
	cm.clients[cli.UUID] = cli
	return cm
}

func (cm *cliManager) CloseClient(c *Client) {
	_ = c.conn.Close()
	cm.RemoveClient(c)
}

// down observer mode
func (cm *cliManager) RemoveClient(cli *Client) {
	defer cli.Cancel()
	if GetConfig().EnableAnalyzeUid {
		ucm := GetUserCliManager()
		//lock
		ucm.Lock()
		defer ucm.Unlock()

		ucm.RemoveClient(cli)
	}
	delete(cm.clients, cli.UUID)
	cm.Close(cli)
}

func (cm *cliManager) getMessage(cli *Client) {
	for {
		// msgType 1 text 2 binary
		messageType, p, err := cli.conn.ReadMessage()
		if err != nil {
			cm.ErrorHandler(NewJError(ErrWsGetMsg, err.Error()))
			return
		}

		// heartbeat  close
		cm.baseHandle(cli, messageType, p)

		if messageType == websocket.TextMessage {
			cm.Event.Message(cli, p)
		}

		if messageType == websocket.BinaryMessage {
			cm.Event.BinaryMessage(cli, p)
		}

		// wm = nextWriter\write\close
		//if err := conn.WriteMessage(messageType, p); err != nil {
		//	j.Cmd(err)
		//	return
	}
}

func (cm *cliManager) baseHandle(cli *Client, messageType int, p []byte) {

	//heartbeat
	if messageType == websocket.PingMessage ||
		messageType == websocket.PongMessage ||
		messageType == websocket.TextMessage ||
		messageType == websocket.BinaryMessage {
		cli.LastTime = time.Now()
	}

	//客户端主动关闭
	if messageType == websocket.CloseMessage {
		cm.CloseClient(cli)
	}

}