jy/src/jfw/front/websocket.go

package front

import (
	"encoding/json"
	"fmt"
	"log"
	//"net/http"
	qutil "qfw/util"
	"qfw/util/redis"
	"strings"
	"sync"
	"time"

	"github.com/go-xweb/httpsession"
	"github.com/go-xweb/xweb"
	"golang.org/x/net/websocket"
)

//socket对象放在内存中，待rpc回调使用
type Wss struct {
	Conn *websocket.Conn
	Time int64
}
type MapSocket struct {
	Map  map[string]*Wss
	Lock sync.Mutex
}

func (m *MapSocket) GC() {
	defer qutil.Catch()
	m.Lock.Lock()
	now := time.Now().Unix()
	for k, v := range m.Map {
		if now-v.Time > 3600 || v.Conn == nil {
			delete(m.Map, k)
		}
	}
	m.Lock.Unlock()
	time.AfterFunc(5*time.Minute, m.GC)
}

var MSPOOL = 20
var MapSocketArr = make([]*MapSocket, MSPOOL)

//初始化
func init() {
	for i := 0; i < MSPOOL; i++ {
		ms := &MapSocket{Map: map[string]*Wss{}}
		go ms.GC()
		MapSocketArr[i] = ms
	}
}

//根据代码和ws做映射
func PutWsByCode(src string, ws *websocket.Conn) {
	defer qutil.Catch()
	n := HashVal(src) % MSPOOL
	ms := MapSocketArr[n]
	ms.Lock.Lock()
	ms.Map[src] = &Wss{ws, time.Now().Unix()}
	ms.Lock.Unlock()
}

//计算代码的hash值
func HashVal(src string) int {
	check := 0
	for i := len(src) / 2; i < len(src); i++ {
		check += int(src[i])
	}
	return check
}

//rpc回调,写到前台
func GetWsByCode(param []string) bool {
	if len(param) < 2 {
		return false
	}
	src := param[0]
	openid := param[1]
	if src == "" {
		return false
	}
	defer qutil.Catch()
	n := HashVal(src) % MSPOOL
	ms := MapSocketArr[n]
	defer func() {
		ms.Lock.Lock()
		delete(ms.Map, src)
		ms.Lock.Unlock()
	}()
	ms.Lock.Lock()
	wss := ms.Map[src]
	ms.Lock.Unlock()
	if wss != nil && wss.Conn != nil {
		infoData := LoginInfo(src, openid, wss.Conn.Sess)
		sendmessage, _ := json.Marshal(infoData)
		if err := websocket.Message.Send(wss.Conn, string(sendmessage)); err != nil {
			log.Println("socket send fail..", err)
			return false
		} else {
			wss.Conn.Close()
		}
		return true
	} else {
		return false

	}
}

//用户登录信息
func LoginInfo(shareid, openid string, Sess interface{}) (infoData map[string]interface{}) {
	if Sess == nil {
		return nil
	}
	sess, _ := Sess.(*httpsession.Session)
	infoData = make(map[string]interface{})
	if openid != "" {
		sess.Set("openid", openid)
		redisheadimg := redis.Get("other", "newUser-"+openid)
		if redisheadimg == nil {
			redisheadimg = ""
		}
		user, _ := mongodb.FindOneByField("user", `{"s_m_openid":"`+openid+`"}`, `{"s_nickname":1,"s_headimage":1,"s_m_openid":1}`)
		if user != nil && len(*user) > 0 {
			infoData["result"] = "ok"
			infoData["s_nickname"] = fmt.Sprint((*user)["s_nickname"])
			infoData["s_headimage"] = fmt.Sprint((*user)["s_headimage"])
			infoData["redisheadimg"] = fmt.Sprint(redisheadimg)
			infoData["encryptId"] = se.EncodeString(qutil.BsonIdToSId((*user)["_id"]))
			infoData["shareid"] = shareid
			(*user)["shareid"] = shareid
			nick := fmt.Sprint((*user)["s_nickname"])
			sess.Set("nickname", nick)
			sess.Set("s_nickname", nick)
			sess.Set("s_m_openid", fmt.Sprint((*user)["s_m_openid"]))
			sess.Set("user", *user)
		}
	}
	return infoData
}

//登录关注
func ServeWss(conn *websocket.Conn) {
	defer qutil.Catch()
	conn.Sess = xweb.RootApp().SessionManager.Session(conn.R, conn.W)
	var shareIds string
	for {
		var shareData string
		err := websocket.Message.Receive(conn, &shareData)
		if err != nil {
			//log.Println("前台socket关闭，后台socket断开并退出循环。。。。")
			break
		} else {
			//心跳监测
			if shareData == "HeartBeat" {
				websocket.Message.Send(conn, "HeartBeat")
				continue
			}
			shareIds = shareData
			shareidlist := strings.Split(shareIds, "___")
			if shareIds != "" && len(shareidlist) > 1 {
				shareidnum := shareidlist[0]
				shareidkop := shareidlist[1]
				PutWsByCode(se.DecodeString(shareidnum), conn)
				PutWsByCode(se.DecodeString(shareidkop), conn)
				websocket.Message.Send(conn, "ok")
			}
		}
	}

}

//实验室
func QrToLabWss(conn *websocket.Conn) {
	defer qutil.Catch()
	var receive, userId string
	var qrToLab_ok, qrToLab_open_ok bool
	//接收消息
	go func() {
		defer qutil.Catch()
		for {
			err := websocket.Message.Receive(conn, &receive)
			if err != nil {
				receive = "close"
				log.Println("websocket接收失败！", err)
				return
			}
			if receive == "close" {
				return
			} else if receive != "close" { //接收到userid
				userId = se.DecodeString(receive)
			}
		}
	}()
	//发送消息
	for {
		time.Sleep(2 * time.Second)
		if receive == "close" { //接收到关闭信息
			conn.Close()
			return
		} else if userId == "" {
			continue
		}
		var reply string
		//是否进入实验室
		if !qrToLab_ok {
			qrToLab_ok, _ = redis.Exists("other", "qrToLab_"+userId)
			if qrToLab_ok {
				reply = "qrToLab_ok"
			}
		}
		//是否打开开关
		if !qrToLab_open_ok {
			qrToLab_open_ok, _ = redis.Exists("other", "qrToLab_open_"+userId)
			if qrToLab_open_ok {
				reply = "qrToLab_open_ok"
			}
		}
		if reply == "" {
			continue
		}
		sendmessage, _ := json.Marshal(reply)
		if err := websocket.Message.Send(conn, sendmessage); err != nil {
			redis.Del("other", "qrToLab_"+userId)
			redis.Del("other", "qrToLab_open_"+userId)
			//log.Println("websocket发送失败！", err)
			conn.Close()
			return
		}
		if reply == "qrToLab_ok" {
			redis.Del("other", "qrToLab_"+userId)
		} else if reply == "qrToLab_open_ok" {
			redis.Del("other", "qrToLab_open_"+userId)
		}
		if qrToLab_ok && qrToLab_open_ok {
			conn.Close()
			return
		}
	}
}