tanner/qotnews
1
0
Fork 1
Code Issues Pull Requests Releases Wiki Activity

adjust feed thread.

Browse Source
This commit is contained in:
Jason Schwarzenberger 2020-11-25 12:34:46 +13:00
parent 5c96092a57
commit 247715a76e
1 changed files with 37 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
apiserver/server.py
View File

@ -13,6 +13,7 @@ import json
import threading import threading
import traceback import traceback
import time import time
from datetime import datetime, timedelta
from urllib.parse import urlparse, parse_qs from urllib.parse import urlparse, parse_qs
import settings import settings
@ -154,10 +155,10 @@ def _add_new_refs():
try: try:
nid = new_id() nid = new_id()
database.put_ref(ref, nid, source, urlref) database.put_ref(ref, nid, source, urlref)
added.append(ref)
logging.info('Added ref ' + ref) logging.info('Added ref ' + ref)
added.append(ref)
except database.IntegrityError: except database.IntegrityError:
logging.info('Unable to add ref ' + ref) #logging.info('Unable to add ref ' + ref)
continue continue
return added return added
@ -180,34 +181,60 @@ def _update_current_story(item):
database.del_ref(item['ref']) database.del_ref(item['ref'])
logging.info('Removed ref {}'.format(item['ref'])) logging.info('Removed ref {}'.format(item['ref']))
def feed_thread(): def feed_thread():
ref_list = [] new_refs = []
update_refs = []
last_check = datetime.now() - timedelta(minutes=20)
try: try:
while True: while True:
# onboard new stories # onboard new stories
if not len(ref_list): time_since_check = datetime.now() - last_check
if not len(new_refs) and time_since_check > timedelta(minutes=15):
added = _add_new_refs() added = _add_new_refs()
ref_list = database.get_reflist() ref_list = database.get_reflist()
ref_list.sort(key=lambda i: i['ref'] in added, reverse=True) new_refs = list(filter(None, [i if i['ref'] in added else None for i in ref_list]))
update_queue = list(filter(None, [i if i['ref'] not in added else None for i in ref_list]))
current_queue_refs = [i['ref'] for i in update_refs]
update_queue = list(filter(None, [i if i['ref'] not in current_queue_refs else None for i in update_queue]))
update_refs += update_queue
logging.info('Added {} new refs'.format(len(added)))
logging.info('Have {} refs in update queue'.format(len(current_queue_refs)))
logging.info('Fetched {} refs for update queue'.format(len(update_queue)))
last_check = datetime.now()
gevent.sleep(1)
# update new stories
if len(new_refs):
item = new_refs.pop(0)
logging.info('Processing new story ref {}'.format(item['ref']))
_update_current_story(item)
gevent.sleep(1)
# update current stories # update current stories
if len(ref_list): if len(update_refs):
item = ref_list.pop(0) item = update_refs.pop(0)
logging.info('Processing existing story ref {}'.format(item['ref']))
_update_current_story(item) _update_current_story(item)
gevent.sleep(1)
gevent.sleep(6) gevent.sleep(1)
except KeyboardInterrupt: except KeyboardInterrupt:
logging.info('Ending feed thread...') logging.info('Ending feed thread...')
except ValueError as e: except ValueError as e:
logging.error('feed_thread error: {} {}'.format(e.__class__.__name__, e)) logging.error('feed_thread error: {} {}'.format(e.__class__.__name__, e))
http_server.stop() http_server.stop()
gevent.kill(feed_thread_ref)
print('Starting Feed thread...') print('Starting Feed thread...')
gevent.spawn(feed_thread) feed_thread_ref = gevent.spawn(feed_thread)
print('Starting HTTP thread...') print('Starting HTTP thread...')
try: try:
http_server.serve_forever() http_server.serve_forever()
except KeyboardInterrupt: except KeyboardInterrupt:
gevent.kill(feed_thread_ref)
logging.info('Exiting...') logging.info('Exiting...')