wide/session/sessions.go

290 lines
6.7 KiB
Go
Raw Normal View History

2014-09-17 10:35:48 +04:00
// 会话操作.
// Wide 服务器端需要维护两种会话:
// 1. HTTP 会话:主要用于验证登录
// 2. Wide 会话:浏览器 tab 打开/刷新会创建一个,并和 HTTP 会话进行关联
//
2014-09-20 06:39:29 +04:00
// 当会话失效时:释放所有和该会话相关的资源,例如运行中的程序进程、事件队列等.
2014-09-17 10:35:48 +04:00
package session
import (
2014-09-22 19:13:07 +04:00
"encoding/json"
2014-09-17 10:35:48 +04:00
"math/rand"
2014-09-19 20:56:32 +04:00
"net/http"
2014-09-19 15:21:13 +04:00
"os"
2014-09-17 10:35:48 +04:00
"strconv"
"sync"
"time"
2014-09-22 19:13:07 +04:00
"github.com/b3log/wide/conf"
2014-09-19 15:21:13 +04:00
"github.com/b3log/wide/event"
2014-09-19 20:56:32 +04:00
"github.com/b3log/wide/util"
2014-09-17 10:35:48 +04:00
"github.com/golang/glog"
"github.com/gorilla/sessions"
2014-09-19 20:56:32 +04:00
"github.com/gorilla/websocket"
2014-09-17 10:35:48 +04:00
)
const (
SessionStateActive = iota // 会话状态:活的
2014-09-20 06:39:29 +04:00
SessionStateClosed // 会话状态:已关闭(这个状态目前暂时没有使用到)
2014-09-17 10:35:48 +04:00
)
2014-09-20 06:39:29 +04:00
var (
// 会话通道. <sid, *util.WSChannel>var
sessionWS = map[string]*util.WSChannel{}
// 输出通道. <sid, *util.WSChannel>
OutputWS = map[string]*util.WSChannel{}
// 通知通道. <sid, *util.WSChannel>
NotificationWS = map[string]*util.WSChannel{}
)
2014-09-19 20:56:32 +04:00
2014-09-17 10:35:48 +04:00
// 用户 HTTP 会话,用于验证登录.
var HTTPSession = sessions.NewCookieStore([]byte("BEYOND"))
// Wide 会话,对应一个浏览器 tab.
type WideSession struct {
2014-09-22 19:13:07 +04:00
Id string // 唯一标识
Username string // 用户名
HTTPSession *sessions.Session // 关联的 HTTP 会话
Processes []*os.Process // 关联的进程集
EventQueue *event.UserEventQueue // 关联的事件队列
State int // 状态
Content *conf.LatestSessionContent // 最近一次会话内容
Created time.Time // 创建时间
Updated time.Time // 最近一次使用时间
2014-09-17 10:35:48 +04:00
}
type Sessions []*WideSession
// 所有 Wide 会话集.
var WideSessions Sessions
2014-09-19 15:21:13 +04:00
// 排它锁,防止并发修改.
2014-09-17 10:35:48 +04:00
var mutex sync.Mutex
2014-09-19 20:56:32 +04:00
// 建立会话通道.
// 通道断开时销毁会话状态,回收相关资源.
func WSHandler(w http.ResponseWriter, r *http.Request) {
sid := r.URL.Query()["sid"][0]
wSession := WideSessions.Get(sid)
if nil == wSession {
glog.Errorf("Session [%s] not found", sid)
return
}
conn, _ := websocket.Upgrade(w, r, nil, 1024, 1024)
wsChan := util.WSChannel{Sid: sid, Conn: conn, Request: r, Time: time.Now()}
sessionWS[sid] = &wsChan
ret := map[string]interface{}{"output": "Ouput initialized", "cmd": "init-session"}
wsChan.Conn.WriteJSON(&ret)
glog.V(4).Infof("Open a new [Session Channel] with session [%s], %d", sid, len(sessionWS))
input := map[string]interface{}{}
for {
if err := wsChan.Conn.ReadJSON(&input); err != nil {
glog.V(3).Infof("[Session Channel] of session [%s] disconnected, releases all resources with it", sid)
2014-09-20 06:39:29 +04:00
for i, s := range WideSessions {
if s.Id == sid {
mutex.Lock()
2014-09-19 20:56:32 +04:00
2014-09-20 06:39:29 +04:00
// 从会话集中移除
WideSessions = append(WideSessions[:i], WideSessions[i+1:]...)
2014-09-19 20:56:32 +04:00
2014-09-20 06:39:29 +04:00
// 关闭用户事件队列
event.UserEventQueues.Close(sid)
2014-09-19 20:56:32 +04:00
2014-09-20 06:39:29 +04:00
// 杀进程
for _, p := range s.Processes {
if err := p.Kill(); nil != err {
glog.Errorf("Can't kill process [%d] of session [%s]", p.Pid, sid)
} else {
glog.V(3).Infof("Killed a process [%d] of session [%s]", p.Pid, sid)
}
}
2014-09-19 20:56:32 +04:00
2014-09-20 06:39:29 +04:00
// 回收所有通道
OutputWS[sid].Close()
delete(OutputWS, sid)
2014-09-19 20:56:32 +04:00
2014-09-20 06:39:29 +04:00
NotificationWS[sid].Close()
delete(NotificationWS, sid)
sessionWS[sid].Close()
delete(sessionWS, sid)
mutex.Unlock()
return
}
}
2014-09-19 20:56:32 +04:00
}
ret = map[string]interface{}{"output": "", "cmd": "session-output"}
if err := wsChan.Conn.WriteJSON(&ret); err != nil {
glog.Error("Session WS ERROR: " + err.Error())
return
}
wsChan.Time = time.Now()
}
}
2014-09-22 19:13:07 +04:00
// 会话内容保存.
func SaveContent(w http.ResponseWriter, r *http.Request) {
data := map[string]interface{}{"succ": true}
defer util.RetJSON(w, r, data)
var args map[string]interface{}
if err := json.NewDecoder(r.Body).Decode(&args); err != nil {
glog.Error(err)
data["succ"] = false
return
}
sid := args["sid"].(string)
wSession := WideSessions.Get(sid)
if nil == wSession {
data["succ"] = false
return
}
wSession.Content.CurrentFile = args["currentFile"].(string)
// TODO: Ugly
fileTree := args["fileTree"].([]interface{})
ft := []string{}
for _, v := range fileTree {
ft = append(ft, v.(string))
}
wSession.Content.FileTree = ft
files := args["files"].([]interface{})
fs := []string{}
for _, v := range files {
fs = append(fs, v.(string))
}
wSession.Content.Files = fs
for _, user := range conf.Wide.Users {
if user.Name == wSession.Username {
user.LatestSessionContent = wSession.Content
// 定时任务会负责持久化
return
}
}
}
2014-09-19 15:21:13 +04:00
// 设置会话关联的进程集.
func (s *WideSession) SetProcesses(ps []*os.Process) {
s.Processes = ps
s.Refresh()
}
// 刷新会话最近一次使用时间.
func (s *WideSession) Refresh() {
s.Updated = time.Now()
}
2014-09-17 10:35:48 +04:00
// 创建一个 Wide 会话.
func (sessions *Sessions) New(httpSession *sessions.Session) *WideSession {
mutex.Lock()
defer mutex.Unlock()
rand.Seed(time.Now().UnixNano())
id := strconv.Itoa(rand.Int())
now := time.Now()
2014-09-20 06:39:29 +04:00
// 创建用户事件队列
2014-09-19 15:21:13 +04:00
userEventQueue := event.UserEventQueues.New(id)
2014-09-17 10:35:48 +04:00
ret := &WideSession{
Id: id,
2014-09-22 19:13:07 +04:00
Username: httpSession.Values["username"].(string),
2014-09-17 10:35:48 +04:00
HTTPSession: httpSession,
2014-09-19 15:21:13 +04:00
EventQueue: userEventQueue,
2014-09-17 10:35:48 +04:00
State: SessionStateActive,
2014-09-22 19:13:07 +04:00
Content: &conf.LatestSessionContent{},
2014-09-17 10:35:48 +04:00
Created: now,
Updated: now,
}
*sessions = append(*sessions, ret)
return ret
}
2014-09-19 15:21:13 +04:00
// 获取 Wide 会话.
func (sessions *Sessions) Get(sid string) *WideSession {
mutex.Lock()
defer mutex.Unlock()
for _, s := range *sessions {
if s.Id == sid {
return s
}
}
return nil
}
2014-09-17 10:35:48 +04:00
// 移除 Wide 会话.
func (sessions *Sessions) Remove(sid string) {
mutex.Lock()
defer mutex.Unlock()
for i, s := range *sessions {
if s.Id == sid {
*sessions = append((*sessions)[:i], (*sessions)[i+1:]...)
2014-09-19 20:56:32 +04:00
glog.V(3).Infof("Removed a session [%s], has [%d] wide sessions currently", sid, len(*sessions))
return
2014-09-17 10:35:48 +04:00
}
}
}
// 获取 HTTP 会话关联的所有 Wide 会话.
func (sessions *Sessions) GetByHTTPSession(httpSession *sessions.Session) []*WideSession {
mutex.Lock()
defer mutex.Unlock()
ret := []*WideSession{}
for _, s := range *sessions {
if s.HTTPSession.ID == httpSession.ID {
ret = append(ret, s)
}
}
return ret
}
// 移除 HTTP 会话关联的所有 Wide 会话.
func (sessions *Sessions) RemoveByHTTPSession(httpSession *sessions.Session) {
mutex.Lock()
defer mutex.Unlock()
for i, s := range *sessions {
if s.HTTPSession.ID == httpSession.ID {
*sessions = append((*sessions)[:i], (*sessions)[i+1:]...)
glog.V(3).Infof("Has [%d] wide sessions currently", len(*sessions))
}
}
}