zydronium/manage.get.gov
1
0
Fork 0
mirror of https://github.com/cisagov/manage.get.gov.git synced 2025-08-11 20:19:38 +02:00
Code Issues Projects Releases Packages Wiki Activity

Add settings for pooling, keep_alive

Browse source
This commit is contained in:
zandercymatics 2023-10-12 10:16:22 -06:00
parent a02d27aaec
commit 3a28a3a362
No known key found for this signature in database
GPG key ID: FF4636ABEC9682B7
3 changed files with 28 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

20
src/epplibwrapper/utility/pool.py
View file

@ -1,9 +1,13 @@
import logging
from geventconnpool import ConnectionPool
from epplibwrapper import RegistryError
from epplibwrapper.errors import LoginError
from epplibwrapper.errors import RegistryError, LoginError
from epplibwrapper.socket import Socket
try:
from epplib.commands import Hello
except ImportError:
pass
logger = logging.getLogger(__name__)
class EppConnectionPool(ConnectionPool):
@ -20,11 +24,17 @@ class EppConnectionPool(ConnectionPool):
return connection
except LoginError as err:
message = "_new_connection failed to execute due to a registry login error."
logger.warning(message, exc_info=True)
logger.error(message, exc_info=True)
raise RegistryError(message) from err
def _keepalive(self, connection):
pass
def _keepalive(self, c):
"""Sends a command to the server to keep the connection alive."""
try:
# Sends a ping to EPPLib
c.send(Hello())
except Exception as err:
logger.error("Failed to keep the connection alive.", exc_info=True)
raise RegistryError("Failed to keep the connection alive.") from err
def create_socket(self, client, login) -> Socket:
"""Creates and returns a socket instance"""