Background task for broadcasting title/uptime changes
このコミットが含まれているのは:
コミット
a3b18bdc9f
|
@ -43,6 +43,7 @@ def create_app(config_file):
|
||||||
'TASK_PERIOD_ROTATE_USERS': config['tasks']['rotate_users'],
|
'TASK_PERIOD_ROTATE_USERS': config['tasks']['rotate_users'],
|
||||||
'TASK_PERIOD_ROTATE_CAPTCHAS': config['tasks']['rotate_captchas'],
|
'TASK_PERIOD_ROTATE_CAPTCHAS': config['tasks']['rotate_captchas'],
|
||||||
'TASK_PERIOD_BROADCAST_USERS_UPDATE': config['tasks']['broadcast_users_update'],
|
'TASK_PERIOD_BROADCAST_USERS_UPDATE': config['tasks']['broadcast_users_update'],
|
||||||
|
'TASK_PERIOD_BROADCAST_STREAM_INFO_UPDATE': config['tasks']['broadcast_stream_info_update'],
|
||||||
'THRESHOLD_USER_NOTWATCHING': config['thresholds']['user_notwatching'],
|
'THRESHOLD_USER_NOTWATCHING': config['thresholds']['user_notwatching'],
|
||||||
'THRESHOLD_USER_TENTATIVE': config['thresholds']['user_tentative'],
|
'THRESHOLD_USER_TENTATIVE': config['thresholds']['user_tentative'],
|
||||||
'THRESHOLD_USER_ABSENT': config['thresholds']['user_absent'],
|
'THRESHOLD_USER_ABSENT': config['thresholds']['user_absent'],
|
||||||
|
@ -82,8 +83,12 @@ def create_app(config_file):
|
||||||
app.captcha_factory = create_captcha_factory(app.config['CAPTCHA_FONTS'])
|
app.captcha_factory = create_captcha_factory(app.config['CAPTCHA_FONTS'])
|
||||||
app.captcha_signer = create_captcha_signer(app.config['SECRET_KEY'])
|
app.captcha_signer = create_captcha_signer(app.config['SECRET_KEY'])
|
||||||
|
|
||||||
|
# State for tasks
|
||||||
app.users_update_buffer = set()
|
app.users_update_buffer = set()
|
||||||
|
app.stream_uptime = None
|
||||||
|
app.stream_title = None
|
||||||
|
|
||||||
|
# Background tasks' asyncio.sleep tasks, cancelled on shutdown
|
||||||
app.background_sleep = set()
|
app.background_sleep = set()
|
||||||
|
|
||||||
@app.after_serving
|
@app.after_serving
|
||||||
|
|
|
@ -399,9 +399,15 @@ const on_websocket_message = (event) => {
|
||||||
|
|
||||||
break;
|
break;
|
||||||
|
|
||||||
case "title":
|
case "info":
|
||||||
console.log("ws title", receipt);
|
console.log("ws info", receipt);
|
||||||
set_title(receipt.title);
|
if (receipt.title !== undefined) {
|
||||||
|
set_title(receipt.title);
|
||||||
|
}
|
||||||
|
if (receipt.uptime !== undefined) {
|
||||||
|
set_frozen_uptime(receipt.uptime);
|
||||||
|
update_uptime();
|
||||||
|
}
|
||||||
break;
|
break;
|
||||||
|
|
||||||
case "ack":
|
case "ack":
|
||||||
|
|
|
@ -1,9 +1,11 @@
|
||||||
import asyncio
|
import asyncio
|
||||||
|
import itertools
|
||||||
from functools import wraps
|
from functools import wraps
|
||||||
|
|
||||||
from quart import current_app
|
from quart import current_app
|
||||||
|
|
||||||
from anonstream.broadcast import broadcast, broadcast_users_update
|
from anonstream.broadcast import broadcast, broadcast_users_update
|
||||||
|
from anonstream.stream import is_online, get_stream_title, get_stream_uptime
|
||||||
from anonstream.wrappers import with_timestamp
|
from anonstream.wrappers import with_timestamp
|
||||||
from anonstream.helpers.user import is_visible
|
from anonstream.helpers.user import is_visible
|
||||||
|
|
||||||
|
@ -27,12 +29,12 @@ def with_period(period):
|
||||||
def periodically(f):
|
def periodically(f):
|
||||||
@wraps(f)
|
@wraps(f)
|
||||||
async def wrapper(*args, **kwargs):
|
async def wrapper(*args, **kwargs):
|
||||||
while True:
|
for iteration in itertools.count():
|
||||||
|
await f(iteration, *args, **kwargs)
|
||||||
try:
|
try:
|
||||||
await sleep_and_collect_task(period)
|
await sleep_and_collect_task(period)
|
||||||
except asyncio.CancelledError:
|
except asyncio.CancelledError:
|
||||||
break
|
break
|
||||||
f(*args, **kwargs)
|
|
||||||
|
|
||||||
return wrapper
|
return wrapper
|
||||||
|
|
||||||
|
@ -40,7 +42,10 @@ def with_period(period):
|
||||||
|
|
||||||
@with_period(CONFIG['TASK_PERIOD_ROTATE_USERS'])
|
@with_period(CONFIG['TASK_PERIOD_ROTATE_USERS'])
|
||||||
@with_timestamp
|
@with_timestamp
|
||||||
def t_sunset_users(timestamp):
|
async def t_sunset_users(iteration, timestamp):
|
||||||
|
if iteration == 0:
|
||||||
|
return
|
||||||
|
|
||||||
tokens = []
|
tokens = []
|
||||||
for token in USERS_BY_TOKEN:
|
for token in USERS_BY_TOKEN:
|
||||||
user = USERS_BY_TOKEN[token]
|
user = USERS_BY_TOKEN[token]
|
||||||
|
@ -67,7 +72,10 @@ def t_sunset_users(timestamp):
|
||||||
)
|
)
|
||||||
|
|
||||||
@with_period(CONFIG['TASK_PERIOD_ROTATE_CAPTCHAS'])
|
@with_period(CONFIG['TASK_PERIOD_ROTATE_CAPTCHAS'])
|
||||||
def t_expire_captchas():
|
async def t_expire_captchas(iteration):
|
||||||
|
if iteration == 0:
|
||||||
|
return
|
||||||
|
|
||||||
to_delete = []
|
to_delete = []
|
||||||
for digest in CAPTCHAS:
|
for digest in CAPTCHAS:
|
||||||
valid = CAPTCHA_SIGNER.validate(
|
valid = CAPTCHA_SIGNER.validate(
|
||||||
|
@ -76,13 +84,52 @@ def t_expire_captchas():
|
||||||
)
|
)
|
||||||
if not valid:
|
if not valid:
|
||||||
to_delete.append(digest)
|
to_delete.append(digest)
|
||||||
|
|
||||||
for digest in to_delete:
|
for digest in to_delete:
|
||||||
CAPTCHAS.pop(digest)
|
CAPTCHAS.pop(digest)
|
||||||
|
|
||||||
@with_period(CONFIG['TASK_PERIOD_BROADCAST_USERS_UPDATE'])
|
@with_period(CONFIG['TASK_PERIOD_BROADCAST_USERS_UPDATE'])
|
||||||
def t_broadcast_users_update():
|
async def t_broadcast_users_update(iteration):
|
||||||
broadcast_users_update()
|
if iteration == 0:
|
||||||
|
return
|
||||||
|
else:
|
||||||
|
broadcast_users_update()
|
||||||
|
|
||||||
|
@with_period(CONFIG['TASK_PERIOD_BROADCAST_STREAM_INFO_UPDATE'])
|
||||||
|
async def t_broadcast_stream_info_update(iteration):
|
||||||
|
if iteration == 0:
|
||||||
|
current_app.stream_title = await get_stream_title()
|
||||||
|
current_app.stream_uptime = get_stream_uptime()
|
||||||
|
else:
|
||||||
|
payload = {}
|
||||||
|
|
||||||
|
# Check if the stream title has changed
|
||||||
|
title = await get_stream_title()
|
||||||
|
if current_app.stream_title != title:
|
||||||
|
current_app.stream_title = title
|
||||||
|
payload['title'] = title
|
||||||
|
|
||||||
|
# Check if the stream uptime has changed differently than expected
|
||||||
|
if current_app.stream_uptime is None:
|
||||||
|
expected_uptime = None
|
||||||
|
else:
|
||||||
|
expected_uptime = (
|
||||||
|
current_app.stream_uptime
|
||||||
|
+ CONFIG['TASK_PERIOD_BROADCAST_STREAM_INFO_UPDATE']
|
||||||
|
)
|
||||||
|
uptime = get_stream_uptime()
|
||||||
|
current_app.stream_uptime = uptime
|
||||||
|
if uptime is None and expected_uptime is None:
|
||||||
|
pass
|
||||||
|
elif uptime is None or expected_uptime is None:
|
||||||
|
payload['uptime'] = uptime
|
||||||
|
elif abs(uptime - expected_uptime) >= 0.0625:
|
||||||
|
payload['uptime'] = uptime
|
||||||
|
|
||||||
|
if payload:
|
||||||
|
broadcast(USERS, payload={'type': 'info', **payload})
|
||||||
|
|
||||||
current_app.add_background_task(t_sunset_users)
|
current_app.add_background_task(t_sunset_users)
|
||||||
current_app.add_background_task(t_expire_captchas)
|
current_app.add_background_task(t_expire_captchas)
|
||||||
current_app.add_background_task(t_broadcast_users_update)
|
current_app.add_background_task(t_broadcast_users_update)
|
||||||
|
current_app.add_background_task(t_broadcast_stream_info_update)
|
||||||
|
|
|
@ -34,6 +34,7 @@ chat_scrollback = 256
|
||||||
rotate_users = 60.0
|
rotate_users = 60.0
|
||||||
rotate_captchas = 60.0
|
rotate_captchas = 60.0
|
||||||
broadcast_users_update = 4.0
|
broadcast_users_update = 4.0
|
||||||
|
broadcast_stream_info_update = 3.0
|
||||||
|
|
||||||
[names]
|
[names]
|
||||||
broadcaster = "Broadcaster"
|
broadcaster = "Broadcaster"
|
||||||
|
|
読み込み中…
新しいイシューから参照