sessions now call remove when done

2 files changed, 37 insertions, 24 deletions

diff --git a/session.go b/session.go
index 8843c81..bce1072 100644
--- a/session.go
+++ b/session.go
@@ -46,7 +46,9 @@ type SessionChan struct {
 type Session struct {
 	ctx             ImportContext
 	state           int
+	removeFunc      func()
 	done            chan bool
+	quit            chan bool
 	cancelIntChan   chan bool
 	progressIntChan chan ProgressData
 	doneIntChan     chan ImportResult
@@ -139,6 +141,15 @@ func (self *SessionChan) Run() {
 	}
 }
 
+func (self *Session) cancel() {
+	rhdl.Println("Session: canceling running import")
+	select {
+	case self.cancelIntChan <- true:
+	default: // session got canceled already??
+	}
+	self.state = SESSION_CANCELED
+}
+
 func (self *SessionChan) Cancel() {
 	select {
 	case self.cancelChan <- true:
@@ -178,6 +189,16 @@ func (self *Session) addDoneHandler(userdata interface{}, cb func(ImportResult,
 	return
 }
 
+func (self *Session) callProgressHandler(p *ProgressData) {
+	for _, cb := range self.progressCBs {
+		if cb.cb != nil {
+			if keep := cb.cb(p.step, p.step_name, p.progress, cb.userdata); !keep {
+				cb.cb = nil
+			}
+		}
+	}
+}
+
 func (self *SessionChan) AddDoneHandler(userdata interface{}, cb func(ImportResult, interface{}) bool) error {
 	res_ch := make(chan sessionAddDoneHandlerResponse)
 	req := sessionAddDoneHandlerRequest{}
@@ -194,16 +215,6 @@ func (self *SessionChan) AddDoneHandler(userdata interface{}, cb func(ImportResu
 	return res.err
 }
 
-func (self *Session) callProgressHandler(p *ProgressData) {
-	for _, cb := range self.progressCBs {
-		if cb.cb != nil {
-			if keep := cb.cb(p.step, p.step_name, p.progress, cb.userdata); !keep {
-				cb.cb = nil
-			}
-		}
-	}
-}
-
 func (self *Session) callDoneHandler(r *ImportResult) {
 	for _, cb := range self.doneCBs {
 		if cb.cb != nil {
@@ -218,20 +229,18 @@ func (self *Session) dispatchRequests() {
 	defer func() { self.done <- true }()
 	for {
 		select {
+		case <-self.quit:
+			if self.state == SESSION_RUNNING {
+				self.cancel()
+			}
+			return
 		case <-self.runChan:
 			if self.state == SESSION_NEW {
 				self.run()
 			}
 		case <-self.cancelChan:
 			if self.state == SESSION_RUNNING {
-				rhdl.Println("Session: canceling running import")
-				select {
-				case self.cancelIntChan <- true:
-				default: // session got canceled already??
-				}
-				self.state = SESSION_CANCELED
-			} else {
-				return
+				self.cancel()
 			}
 		case req := <-self.addProgressChan:
 			req.response <- self.addProgressHandler(req.userdata, req.callback)
@@ -242,6 +251,9 @@ func (self *Session) dispatchRequests() {
 		case r := <-self.doneIntChan:
 			self.state = SESSION_DONE
 			self.callDoneHandler(&r)
+			if self.removeFunc != nil {
+				self.removeFunc()
+			}
 		}
 	}
 }
@@ -256,7 +268,7 @@ func (self *Session) getInterface() *SessionChan {
 }
 
 func (self *Session) Cleanup() {
-	self.cancelChan <- true
+	self.quit <- true
 	rhdl.Printf("waiting for session to close")
 	<-self.done
 	close(self.done)
@@ -271,9 +283,10 @@ func (self *Session) Cleanup() {
 	rhdl.Printf("session is now cleaned up")
 }
 
-func NewSession(ctx *ImportContext) (session *Session) {
+func NewSession(ctx *ImportContext, removeFunc func()) (session *Session) {
 	session = new(Session)
 	session.state = SESSION_NEW
+	session.removeFunc = removeFunc
 	session.ctx = *ctx
 	session.done = make(chan bool)
 	session.cancelIntChan = make(chan bool, 1)
diff --git a/session_store.go b/session_store.go
index 8dd74c7..505bd75 100644
--- a/session_store.go
+++ b/session_store.go
@@ -84,7 +84,7 @@ func (self *SessionStore) new(ctx *ImportContext) (resp newSessionResponse) {
 	if _, exists := self.store[ctx.UserName]; !exists {
 		self.store[ctx.UserName] = make(map[string]*Session)
 	}
-	self.store[ctx.UserName][resp.id] = NewSession(ctx)
+	self.store[ctx.UserName][resp.id] = NewSession(ctx, func() { self.GetInterface().Remove(ctx.UserName, resp.id) })
 	resp.session = self.store[ctx.UserName][resp.id].getInterface()
 	rhdl.Printf("SessionStore: created session for '%s' -> %s", ctx.UserName, resp.id)
 	return
@@ -197,9 +197,9 @@ func NewSessionStore(conf *Config) (store *SessionStore, err error) {
 	store.quit = make(chan bool)
 	store.done = make(chan bool)
 	store.store = make(map[string]map[string]*Session)
-	store.newChan = make(chan newSessionRequest)
-	store.getChan = make(chan getSessionRequest)
-	store.removeChan = make(chan removeSessionRequest)
+	store.newChan = make(chan newSessionRequest, 10)
+	store.getChan = make(chan getSessionRequest, 10)
+	store.removeChan = make(chan removeSessionRequest, 10)
 
 	go store.dispatchRequests()
 	return