// // rhimportd // // The Radio Helsinki Rivendell Import Daemon // // // Copyright (C) 2015-2016 Christian Pointner // // This file is part of rhimportd. // // rhimportd is free software: you can redistribute it and/or modify // it under the terms of the GNU General Public License as published by // the Free Software Foundation, either version 3 of the License, or // any later version. // // rhimportd is distributed in the hope that it will be useful, // but WITHOUT ANY WARRANTY; without even the implied warranty of // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the // GNU General Public License for more details. // // You should have received a copy of the GNU General Public License // along with rhimportd. If not, see . // package rhimport import ( "fmt" "net/http" "time" ) const ( SESSION_NEW = iota SESSION_RUNNING SESSION_CANCELED SESSION_DONE SESSION_TIMEOUT ) type SessionProgressCB struct { cb ProgressCB userdata interface{} } type SessionDoneCB struct { cb DoneCB userdata interface{} } type ProgressData struct { Step int StepName string Current float64 Total float64 Title string Cart uint Cut uint } type sessionAddProgressHandlerResponse struct { err error } type sessionAddProgressHandlerRequest struct { userdata interface{} callback ProgressCB response chan<- sessionAddProgressHandlerResponse } type sessionAddDoneHandlerResponse struct { err error } type sessionAddDoneHandlerRequest struct { userdata interface{} callback DoneCB response chan<- sessionAddDoneHandlerResponse } type attachUploaderResponse struct { cancel <-chan bool attachment chan<- AttachmentChunk } type attachUploaderRequest struct { response chan<- attachUploaderResponse } type Session struct { ctx Context state int removeFunc func() done chan bool quit chan bool timer *time.Timer cancelIntChan chan bool progressRateLimit time.Duration progressIntChan chan ProgressData doneIntChan chan Result runChan chan time.Duration cancelChan chan bool addProgressChan chan sessionAddProgressHandlerRequest addDoneChan chan sessionAddDoneHandlerRequest attachUploaderChan chan attachUploaderRequest progressCBs []*SessionProgressCB doneCBs []*SessionDoneCB cancelUploader chan bool } func sessionProgressCallback(step int, stepName string, current, total float64, title string, cart, cut uint, userdata interface{}) bool { out := userdata.(chan<- ProgressData) out <- ProgressData{step, stepName, current, total, title, cart, cut} return true } func sessionRun(ctx Context, done chan<- Result) { err := ctx.SanityCheck() if err != nil { done <- Result{ResponseCode: http.StatusBadRequest, ErrorString: err.Error()} return } var res *Result res, err = FetchFile(&ctx) if err != nil { done <- Result{ResponseCode: http.StatusInternalServerError, ErrorString: err.Error()} return } if res.ResponseCode != http.StatusOK { done <- *res return } if res, err = NormalizeFile(&ctx); err != nil { done <- Result{ResponseCode: http.StatusInternalServerError, ErrorString: err.Error()} return } if res.ResponseCode != http.StatusOK { done <- *res return } if res, err = ImportFile(&ctx); err != nil { res.ResponseCode = http.StatusInternalServerError res.ErrorString = err.Error() } done <- *res } func (self *Session) run(timeout time.Duration) { self.ctx.ProgressCallBack = sessionProgressCallback self.ctx.ProgressCallBackData = (chan<- ProgressData)(self.progressIntChan) self.ctx.Cancel = self.cancelIntChan go sessionRun(self.ctx, self.doneIntChan) self.state = SESSION_RUNNING if timeout > 3*time.Hour { self.ctx.stdlog.Printf("requested session timeout (%v) is to high - lowering to 3h", timeout) timeout = 3 * time.Hour } self.timer.Reset(timeout) return } func (self *Session) cancel() { self.ctx.dbglog.Println("Session: canceling running import") select { case self.cancelIntChan <- true: default: // session got canceled already?? } self.state = SESSION_CANCELED } func (self *Session) addProgressHandler(userdata interface{}, cb ProgressCB) (resp sessionAddProgressHandlerResponse) { if self.state != SESSION_NEW && self.state != SESSION_RUNNING { resp.err = fmt.Errorf("session is already done/canceled") } self.progressCBs = append(self.progressCBs, &SessionProgressCB{cb, userdata}) return } func (self *Session) addDoneHandler(userdata interface{}, cb DoneCB) (resp sessionAddDoneHandlerResponse) { if self.state != SESSION_NEW && self.state != SESSION_RUNNING { resp.err = fmt.Errorf("session is already done/canceled") } self.doneCBs = append(self.doneCBs, &SessionDoneCB{cb, userdata}) return } func (self *Session) callProgressHandler(p *ProgressData) { for _, cb := range self.progressCBs { if cb.cb != nil { if keep := cb.cb(p.Step, p.StepName, p.Current, p.Total, p.Title, p.Cart, p.Cut, cb.userdata); !keep { cb.cb = nil } } } } func (self *Session) callDoneHandler(r *Result) { for _, cb := range self.doneCBs { if cb.cb != nil { if keep := cb.cb(*r, cb.userdata); !keep { cb.cb = nil } } } } func (self *Session) attachUploader() (resp attachUploaderResponse) { if self.cancelUploader != nil { return } self.cancelUploader = make(chan bool, 1) resp.cancel = self.cancelUploader resp.attachment = self.ctx.AttachmentChan return } func (self *Session) dispatchRequests() { defer func() { if self.cancelUploader != nil { close(self.cancelUploader) } self.done <- true }() var lastProgress *ProgressData progressPending := 0 pt := time.NewTimer(self.progressRateLimit) pt.Stop() for { select { case <-self.quit: if self.state == SESSION_RUNNING { self.cancel() } return case <-self.timer.C: if self.state == SESSION_RUNNING { self.cancel() } self.state = SESSION_TIMEOUT r := &Result{ResponseCode: http.StatusInternalServerError, ErrorString: "session timed out"} self.callDoneHandler(r) if self.removeFunc != nil { self.removeFunc() } case t := <-self.runChan: if self.state == SESSION_NEW { self.run(t) } case <-self.cancelChan: if self.state == SESSION_RUNNING { self.cancel() } case req := <-self.addProgressChan: req.response <- self.addProgressHandler(req.userdata, req.callback) case req := <-self.addDoneChan: req.response <- self.addDoneHandler(req.userdata, req.callback) case <-pt.C: if progressPending > 1 && lastProgress != nil { self.callProgressHandler(lastProgress) } progressPending = 0 lastProgress = nil case p := <-self.progressIntChan: if self.state == SESSION_RUNNING { if lastProgress == nil { self.callProgressHandler(&p) pt.Reset(self.progressRateLimit) } else if lastProgress.Step != p.Step { self.callProgressHandler(lastProgress) self.callProgressHandler(&p) pt.Reset(self.progressRateLimit) } lastProgress = &p progressPending++ } case r := <-self.doneIntChan: if self.state != SESSION_TIMEOUT { self.timer.Stop() self.state = SESSION_DONE self.callDoneHandler(&r) if self.removeFunc != nil { self.removeFunc() } } case req := <-self.attachUploaderChan: req.response <- self.attachUploader() } } } // ********************************************************* // Public Interface type SessionChan struct { runChan chan<- time.Duration cancelChan chan<- bool addProgressChan chan<- sessionAddProgressHandlerRequest addDoneChan chan<- sessionAddDoneHandlerRequest attachUploaderChan chan<- attachUploaderRequest } func (self *SessionChan) Run(timeout time.Duration) { select { case self.runChan <- timeout: default: // command is already pending or session is about to be closed/removed } } func (self *SessionChan) Cancel() { select { case self.cancelChan <- true: default: // cancel is already pending or session is about to be closed/removed } } func (self *SessionChan) AddProgressHandler(userdata interface{}, cb ProgressCB) error { resCh := make(chan sessionAddProgressHandlerResponse) req := sessionAddProgressHandlerRequest{} req.userdata = userdata req.callback = cb req.response = resCh select { case self.addProgressChan <- req: default: return fmt.Errorf("session is about to be closed/removed") } res := <-resCh return res.err } func (self *SessionChan) AddDoneHandler(userdata interface{}, cb DoneCB) error { resCh := make(chan sessionAddDoneHandlerResponse) req := sessionAddDoneHandlerRequest{} req.userdata = userdata req.callback = cb req.response = resCh select { case self.addDoneChan <- req: default: return fmt.Errorf("session is about to be closed/removed") } res := <-resCh return res.err } func (self *SessionChan) AttachUploader() (<-chan bool, chan<- AttachmentChunk) { resCh := make(chan attachUploaderResponse) req := attachUploaderRequest{} req.response = resCh select { case self.attachUploaderChan <- req: default: // session is about to be closed/removed return nil, nil } res := <-resCh return res.cancel, res.attachment } // ********************************************************* // Semi-Public Interface (only used by sessionStore) func (self *Session) getInterface() *SessionChan { ch := &SessionChan{} ch.runChan = self.runChan ch.cancelChan = self.cancelChan ch.addProgressChan = self.addProgressChan ch.addDoneChan = self.addDoneChan ch.attachUploaderChan = self.attachUploaderChan return ch } func (self *Session) cleanup() { self.quit <- true self.ctx.dbglog.Printf("Session: waiting for session to close") <-self.done close(self.quit) close(self.done) self.timer.Stop() // don't close the channels we give out because this might lead to a panic if // somebody wites to an already removed session // close(self.cancelIntChan) // close(self.progressIntChan) // close(self.doneIntChan) // close(self.runChan) // close(self.cancelChan) // close(self.addProgressChan) // close(self.addDoneChan) // close(self.attachUploader) self.ctx.dbglog.Printf("Session: cleanup is now done") } func newSession(ctx *Context, removeFunc func()) (session *Session) { session = new(Session) session.state = SESSION_NEW session.removeFunc = removeFunc session.ctx = *ctx session.quit = make(chan bool, 1) session.done = make(chan bool) session.timer = time.NewTimer(10 * time.Second) session.cancelIntChan = make(chan bool, 1) session.progressRateLimit = 100 * time.Millisecond // TODO: hardcoded value session.progressIntChan = make(chan ProgressData, 10) session.doneIntChan = make(chan Result, 1) session.runChan = make(chan time.Duration, 1) session.cancelChan = make(chan bool, 1) session.addProgressChan = make(chan sessionAddProgressHandlerRequest, 10) session.addDoneChan = make(chan sessionAddDoneHandlerRequest, 10) session.attachUploaderChan = make(chan attachUploaderRequest, 1) go session.dispatchRequests() return }