Added docstrings for communicators module.
This commit is contained in:
parent
1a486833bb
commit
5d2bed0f14
@ -11,12 +11,44 @@ from redis_super_storage import RedisSuperStorage
|
||||
|
||||
|
||||
class ProducerCommunicator:
|
||||
"""
|
||||
This class is used to communicate with producers.
|
||||
The addresses of producers are fetched from `RedisSuperStorage`.
|
||||
"""
|
||||
|
||||
def __init__(self, redis_store: RedisSuperStorage):
|
||||
"""
|
||||
Upon creating this object. A requests session is created on order to take advantage of keep-alive connections.
|
||||
|
||||
:param redis_store: A `RedisSuperStorage` instance.
|
||||
"""
|
||||
self._redis_store = redis_store
|
||||
self._session = requests.Session()
|
||||
|
||||
def push_ip_update(self, newip: str):
|
||||
"""
|
||||
This method is used to push an ip update to all known consumers.
|
||||
The list of consumers are read from the `RedisSuperStorage` instance.
|
||||
(The list of producers are maintained by the api endpoint.)
|
||||
|
||||
The uuid of this consumer is acquired directly from the `LOCAL_UUID` envvar.
|
||||
|
||||
A timeout of 5 seconds is hardcoded for each producer individually. Timeout is logged as warning.
|
||||
|
||||
Called URL::
|
||||
|
||||
http://<producer ip>/ip
|
||||
|
||||
Body::
|
||||
|
||||
{
|
||||
"uuid" : "str: LOCAL_UUID",
|
||||
"ip": "str: provided by param"
|
||||
}
|
||||
|
||||
|
||||
:param newip: New ipaddress to be annouced.
|
||||
"""
|
||||
|
||||
for key, ip in self._redis_store.get_producer_list().items():
|
||||
|
||||
@ -32,12 +64,44 @@ class ProducerCommunicator:
|
||||
|
||||
|
||||
class ConsumerCommunicator:
|
||||
"""
|
||||
This class is used to communicate with consumers.
|
||||
The addresses of consumers are fetched from the `RedisSuperStorage`.
|
||||
"""
|
||||
|
||||
def __init__(self, redis_store: RedisSuperStorage):
|
||||
"""
|
||||
Upon creating this object. A requests session is created on order to take advantage of keep-alive connections.
|
||||
|
||||
:param redis_store: A `RedisSuperStorage` instance.
|
||||
"""
|
||||
self._redis_store = redis_store
|
||||
self._session = requests.Session()
|
||||
|
||||
def targeted_snyc(self, ip: str):
|
||||
"""
|
||||
This method works similarly to `sync_all` however the target is not fetched from the `RedisSuperStorage` instance.
|
||||
The results are processed the same way (saved to redis).
|
||||
|
||||
A timeout of 5 seconds is hardcoded for this function. Timeout is logged as warning.
|
||||
|
||||
This method is preferred when the remote is unknown (uuid is unknown). Mostly when the application just started up,
|
||||
and an initial syncrhronization to all consumers is required.
|
||||
|
||||
See `sync_all` for more information.
|
||||
|
||||
Called URL::
|
||||
|
||||
http://<consumer ip>/sync
|
||||
|
||||
Body::
|
||||
|
||||
{
|
||||
"uuid" : "str: LOCAL_UUID"
|
||||
}
|
||||
|
||||
:param ip: The ip address of the consumer to be synced to.
|
||||
"""
|
||||
try:
|
||||
# request synchronization
|
||||
response = self._session.post(f"http://{ip}/sync", json={'uuid': os.environ['LOCAL_UUID']}, timeout=5)
|
||||
@ -50,6 +114,28 @@ class ConsumerCommunicator:
|
||||
self._redis_store.update_consumer(response.json()['uuid'], ip)
|
||||
|
||||
def sync_all(self):
|
||||
"""
|
||||
This method is used to syncronize with each known consumer.
|
||||
The list of consumers are acquired from the `RedisSuperStorage` instance.
|
||||
|
||||
This syncrhonization run causes the followings:
|
||||
* Each consumer known by this consumer is checked for availability
|
||||
* Each consumer this consumer communicated with updated the availability of this consumer.
|
||||
* Each consumer which had no information of this consumer, now have.
|
||||
|
||||
A timeout of 5 seconds is hardcoded for each consumer individually. Timeout is logged as warning.
|
||||
|
||||
Called URL::
|
||||
|
||||
http://<consumer ip>/sync
|
||||
|
||||
Body::
|
||||
|
||||
{
|
||||
"uuid" : "str: LOCAL_UUID"
|
||||
}
|
||||
|
||||
"""
|
||||
for uuid, info in self._redis_store.get_consumer_list().items():
|
||||
ip = info['ip']
|
||||
try:
|
||||
|
Reference in New Issue
Block a user