diff --git a/code/main.py b/code/main.py index 51f352f..5b84253 100644 --- a/code/main.py +++ b/code/main.py @@ -1,8 +1,10 @@ import time import logging import os +import json from configparser import ConfigParser from pathlib import Path +from queue import Queue from initialise_app import initialise_app from bandwidth import measure as measure_bandwidth from database import connect @@ -29,29 +31,50 @@ def event( # load environment variables replicates = int(os.getenv('REPLICATES', default=REPLICATES)) user_id = int(os.getenv('USER_ID', default=USER_ID)) - # setup database connection - db = connect() + # setup data queue + data_queue: Queue = Queue() # run event for rep_num in range(replicates): - logging.info(f'running replicate {rep_num}') + logging.debug(f'running replicate {rep_num+1} of {replicates}') # do measurement try: - res = measure_bandwidth() + data = measure_bandwidth() except Exception as e: logging.error(f'failed to measure bandwidth: {e}') continue - # upload result to database + # add data to queue + data_queue.put(data) + # polite pause + time.sleep(1) + # check if queue should be flushed + queue_size = data_queue.qsize() + if queue_size == 0: + logging.debug('queue empty, continuing') + return + # flush queue to database + db = connect() + for i in range(queue_size): + # get data + data = data_queue.get() + # prepare payload payload_dict = { 'user_id': user_id, - 'data': res + 'data': data } + # send to database try: db_id = db.insert_one(payload_dict).inserted_id except Exception as e: - logging.error(f'failed sending results to database: {e}') + logging.error('failed sending data to database') + logging.info( + 'failed pushing to database:\n' + f'{json.dumps(payload_dict, indent=4)}\n' + f'with error: {e}' + ) + data_queue.put(data) # put data back in queue continue - logging.debug(f'data sent to database received db_id: {db_id}') - time.sleep(1) + else: + logging.debug(f'data sent to database received id: {db_id}') logging.debug('finished') @@ -60,6 +83,7 @@ if __name__ == '__main__': trigger_time = time.time() while True: if time.time() >= trigger_time: + logging.debug('triggered event') # set new trigger time trigger_interval = float( os.getenv( @@ -75,4 +99,5 @@ if __name__ == '__main__': logging.error(e) else: logging.info('finished event') + # polite pause time.sleep(1)