use global semaphore in push_to_actor to prevent overloading system with outbound connections
This commit is contained in:
parent
f34b42ae78
commit
94e669cf7d
|
@ -30,6 +30,7 @@ if "actorKeys" not in DATABASE:
|
||||||
PRIVKEY = RSA.importKey(DATABASE["actorKeys"]["privateKey"])
|
PRIVKEY = RSA.importKey(DATABASE["actorKeys"]["privateKey"])
|
||||||
PUBKEY = PRIVKEY.publickey()
|
PUBKEY = PRIVKEY.publickey()
|
||||||
|
|
||||||
|
sem = asyncio.Semaphore(500)
|
||||||
|
|
||||||
from . import app, CONFIG
|
from . import app, CONFIG
|
||||||
from .remote_actor import fetch_actor
|
from .remote_actor import fetch_actor
|
||||||
|
@ -77,7 +78,6 @@ get_actor_inbox = lambda actor: actor.get('endpoints', {}).get('sharedInbox', ac
|
||||||
|
|
||||||
async def push_message_to_actor(actor, message, our_key_id):
|
async def push_message_to_actor(actor, message, our_key_id):
|
||||||
inbox = get_actor_inbox(actor)
|
inbox = get_actor_inbox(actor)
|
||||||
|
|
||||||
url = urlsplit(inbox)
|
url = urlsplit(inbox)
|
||||||
|
|
||||||
# XXX: Digest
|
# XXX: Digest
|
||||||
|
@ -93,15 +93,17 @@ async def push_message_to_actor(actor, message, our_key_id):
|
||||||
|
|
||||||
logging.debug('%r >> %r', inbox, message)
|
logging.debug('%r >> %r', inbox, message)
|
||||||
|
|
||||||
try:
|
global sem
|
||||||
async with aiohttp.ClientSession(trace_configs=[http_debug()]) as session:
|
async with sem:
|
||||||
async with session.post(inbox, data=data, headers=headers) as resp:
|
try:
|
||||||
if resp.status == 202:
|
async with aiohttp.ClientSession(trace_configs=[http_debug()]) as session:
|
||||||
return
|
async with session.post(inbox, data=data, headers=headers) as resp:
|
||||||
resp_payload = await resp.text()
|
if resp.status == 202:
|
||||||
logging.debug('%r >> resp %r', inbox, resp_payload)
|
return
|
||||||
except Exception as e:
|
resp_payload = await resp.text()
|
||||||
logging.info('Caught %r while pushing to %r.', e, inbox)
|
logging.debug('%r >> resp %r', inbox, resp_payload)
|
||||||
|
except Exception as e:
|
||||||
|
logging.info('Caught %r while pushing to %r.', e, inbox)
|
||||||
|
|
||||||
|
|
||||||
async def follow_remote_actor(actor_uri):
|
async def follow_remote_actor(actor_uri):
|
||||||
|
@ -308,5 +310,4 @@ async def inbox(request):
|
||||||
|
|
||||||
return aiohttp.web.Response(body=b'{}', content_type='application/activity+json')
|
return aiohttp.web.Response(body=b'{}', content_type='application/activity+json')
|
||||||
|
|
||||||
|
|
||||||
app.router.add_post('/inbox', inbox)
|
app.router.add_post('/inbox', inbox)
|
||||||
|
|
Loading…
Reference in a new issue