Exit on critical errors to allow process to restart.

poller.go

@@ -57,6 +57,8 @@ func (p *poller) poll(interval time.Duration, quit <-chan bool) <-chan *job {
 	conn, err := GetConn()
 	if err != nil {
 		logger.Criticalf("Error on getting connection in poller %s", p)
+		close(jobs)
+		return jobs
 	} else {
 		p.open(conn)
 		p.start(conn)
@@ -70,6 +72,7 @@ func (p *poller) poll(interval time.Duration, quit <-chan bool) <-chan *job {
 			conn, err := GetConn()
 			if err != nil {
 				logger.Criticalf("Error on getting connection in poller %s", p)
+				return
 			} else {
 				p.finish(conn)
 				p.close(conn)
@@ -85,11 +88,14 @@ func (p *poller) poll(interval time.Duration, quit <-chan bool) <-chan *job {
 				conn, err := GetConn()
 				if err != nil {
 					logger.Criticalf("Error on getting connection in poller %s", p)
+					return
 				}
 
 				job, err := p.getJob(conn)
 				if err != nil {
-					logger.Errorf("Error on %v getting job from %v: %v", p, p.Queues, err)
+					logger.Criticalf("Error on %v getting job from %v: %v", p, p.Queues, err)
+					PutConn(conn)
+					return
 				}
 				if job != nil {
 					conn.Send("INCR", fmt.Sprintf("%sstat:processed:%v", namespace, p))
@@ -106,6 +112,7 @@ func (p *poller) poll(interval time.Duration, quit <-chan bool) <-chan *job {
 						conn, err := GetConn()
 						if err != nil {
 							logger.Criticalf("Error on getting connection in poller %s", p)
+							return
 						}
 
 						conn.Send("LPUSH", fmt.Sprintf("%squeue:%s", namespace, job.Queue), buf)

worker.go

@@ -82,6 +82,7 @@ func (w *worker) work(jobs <-chan *job, monitor *sync.WaitGroup) {
 	conn, err := GetConn()
 	if err != nil {
 		logger.Criticalf("Error on getting connection in worker %v", w)
+		return
 	} else {
 		w.open(conn)
 		PutConn(conn)
@@ -91,15 +92,16 @@ func (w *worker) work(jobs <-chan *job, monitor *sync.WaitGroup) {
 
 	go func() {
 		defer func() {
+			defer monitor.Done()
+
 			conn, err := GetConn()
 			if err != nil {
 				logger.Criticalf("Error on getting connection in worker %v", w)
+				return
 			} else {
 				w.close(conn)
 				PutConn(conn)
 			}
-
-			monitor.Done()
 		}()
 		for job := range jobs {
 			if workerFunc, ok := workers[job.Payload.Class]; ok {
@@ -113,6 +115,7 @@ func (w *worker) work(jobs <-chan *job, monitor *sync.WaitGroup) {
 				conn, err := GetConn()
 				if err != nil {
 					logger.Criticalf("Error on getting connection in worker %v", w)
+					return
 				} else {
 					w.finish(conn, job, errors.New(errorLog))
 					PutConn(conn)
@@ -128,6 +131,7 @@ func (w *worker) run(job *job, workerFunc workerFunc) {
 		conn, errCon := GetConn()
 		if errCon != nil {
 			logger.Criticalf("Error on getting connection in worker %v", w)
+			return
 		} else {
 			w.finish(conn, job, err)
 			PutConn(conn)
@@ -142,6 +146,7 @@ func (w *worker) run(job *job, workerFunc workerFunc) {
 	conn, err := GetConn()
 	if err != nil {
 		logger.Criticalf("Error on getting connection in worker %v", w)
+		return
 	} else {
 		w.start(conn, job)
 		PutConn(conn)