commit fec669ea34ecf533be753c9fc47ecac3ac6491ff Author: poiuty Date: Wed Jul 27 11:28:07 2022 +0300 update diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..445e913 --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +.DS_Store +/app/chaturbate/chaturbate +/app/bongacams/bongacams diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..8531c11 --- /dev/null +++ b/LICENSE @@ -0,0 +1,21 @@ +MIT License + +Copyright (c) 2020 poiuty + +Permission is hereby granted, free of charge, to any person obtaining a copy +of this software and associated documentation files (the "Software"), to deal +in the Software without restriction, including without limitation the rights +to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +copies of the Software, and to permit persons to whom the Software is +furnished to do so, subject to the following conditions: + +The above copyright notice and this permission notice shall be included in all +copies or substantial portions of the Software. + +THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +SOFTWARE. diff --git a/app/cmd.go b/app/cmd.go new file mode 100644 index 0000000..f8b6895 --- /dev/null +++ b/app/cmd.go @@ -0,0 +1,117 @@ +package main + +import ( + "fmt" + "net/http" + "net/url" + "runtime" + "strings" + "sync" + // "time" +) + +type Info struct { + room string + Server string `json:"server"` + Proxy string `json:"proxy"` + Start int64 `json:"start"` + Last int64 `json:"last"` + Income int64 `json:"income"` +} + +type Debug struct { + Goroutines int + Alloc uint64 + HeapSys uint64 + Uptime int64 +} + +type Worker struct { + chQuit chan struct{} +} + +type Workers struct { + sync.RWMutex + Map map[string]*Worker +} + +var ( + memInfo runtime.MemStats + chWorker = &Workers{Map: make(map[string]*Worker)} +) + +func removeRoom(room string) { + if checkWorker(room) { + chWorker.Lock() + //fmt.Printf("%v remove %v from chWorker.Map \n", time.Now().UnixMilli(), room ) + delete(chWorker.Map, room) + chWorker.Unlock() + } +} + +func checkWorker(room string) bool { + chWorker.RLock() + defer chWorker.RUnlock() + if _, ok := chWorker.Map[room]; ok { + return true + } + return false +} + +func listRooms() string { + rooms.Json <- "" + s := <-rooms.Json + return s +} + +func listHandler(w http.ResponseWriter, _ *http.Request) { + fmt.Fprint(w, listRooms()) +} + +func debugHandler(w http.ResponseWriter, _ *http.Request) { + runtime.ReadMemStats(&memInfo) + j, err := json.Marshal(Debug{runtime.NumGoroutine(), memInfo.Alloc, memInfo.HeapSys, uptime}) + if err == nil { + fmt.Fprint(w, string(j)) + } +} + +func cmdHandler(w http.ResponseWriter, r *http.Request) { + if !conf.List[r.Header.Get("X-REAL-IP")] { + fmt.Fprint(w, "403") + return + } + + params := r.URL.Query() + if len(params["room"]) > 0 && len(params["server"]) > 0 && len(params["proxy"]) > 0 { + room := params["room"][0] + server := params["server"][0] + proxy := params["proxy"][0] + if checkWorker(room) { + fmt.Println("Already track:", room) + return + } + + info, ok := getRoomInfo(room) + if !ok { + fmt.Println("No room in MySQL:", room) + return + } + + chQuit := make(chan struct{}) + + chWorker.Lock() + chWorker.Map[room] = &Worker{chQuit: chQuit} + chWorker.Unlock() + + go statRoom(chQuit, room, server, proxy, info, url.URL{Scheme: "wss", Host: server + ".bcccdn.com", Path: "/websocket"}) + + } + if len(params["exit"]) > 0 { + room := strings.Join(params["exit"], "") + if checkWorker(room) { + close(chWorker.Map[room].chQuit) // exit gorutine + removeRoom(room) + } + } +} diff --git a/app/conf.go b/app/conf.go new file mode 100644 index 0000000..6176b5b --- /dev/null +++ b/app/conf.go @@ -0,0 +1,33 @@ +package main + +type Conf struct { + Conn map[string]string + Proxy map[string]string + List map[string]bool +} + +var conf = &Conf{ + Conn: make(map[string]string), + Proxy: make(map[string]string), + List: make(map[string]bool), +} + +func startConfig() { + // DB + conf.Conn = map[string]string{ + "mysql": "user:passwd@unix(/var/run/mysqld/mysqld.sock)/base?interpolateParams=true", + "click": "tcp://127.0.0.1:9000/base?compress=true&debug=false", + } + + // 3proxy + conf.Proxy = map[string]string{ + "us": "ip:port", + "fi": "ip:port", + } + + // allow ips + conf.List = map[string]bool{ + "::1": true, + "127.0.0.1": true, + } +} diff --git a/app/echo.go b/app/echo.go new file mode 100644 index 0000000..bbd46d2 --- /dev/null +++ b/app/echo.go @@ -0,0 +1,89 @@ +package main + +import ( + "github.com/gorilla/websocket" + "net/http" + //"fmt" +) + +func newHub() *Hub { + return &Hub{ + broadcast: make(chan []byte), + register: make(chan *Client), + unregister: make(chan *Client), + clients: make(map[*Client]bool), + } +} + +type Hub struct { + clients map[*Client]bool + broadcast chan []byte + register chan *Client + unregister chan *Client +} + +type Client struct { + hub *Hub + conn *websocket.Conn + send chan []byte +} + +func (h *Hub) run() { + for { + select { + case client := <-h.register: + h.clients[client] = true + case client := <-h.unregister: + if _, ok := h.clients[client]; ok { + delete(h.clients, client) + close(client.send) + } + case message := <-h.broadcast: + //fmt.Println("map channel:", len(h.broadcast), cap(h.broadcast)) + for client := range h.clients { + select { + case client.send <- message: + default: + close(client.send) + delete(h.clients, client) + } + } + } + } +} + +func (c *Client) writePump() { + for { + message, ok := <-c.send + if !ok { + // The hub closed the channel. + c.conn.WriteMessage(websocket.CloseMessage, []byte{}) + return + } + c.conn.WriteMessage(1, message) + } + c.conn.Close() +} + +func (c *Client) readPump() { + for { + // Client close connection + _, _, err := c.conn.ReadMessage() + if err != nil { + break + } + } + c.hub.unregister <- c + c.conn.Close() +} + +func (hub *Hub) wsHandler(w http.ResponseWriter, r *http.Request) { + conn, err := websocket.Upgrade(w, r, w.Header(), 1024, 1024) + if err != nil { + return + } + client := &Client{hub: hub, conn: conn, send: make(chan []byte)} + client.hub.register <- client + go client.readPump() + go client.writePump() +} diff --git a/app/main.go b/app/main.go new file mode 100644 index 0000000..77aeb20 --- /dev/null +++ b/app/main.go @@ -0,0 +1,86 @@ +package main + +import ( + "log" + "math/rand" + "net" + "net/http" + "os" + "time" + + _ "github.com/ClickHouse/clickhouse-go" + _ "github.com/go-sql-driver/mysql" + "github.com/jmoiron/sqlx" + jsoniter "github.com/json-iterator/go" +) + +type Rooms struct { + Count chan int + Json chan string + Add chan Info + Del chan string +} + +var hub = newHub() +var Mysql, Clickhouse *sqlx.DB +var json = jsoniter.ConfigCompatibleWithStandardLibrary + +var save = make(chan saveData, 100) +var slog = make(chan saveLog, 100) + +var rooms = &Rooms{ + Count: make(chan int), + Json: make(chan string), + Add: make(chan Info), + Del: make(chan string), +} + +func main() { + rand.Seed(time.Now().UnixNano()) + + startConfig() + + initMysql() + initClickhouse() + + go hub.run() + go mapRooms() + go announceCount() + go saveDB() + go saveLogs() + + http.HandleFunc("/bongacams/ws/", hub.wsHandler) + http.HandleFunc("/bongacams/cmd/", cmdHandler) + http.HandleFunc("/bongacams/list/", listHandler) + http.HandleFunc("/bongacams/debug/", debugHandler) + + const SOCK = "/tmp/bongacams.sock" + os.Remove(SOCK) + unixListener, err := net.Listen("unix", SOCK) + if err != nil { + log.Fatal("Listen (UNIX socket): ", err) + } + defer unixListener.Close() + os.Chmod(SOCK, 0777) + log.Fatal(http.Serve(unixListener, nil)) +} + +func initMysql() { + db, err := sqlx.Connect("mysql", conf.Conn["mysql"]) + if err != nil { + panic(err) + } + Mysql = db +} + +func initClickhouse() { + db, err := sqlx.Connect("clickhouse", conf.Conn["click"]) + if err != nil { + panic(err) + } + Clickhouse = db +} + +func randInt(min int, max int) int { + return min + rand.Intn(max-min) +} diff --git a/app/save.go b/app/save.go new file mode 100644 index 0000000..c327406 --- /dev/null +++ b/app/save.go @@ -0,0 +1,143 @@ +package main + +import ( + "fmt" + "time" +) + +type tID struct { + Id int64 `db:"id"` +} + +type saveData struct { + Room string + From string + Rid int64 + Amount int64 + Now int64 +} + +type saveLog struct { + Rid int64 + Now int64 + Mes string +} + +type DonatorCache struct { + Id int64 + Last int64 +} + +func getDonId(name string) int64 { + donator := new(tID) + err := Mysql.Get(donator, "SELECT id FROM donator WHERE name=?", name) + if err != nil { + res, _ := Mysql.Exec("INSERT INTO donator (`name`) VALUES (?)", name) + donator.Id, _ = res.LastInsertId() + } + return donator.Id +} + +func getRoomInfo(name string) (*tID, bool) { + result := true + room := new(tID) + err := Mysql.Get(room, "SELECT id FROM room WHERE name=?", name) + if err != nil { + result = false + } + return room, result +} + +func saveDB() { + last := time.Now().Unix() + bulk := make(map[int]saveData) + data := make(map[string]*DonatorCache) + + for { + select { + case m := <-save: + //fmt.Println("Save channel:", len(save), cap(save)) + + now := time.Now().Unix() + + if _, ok := data[m.From]; ok { + data[m.From].Last = now + } else { + data[m.From] = &DonatorCache{Id: getDonId(m.From), Last: now} + } + + if randInt(0, 10000) == 777 { // 0.001% + l := len(data) + for k, v := range data { + if now > v.Last+60*60*48 { + delete(data, k) + } + } + fmt.Println("Clean map:", l, "=>", len(data)) + } + + Mysql.Exec("UPDATE `room` SET `last` = ? WHERE `id` = ?", m.Now, m.Rid) + + num := len(bulk) + + bulk[num] = m + + if num >= 999 || now >= last+10 { + tx, err := Mysql.Begin() + if err == nil { + for _, v := range bulk { + tx.Exec("INSERT INTO `stat` (`did`, `rid`, `token`, `time`) VALUES (?, ?, ?, ?)", data[v.From].Id, v.Rid, v.Amount, v.Now) + } + } + tx.Commit() + + tx, err = Clickhouse.Begin() + if err == nil { + st, _ := tx.Prepare("INSERT INTO stat VALUES (?, ?, ?, ?)") + //fmt.Println("G:", err) + for _, v := range bulk { + st.Exec(uint32(data[v.From].Id), uint32(v.Rid), uint32(v.Amount), time.Unix(v.Now, 0)) + //fmt.Println("B:", aaa, sss) + } + tx.Commit() + st.Close() + } + + last = now + bulk = make(map[int]saveData) + } + if m.Amount > 99 { + msg, err := json.Marshal(AnnounceDonate{Room: m.Room, Donator: m.From, Amount: m.Amount}) + if err == nil { + hub.broadcast <- msg + } + } + } + } +} + +func saveLogs() { + last := time.Now().Unix() + bulk := make(map[int]saveLog) + for { + select { + case m := <-slog: + if len(m.Mes) > 0 { + num := len(bulk) + bulk[num] = m + now := time.Now().Unix() + if num >= 2047 || now >= last+10 { + tx, err := Mysql.Begin() + if err == nil { + for _, v := range bulk { + tx.Exec("INSERT INTO `logs` (`rid`, `time`, `mes`) VALUES (?, ?, ?)", v.Rid, v.Now, v.Mes) + } + tx.Commit() + } + last = now + bulk = make(map[int]saveLog) + } + } + } + } +} diff --git a/app/worker.go b/app/worker.go new file mode 100644 index 0000000..f69c330 --- /dev/null +++ b/app/worker.go @@ -0,0 +1,284 @@ +package main + +import ( + "fmt" + "github.com/gorilla/websocket" + jsoniter "github.com/json-iterator/go" + "net/http" + "net/url" + "strings" + "time" + //"bytes" +) + +var uptime = time.Now().Unix() + +type AuthResponse struct { + Status string `json:"status"` + LocalData struct { + DataKey string `json:"dataKey"` + } `json:"localData"` + UserData struct { + Username string `json:"username"` + DisplayName string `json:"displayName"` + Location string `json:"location"` + Chathost string `json:"chathost"` + IsRu bool `json:"isRu"` + } `json:"userData"` +} + +type ServerResponse struct { + TS int64 `json:"ts"` + Type string `json:"type"` + Body jsoniter.RawMessage `json:"body"` +} + +type DonateResponse struct { + F struct { + Username string `json:"username"` + } `json:"f"` + A int64 `json:"a"` +} + +type AnnounceCount struct { + Count int `json:"count"` +} + +type AnnounceDonate struct { + Room string `json:"room"` + Donator string `json:"donator"` + Amount int64 `json:"amount"` +} + +func mapRooms() { + + data := make(map[string]*Info) + + for { + select { + case m := <-rooms.Add: + data[m.room] = &Info{Server: m.Server, Proxy: m.Proxy, Start: m.Start, Last: m.Last, Income: m.Income} + + case s := <-rooms.Json: + j, err := json.Marshal(data) + if err == nil { + s = string(j) + } + rooms.Json <- s + + case <-rooms.Count: + rooms.Count <- len(data) + + case key := <-rooms.Del: + delete(data, key) + removeRoom(key) + } + } +} + +func announceCount() { + for { + time.Sleep(30 * time.Second) + rooms.Count <- 0 + l := <-rooms.Count + msg, err := json.Marshal(AnnounceCount{Count: l}) + if err == nil { + hub.broadcast <- msg + } + } +} + +func getAMF(room string) (bool, *AuthResponse) { + + v := &AuthResponse{} + + req, err := http.NewRequest(http.MethodPost, "https://rt.bongocams.com/tools/amf.php?res=771840&t=1654437233142", strings.NewReader(`method=getRoomData&args[]=`+room)) + if err != nil { + fmt.Println(err.Error()) + return false, v + } + req.Header.Add("Content-Type", "application/x-www-form-urlencoded; charset=UTF-8") + req.Header.Add("X-Requested-With", "XMLHttpRequest") + req.Header.Add("Accept", "application/json") + req.Header.Add("Referrer", "https://bongacams.com") + req.Header.Add("User-agent", "curl/7.79.1") + + rsp, err := http.DefaultClient.Do(req) + if err != nil { + fmt.Println(err.Error()) + return false, v + } + defer rsp.Body.Close() + + if err = json.NewDecoder(rsp.Body).Decode(v); err != nil { + fmt.Println(err.Error()) + return false, v + } + + return true, v +} + +func statRoom(chQuit chan struct{}, room, server, proxy string, info *tID, u url.URL) { + //fmt.Println("Start", room, "server", server, "proxy", proxy) + + ok, v := getAMF(room) + if !ok { + fmt.Println("exit: no amf parms") + return + } + + Dialer := *websocket.DefaultDialer + + if _, ok := conf.Proxy[proxy]; ok { + Dialer = websocket.Dialer{ + Proxy: http.ProxyURL(&url.URL{ + Scheme: "http", // or "https" depending on your proxy + Host: conf.Proxy[proxy], + Path: "/", + }), + HandshakeTimeout: 45 * time.Second, // https://pkg.go.dev/github.com/gorilla/websocket + } + } + + now := time.Now().Unix() + workerData := Info{room, server, proxy, now, now, 0} + rooms.Add <- workerData + + defer func() { + fmt.Println("defer remove map", room) + rooms.Del <- room + }() + + c, _, err := Dialer.Dial(u.String(), nil) + if err != nil { + fmt.Println(err.Error(), room) + return + } + + defer func() { + fmt.Println("defer close", room) + c.Close() + }() + + c.SetReadDeadline(time.Now().Add(60 * time.Second)) + + fmt.Println("send first", room) + + if err = c.WriteMessage(websocket.TextMessage, []byte(fmt.Sprintf(`{"id":%d,"name":"joinRoom","args":["%s",{"username":"%s","displayName":"%s","location":"%s","chathost":"%s","isRu":%t,"isPerformer":false,"hasStream":false,"isLogged":false,"isPayable":false,"showType":"public"},"%s"]}`, 1, v.UserData.Chathost, v.UserData.Username, v.UserData.DisplayName, v.UserData.Location, v.UserData.Chathost, v.UserData.IsRu, v.LocalData.DataKey))); err != nil { + fmt.Println(err.Error()) + return + } + + fmt.Println("read first", room) + + _, message, err := c.ReadMessage() + if err != nil { + fmt.Println(err.Error(), room) + return + } + + fmt.Println(room, len(string(message)), string(message)) + + slog <- saveLog{info.Id, now, string(message)} + + if string(message) == `{"id":1,"result":{"audioAvailable":false,"freeShow":false},"error":null}` { + fmt.Println("room offline, exit", room) + return + } + + fmt.Println("send second", room) + + if err = c.WriteMessage(websocket.TextMessage, []byte(fmt.Sprintf(`{"id":%d,"name":"ChatModule.connect","args":["public-chat"]}`, 2))); err != nil { + fmt.Println(err.Error(), room) + return + } + + fmt.Println("read second", room) + _, message, err = c.ReadMessage() + if err != nil { + fmt.Println(err.Error(), room) + return + } + + fmt.Println(room, len(string(message)), string(message)) + + slog <- saveLog{info.Id, now, string(message)} + quit := make(chan bool) + pid := 3 + + defer func() { + fmt.Println("defer quit", room) + quit <- true + }() + + go func() { + ticker := time.NewTicker(30 * time.Second) + defer ticker.Stop() + for { + select { + case <-quit: + return + case <-ticker.C: + if err = c.WriteMessage(websocket.TextMessage, []byte(fmt.Sprintf(`{"id":%d,"name":"ping"}`, pid))); err != nil { + fmt.Println(err.Error(), room) + close(chWorker.Map[room].chQuit) + return + } + pid++ + break + } + } + }() + + for { + select { + case <-chQuit: + fmt.Println("Exit room:", room) + return + + default: + c.SetReadDeadline(time.Now().Add(30 * time.Minute)) + _, message, err := c.ReadMessage() + if err != nil { + fmt.Println(err.Error()) + return + } + + now = time.Now().Unix() + + slog <- saveLog{info.Id, now, string(message)} + + m := &ServerResponse{} + + if err = json.Unmarshal(message, m); err != nil { + fmt.Println(err.Error(), room) + continue + } + + workerData.Last = now + rooms.Add <- workerData + + if m.Type == "ServerMessageEvent:PERFORMER_STATUS_CHANGE" && string(m.Body) == `"offline"` { + fmt.Println(m.Type, room) + return + } + + if m.Type == "ServerMessageEvent:ROOM_CLOSE" { + fmt.Println(m.Type, room) + return + } + + if m.Type == "ServerMessageEvent:INCOMING_TIP" { + d := &DonateResponse{} + if err = json.Unmarshal(m.Body, d); err == nil { + //fmt.Println(d.F.Username, " send ", d.A, "tokens") + + save <- saveData{room, d.F.Username, info.Id, d.A, now} + + workerData.Income += d.A + rooms.Add <- workerData + } + } + } + } +} diff --git a/test/main.go b/test/main.go new file mode 100644 index 0000000..d1d27b7 --- /dev/null +++ b/test/main.go @@ -0,0 +1,22 @@ +package main + +import ( + "os" + "fmt" + "net/url" +) + +func main() { + if len(os.Args) < 4 { + fmt.Println("./test room server proxy") + return + } + + room := os.Args[1] + server := os.Args[2] + proxy := os.Args[3] + + u := url.URL{Scheme: "wss", Host: server + ".bcccdn.com", Path: "/websocket"} + + startRoom(room, server, proxy, u) +} diff --git a/test/worker.go b/test/worker.go new file mode 100644 index 0000000..2358d0b --- /dev/null +++ b/test/worker.go @@ -0,0 +1,169 @@ +package main + +import ( + "bytes" + "encoding/json" + "fmt" + "net/http" + "net/url" + "strings" + "time" + + "github.com/gorilla/websocket" +) + +type AuthResponse struct { + Status string `json:"status"` + LocalData struct { + DataKey string `json:"dataKey"` + } `json:"localData"` + UserData struct { + Username string `json:"username"` + DisplayName string `json:"displayName"` + Location string `json:"location"` + Chathost string `json:"chathost"` + IsRu bool `json:"isRu"` + } `json:"userData"` +} + +type ServerResponse struct { + TS int64 `json:"ts"` + Type string `json:"type"` + Body json.RawMessage `json:"body"` +} + +type DonateResponse struct { + F struct { + Username string `json:"username"` + } `json:"f"` + A int `json:"a"` +} + +func getAMF(room string) (bool, *AuthResponse) { + + v := &AuthResponse{} + + req, err := http.NewRequest(http.MethodPost, "https://rt.bongocams.com/tools/amf.php?res=771840&t=1654437233142", strings.NewReader(`method=getRoomData&args[]=`+room)) + if err != nil { + fmt.Println(err.Error()) + return false, v + } + req.Header.Add("Content-Type", "application/x-www-form-urlencoded; charset=UTF-8") + req.Header.Add("X-Requested-With", "XMLHttpRequest") + req.Header.Add("Accept", "application/json") + req.Header.Add("Referrer", "https://bongacams.com") + req.Header.Add("User-agent", "curl/7.79.1") + + rsp, err := http.DefaultClient.Do(req) + if err != nil { + fmt.Println(err.Error()) + return false, v + } + defer rsp.Body.Close() + + if err = json.NewDecoder(rsp.Body).Decode(v); err != nil { + fmt.Println(err.Error()) + return false, v + } + + return true, v +} + +func startRoom(room, server, proxy string, u url.URL) { + // curl -vvv -X POST -H "Content-Type: application/x-www-form-urlencoded; charset=UTF-8" -H "X-Requested-With: XMLHttpRequest" -d "method=getRoomData" -d "args[]=Icehotangel" "https://rt.bongocams.com/tools/amf.php?res=771840&t=1654437233142" + + fmt.Println("Start", room, "server", server, "proxy", proxy) + + ok, v := getAMF(room) + if !ok { + fmt.Println("exit: no amf parms") + return + } + + Dialer := *websocket.DefaultDialer + + proxyMap := make(map[string]string) + proxyMap["us"] = "aaa:port" + proxyMap["fi"] = "bbb:port" + + if _, ok := proxyMap[proxy]; ok { + Dialer = websocket.Dialer{ + Proxy: http.ProxyURL(&url.URL{ + Scheme: "http", // or "https" depending on your proxy + Host: proxyMap[proxy], + Path: "/", + }), + HandshakeTimeout: 45 * time.Second, // https://pkg.go.dev/github.com/gorilla/websocket + } + } + + c, _, err := Dialer.Dial(u.String(), nil) + if err != nil { + fmt.Println(err.Error()) + return + } + + pid := 1 + now := time.Now().Unix() + ping := now+30 + + if err = c.WriteMessage(websocket.TextMessage, []byte(fmt.Sprintf(`{"id":%d,"name":"joinRoom","args":["%s",{"username":"%s","displayName":"%s","location":"%s","chathost":"%s","isRu":%t,"isPerformer":false,"hasStream":false,"isLogged":false,"isPayable":false,"showType":"public"},"%s"]}`, pid, v.UserData.Chathost, v.UserData.Username, v.UserData.DisplayName, v.UserData.Location, v.UserData.Chathost, v.UserData.IsRu, v.LocalData.DataKey))); err != nil { + fmt.Println(err.Error()) + return + } + + + defer c.Close() + for { + _, message, err := c.ReadMessage() + if err != nil { + fmt.Println("return") + return + } + + if pid == 1 { + pid++ + fmt.Println("send ChatModule.connect") + if err = c.WriteMessage(websocket.TextMessage, []byte(fmt.Sprintf(`{"id":%d,"name":"ChatModule.connect","args":["public-chat"]}`, pid))); err != nil { + fmt.Println(err.Error()) + return + } + continue + } + + now = time.Now().Unix() + + if(now > ping){ + pid++ + fmt.Println("send ping", pid) + if err = c.WriteMessage(websocket.TextMessage, []byte(fmt.Sprintf(`{"id":%d,"name":"ping"}`, pid))); err != nil { + return + } + ping = now+30 + } + + //fmt.Println(string(message))d + + m := &ServerResponse{} + + if err = json.Unmarshal(message, m); err != nil { + fmt.Println(err.Error()) + continue + } + + if m.Type == "ServerMessageEvent:PERFORMER_STATUS_CHANGE" && bytes.Contains(m.Body, []byte(`offile`)) { + return + } + + if m.Type == "ServerMessageEvent:ROOM_CLOSE" { + return + } + + if m.Type == "ServerMessageEvent:INCOMING_TIP" { + d := &DonateResponse{} + if err = json.Unmarshal(m.Body, d); err == nil { + fmt.Println(d.F.Username, " send ", d.A, "tokens") + } + } + } +}