backup/warcprox
1
0
Fork 0
mirror of https://github.com/internetarchive/warcprox.git synced 2025-01-18 13:22:09 +01:00
Code Issues Projects Releases Activity

shut down postfetch processors

Browse Source
This commit is contained in:
Noah Levitt 2018-01-15 15:37:26 -08:00
parent c9a39958db
commit b7d176be28
1 changed files with 9 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

10
warcprox/__init__.py
View File

@ -136,6 +136,7 @@ class BasePostfetchProcessor(threading.Thread):
except queue.Empty: except queue.Empty:
if self.stop.is_set(): if self.stop.is_set():
break break
logging.info('%s shutting down', self)
self._shutdown() self._shutdown()
except Exception as e: except Exception as e:
if isinstance(e, OSError) and e.errno == 28: if isinstance(e, OSError) and e.errno == 28:
@ -190,7 +191,6 @@ class ListenerPostfetchProcessor(BaseStandardPostfetchProcessor):
BaseStandardPostfetchProcessor.__init__(self, inq, outq, profile) BaseStandardPostfetchProcessor.__init__(self, inq, outq, profile)
self.listener = listener self.listener = listener
self.name = listener.__class__.__name__ self.name = listener.__class__.__name__
logging.info('self.name=%s', self.name)
def _process_url(self, recorded_url): def _process_url(self, recorded_url):
return self.listener.notify(recorded_url, recorded_url.warc_records) return self.listener.notify(recorded_url, recorded_url.warc_records)
@ -200,6 +200,14 @@ class ListenerPostfetchProcessor(BaseStandardPostfetchProcessor):
self.listener.start() self.listener.start()
BaseStandardPostfetchProcessor.start(self) BaseStandardPostfetchProcessor.start(self)
def _shutdown(self):
if hasattr(self.listener, 'stop'):
try:
self.listener.stop()
except:
self.logger.error(
'%s raised exception', listener.stop, exc_info=True)
# monkey-patch log levels TRACE and NOTICE # monkey-patch log levels TRACE and NOTICE
TRACE = 5 TRACE = 5
def _logger_trace(self, msg, *args, **kwargs): def _logger_trace(self, msg, *args, **kwargs):