improved response handling at websocket interface

2 files changed, 129 insertions, 25 deletions

diff --git a/src/helsinki.at/rhimport/session_store.go b/src/helsinki.at/rhimport/session_store.go
index 7fe6585..0d149c6 100644
--- a/src/helsinki.at/rhimport/session_store.go
+++ b/src/helsinki.at/rhimport/session_store.go
@@ -58,6 +58,19 @@ type getSessionRequest struct {
 	response chan getSessionResponse
 }
 
+type listSessionsResponse struct {
+	sessions     map[string]string
+	responsecode int
+	errorstring  string
+}
+
+type listSessionsRequest struct {
+	user     string
+	password string
+	trusted  bool
+	response chan listSessionsResponse
+}
+
 type removeSessionResponse struct {
 	responsecode int
 	errorstring  string
@@ -80,6 +93,7 @@ type SessionStore struct {
 	done       chan bool
 	newChan    chan newSessionRequest
 	getChan    chan getSessionRequest
+	listChan   chan listSessionsRequest
 	removeChan chan removeSessionRequest
 }
 
@@ -134,6 +148,30 @@ func (self *SessionStore) get(user, id string) (resp getSessionResponse) {
 	return
 }
 
+func (self *SessionStore) list(user, password string, trusted bool) (resp listSessionsResponse) {
+	resp.responsecode = http.StatusOK
+	resp.errorstring = "OK"
+	// TODO: enable this check as soon as the session store handles the rddb itself
+	// if !trusted {
+	// 	if ok, err := self.rddb.CheckPassword(user, password); err != nil {
+	// 		resp.responsecode = http.StatusInternalServerError
+	// 		resp.errorstring = err.Error()
+	// 		return
+	// 	} else if !ok {
+	// 		resp.responsecode = http.StatusUnauthorized
+	// 		resp.errorstring = "invalid username and/or password"
+	// 		return
+	// 	}
+	// }
+	resp.sessions = make(map[string]string)
+	if sessions, exists := self.store[user]; exists {
+		for id, e := range sessions {
+			resp.sessions[id] = e.refId
+		}
+	}
+	return
+}
+
 func (self *SessionStore) remove(user, id string) (resp removeSessionResponse) {
 	resp.responsecode = http.StatusOK
 	resp.errorstring = "OK"
@@ -164,6 +202,8 @@ func (self *SessionStore) dispatchRequests() {
 			req.response <- self.new(req.ctx, req.refId)
 		case req := <-self.getChan:
 			req.response <- self.get(req.user, req.id)
+		case req := <-self.listChan:
+			req.response <- self.list(req.user, req.password, req.trusted)
 		case req := <-self.removeChan:
 			req.response <- self.remove(req.user, req.id)
 		}
@@ -176,6 +216,7 @@ func (self *SessionStore) dispatchRequests() {
 type SessionStoreChan struct {
 	newChan    chan<- newSessionRequest
 	getChan    chan<- getSessionRequest
+	listChan   chan listSessionsRequest
 	removeChan chan<- removeSessionRequest
 }
 
@@ -203,6 +244,19 @@ func (self *SessionStoreChan) Get(user, id string) (*SessionChan, string, int, s
 	return res.session, res.refId, res.responsecode, res.errorstring
 }
 
+func (self *SessionStoreChan) List(user, password string, trusted bool) (map[string]string, int, string) {
+	res_ch := make(chan listSessionsResponse)
+	req := listSessionsRequest{}
+	req.user = user
+	req.password = password
+	req.trusted = trusted
+	req.response = res_ch
+	self.listChan <- req
+
+	res := <-res_ch
+	return res.sessions, res.responsecode, res.errorstring
+}
+
 func (self *SessionStoreChan) Remove(user, id string) (int, string) {
 	res_ch := make(chan removeSessionResponse)
 	req := removeSessionRequest{}
@@ -219,6 +273,7 @@ func (self *SessionStore) GetInterface() *SessionStoreChan {
 	ch := &SessionStoreChan{}
 	ch.newChan = self.newChan
 	ch.getChan = self.getChan
+	ch.listChan = self.listChan
 	ch.removeChan = self.removeChan
 	return ch
 }
@@ -230,6 +285,7 @@ func (self *SessionStore) Cleanup() {
 	close(self.done)
 	close(self.newChan)
 	close(self.getChan)
+	close(self.listChan)
 	close(self.removeChan)
 }
 
@@ -241,6 +297,7 @@ func NewSessionStore(conf *Config) (store *SessionStore, err error) {
 	store.store = make(map[string]map[string]*SessionStoreElement)
 	store.newChan = make(chan newSessionRequest, 10)
 	store.getChan = make(chan getSessionRequest, 10)
+	store.listChan = make(chan listSessionsRequest, 10)
 	store.removeChan = make(chan removeSessionRequest, 10)
 
 	go store.dispatchRequests()
diff --git a/src/helsinki.at/rhimportd/ctrlWebSocket.go b/src/helsinki.at/rhimportd/ctrlWebSocket.go
index ea7bf24..ca1092c 100644
--- a/src/helsinki.at/rhimportd/ctrlWebSocket.go
+++ b/src/helsinki.at/rhimportd/ctrlWebSocket.go
@@ -75,27 +75,74 @@ func newWebSocketRequestData(conf *rhimport.Config) *webSocketRequestData {
 	return rd
 }
 
-type webSocketResponseData struct {
-	ResponseCode     int     `json:"RESPONSE_CODE"`
-	Type             string  `json:"TYPE"`
-	ErrorString      string  `json:"ERROR_STRING"`
-	Id               string  `json:"ID"`
-	RefId            string  `json:"REFERENCE_ID"`
-	ProgressStep     int     `json:"PROGRESS_STEP"`
-	ProgressStepName string  `json:"PROGRESS_STEP_NAME"`
-	Progress         float64 `json:"PROGRESS"`
-	Cart             uint    `json:"CART_NUMBER"`
-	Cut              uint    `json:"CUT_NUMBER"`
+type webSocketResponseBaseData struct {
+	ResponseCode int    `json:"RESPONSE_CODE"`
+	Type         string `json:"TYPE"`
+	ErrorString  string `json:"ERROR_STRING"`
+	Id           string `json:"ID"`
+	RefId        string `json:"REFERENCE_ID"`
 }
 
-func sendWebSocketResponse(ws *websocket.Conn, rd *webSocketResponseData) {
-	if err := ws.WriteJSON(*rd); err != nil {
+type webSocketResponseProgressData struct {
+	webSocketResponseBaseData
+	Step     int     `json:"PROGRESS_STEP"`
+	StepName string  `json:"PROGRESS_STEP_NAME"`
+	Progress float64 `json:"PROGRESS"`
+}
+
+type webSocketResponseDoneData struct {
+	webSocketResponseBaseData
+	Cart uint `json:"CART_NUMBER"`
+	Cut  uint `json:"CUT_NUMBER"`
+}
+
+func sendWebSocketResponse(ws *websocket.Conn, rd interface{}) {
+	if err := ws.WriteJSON(rd); err != nil {
 		rhdl.Println("WebScoket Client", ws.RemoteAddr(), "write error:", err)
 	}
 }
 
-func sendWebSocketErrorResponse(ws *websocket.Conn, id string, code int, err_str string) {
-	sendWebSocketResponse(ws, &webSocketResponseData{ResponseCode: code, Type: "ERROR", ErrorString: err_str, Id: id})
+func sendWebSocketAckResponse(ws *websocket.Conn, code int, err_str, id, refid string) {
+	rd := &webSocketResponseBaseData{}
+	rd.ResponseCode = code
+	rd.Type = "ACK"
+	rd.ErrorString = err_str
+	rd.Id = id
+	rd.RefId = refid
+	sendWebSocketResponse(ws, rd)
+}
+
+func sendWebSocketErrorResponse(ws *websocket.Conn, code int, err_str string) {
+	rd := &webSocketResponseBaseData{}
+	rd.ResponseCode = code
+	rd.Type = "ERROR"
+	rd.ErrorString = err_str
+	sendWebSocketResponse(ws, rd)
+}
+
+func sendWebSocketProgressResponse(ws *websocket.Conn, id, refid string, step int, step_name string, progress float64) {
+	rd := &webSocketResponseProgressData{}
+	rd.ResponseCode = http.StatusOK
+	rd.Type = "PROGRESS"
+	rd.ErrorString = "OK"
+	rd.Id = id
+	rd.RefId = refid
+	rd.Step = step
+	rd.StepName = step_name
+	rd.Progress = progress
+	sendWebSocketResponse(ws, rd)
+}
+
+func sendWebSocketDoneResponse(ws *websocket.Conn, code int, err_str, id, refid string, cart, cut uint) {
+	rd := &webSocketResponseDoneData{}
+	rd.ResponseCode = code
+	rd.Type = "DONE"
+	rd.ErrorString = err_str
+	rd.Id = id
+	rd.RefId = refid
+	rd.Cart = cart
+	rd.Cut = cut
+	sendWebSocketResponse(ws, rd)
 }
 
 type webSocketSession struct {
@@ -195,39 +242,39 @@ func webSocketSessionHandler(reqchan <-chan webSocketRequestData, ws *websocket.
 			switch reqdata.Command {
 			case "new":
 				if session.id != "" {
-					sendWebSocketErrorResponse(ws, "", http.StatusBadRequest, "This connection already handles a session")
+					sendWebSocketErrorResponse(ws, http.StatusBadRequest, "This connection already handles a session")
 				} else {
 					code, errstring := session.startNewSession(&reqdata, conf, rddb, sessions)
 					if code != http.StatusOK {
-						sendWebSocketErrorResponse(ws, "", code, errstring)
+						sendWebSocketErrorResponse(ws, code, errstring)
 					} else {
-						sendWebSocketResponse(ws, &webSocketResponseData{ResponseCode: code, Type: "ACK", Id: session.id, RefId: session.refId})
+						sendWebSocketAckResponse(ws, code, "OK", session.id, session.refId)
 					}
 				}
 			case "cancel":
 				if session.id == "" {
-					sendWebSocketErrorResponse(ws, "", http.StatusBadRequest, "This connection doesn't handle any session")
+					sendWebSocketErrorResponse(ws, http.StatusBadRequest, "This connection doesn't handle any session")
 				} else {
 					session.session.Cancel()
 				}
 			case "reconnect":
 				if session.id != "" {
-					sendWebSocketErrorResponse(ws, "", http.StatusBadRequest, "This connection already handles a session")
+					sendWebSocketErrorResponse(ws, http.StatusBadRequest, "This connection already handles a session")
 				} else {
 					code, errstring := session.reconnectSession(&reqdata, sessions)
 					if code != http.StatusOK {
-						sendWebSocketErrorResponse(ws, "", code, errstring)
+						sendWebSocketErrorResponse(ws, code, errstring)
 					} else {
-						sendWebSocketResponse(ws, &webSocketResponseData{ResponseCode: code, Type: "ACK", Id: session.id, RefId: session.refId})
+						sendWebSocketAckResponse(ws, code, "OK", session.id, session.refId)
 					}
 				}
 			default:
-				sendWebSocketErrorResponse(ws, "", http.StatusBadRequest, fmt.Sprintf("unknown command '%s'", reqdata.Command))
+				sendWebSocketErrorResponse(ws, http.StatusBadRequest, fmt.Sprintf("unknown command '%s'", reqdata.Command))
 			}
 		case p := <-session.progresschan:
-			sendWebSocketResponse(ws, &webSocketResponseData{http.StatusOK, "PROGRESS", "", session.id, session.refId, p.Step, p.StepName, p.Progress * 100, 0, 0})
+			sendWebSocketProgressResponse(ws, session.id, session.refId, p.Step, p.StepName, p.Progress*100)
 		case d := <-session.donechan:
-			sendWebSocketResponse(ws, &webSocketResponseData{d.ResponseCode, "DONE", d.ErrorString, session.id, session.refId, 0, "", 100.0, d.Cart, d.Cut})
+			sendWebSocketDoneResponse(ws, d.ResponseCode, d.ErrorString, session.id, session.refId, d.Cart, d.Cut)
 			// TODO: send close message at this point?
 		}
 	}