-
Notifications
You must be signed in to change notification settings - Fork 11
/
ns.py
137 lines (110 loc) · 4.26 KB
/
ns.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
#!/usr/bin/env python
# -*- coding: utf-8 -*-
#
# Copyright (c) 2011, 2012, 2014, 2015, 2021 Pytroll Community
#
# Author(s):
#
# Martin Raspaud <martin.raspaud@smhi.se>
# Panu Lahtinen <panu.lahtinen@fmi.fi"
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
"""Manage other's subscriptions.
Default port is 5557, if $POSTTROLL_NAMESERVER_PORT is not defined.
"""
import datetime as dt
import logging
import os
import time
import warnings
from posttroll import config
from posttroll.address_receiver import AddressReceiver
from posttroll.message import Message
# pylint: enable=E0611
DEFAULT_NAMESERVER_PORT = 5557
logger = logging.getLogger(__name__)
def get_configured_nameserver_port():
"""Get the configured nameserver port."""
try:
port = int(os.environ["NAMESERVER_PORT"])
warnings.warn("NAMESERVER_PORT is pending deprecation, please use POSTTROLL_NAMESERVER_PORT instead.",
PendingDeprecationWarning)
except KeyError:
port = DEFAULT_NAMESERVER_PORT
return config.get("nameserver_port", port)
# Client functions.
def get_pub_addresses(names=None, timeout=10, nameserver="localhost"):
"""Get the addresses of the publishers.
Kwargs:
- names: names of the publishers
- nameserver: nameserver address to query the publishers from (default: localhost).
"""
addrs = []
if names is None:
names = ["", ]
for name in names:
then = dt.datetime.now() + dt.timedelta(seconds=timeout)
while dt.datetime.now() < then:
addrs += get_pub_address(name, nameserver=nameserver, timeout=timeout)
if addrs:
break
time.sleep(timeout / 20.0)
return addrs
def get_pub_address(name, timeout=10, nameserver="localhost"):
"""Get the address of the named publisher.
Args:
name: name of the publishers
timeout: how long to wait for an address, in seconds.
nameserver: nameserver address to query the publishers from (default: localhost).
"""
if config["backend"] not in ["unsecure_zmq", "secure_zmq"]:
raise NotImplementedError(f"Did not recognize backend: {config['backend']}")
from posttroll.backends.zmq.ns import zmq_get_pub_address
return zmq_get_pub_address(name, timeout, nameserver)
# Server part.
def get_active_address(name, arec):
"""Get the addresses of the active modules for a given publisher *name*."""
addrs = arec.get(name)
if addrs:
return Message("/oper/ns", "info", addrs)
else:
return Message("/oper/ns", "info", "")
class NameServer:
"""The name server."""
def __init__(self, max_age=None, multicast_enabled=True, restrict_to_localhost=False):
"""Initialize nameserver."""
self.loop = True
self.listener = None
self._max_age = max_age or dt.timedelta(minutes=10)
self._multicast_enabled = multicast_enabled
self._restrict_to_localhost = restrict_to_localhost
backend = config["backend"]
if backend not in ["unsecure_zmq", "secure_zmq"]:
raise NotImplementedError(f"Did not recognize backend: {backend}")
from posttroll.backends.zmq.ns import ZMQNameServer
self._ns = ZMQNameServer()
def run(self, *args):
"""Run the listener and answer to requests."""
del args
arec = AddressReceiver(max_age=self._max_age,
multicast_enabled=self._multicast_enabled,
restrict_to_localhost=self._restrict_to_localhost)
arec.start()
try:
return self._ns.run(arec)
finally:
arec.stop()
def stop(self):
"""Stop the nameserver."""
return self._ns.stop()