src/nepi/resources/omf/omf6_api.py

   1 #
   2 #    NEPI, a framework to manage network experiments
   3 #    Copyright (C) 2013 INRIA
   4 #
   5 #    This program is free software: you can redistribute it and/or modify
   6 #    it under the terms of the GNU General Public License as published by
   7 #    the Free Software Foundation, either version 3 of the License, or
   8 #    (at your option) any later version.
   9 #
  10 #    This program is distributed in the hope that it will be useful,
  11 #    but WITHOUT ANY WARRANTY; without even the implied warranty of
  12 #    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  13 #    GNU General Public License for more details.
  14 #
  15 #    You should have received a copy of the GNU General Public License
  16 #    along with this program.  If not, see <http://www.gnu.org/licenses/>.
  17 #
  18 # Author: Alina Quereilhac <alina.quereilhac@inria.fr>
  19 #         Julien Tribino <julien.tribino@inria.fr>
  20
  21 import ssl
  22 import sys
  23 import time
  24 import hashlib
  25 import threading
  26
  27 from nepi.util.timefuncs import tsformat
  28 import os
  29
  30 from nepi.util.logger import Logger
  31
  32 from nepi.resources.omf.omf_client import OMFClient
  33 from nepi.resources.omf.messages_6 import MessageHandler
  34
  35 class OMF6API(Logger):
  36     """
  37     .. class:: Class Args :
  38
  39         :param slice: Xmpp Slice
  40         :type slice: str
  41         :param host: Xmpp Server
  42         :type host: str
  43         :param port: Xmpp Port
  44         :type port: str
  45         :param password: Xmpp password
  46         :type password: str
  47         :param xmpp_root: Root of the Xmpp Topic Architecture
  48         :type xmpp_root: str
  49
  50     .. note::
  51
  52        This class is the implementation of an OMF 5.4 API.
  53        Since the version 5.4.1, the Topic Architecture start with OMF_5.4
  54        instead of OMF used for OMF5.3
  55
  56     """
  57     def __init__(self, host, user = "nepi", port="5222", password="1234",
  58             exp_id = None):
  59         """
  60
  61         :param slice: Xmpp Slice
  62         :type slice: str
  63         :param host: Xmpp Server
  64         :type host: str
  65         :param port: Xmpp Port
  66         :type port: str
  67         :param password: Xmpp password
  68         :type password: str
  69         :param xmpp_root: Root of the Xmpp Topic Architecture
  70         :type xmpp_root: str
  71
  72         """
  73         super(OMF6API, self).__init__("OMF6API")
  74         self._exp_id = exp_id
  75         self._user = user # name of the machine that run Nepi
  76         self._host = host # name of the xmpp server
  77         self._port = port # port of the xmpp server
  78         self._password = password # password to connect to xmpp
  79         self._jid = "%s-%s@%s" % (self._user, self._exp_id, self._host)
  80         self._src = "xmpp://" + self._jid
  81
  82         self._topics = []
  83
  84         # OMF xmpp client
  85         self._client = None
  86
  87         # message handler
  88         self._message = None
  89
  90         if sys.version_info < (3, 0):
  91             reload(sys)
  92             sys.setdefaultencoding('utf8')
  93
  94         # instantiate the xmpp client
  95         self._init_client()
  96
  97         # register nepi topic
  98         self._enroll_nepi()
  99
 100
 101     def _init_client(self):
 102         """ Initialize XMPP Client
 103
 104         """
 105         xmpp = OMFClient(self._jid, self._password)
 106         # PROTOCOL_SSLv3 required for compatibility with OpenFire
 107         xmpp.ssl_version = ssl.PROTOCOL_SSLv3
 108
 109         if xmpp.connect((self._host, self._port)):
 110             xmpp.process(block=False)
 111             self.check_ready(xmpp)
 112             self._client = xmpp
 113             self._message = MessageHandler()
 114         else:
 115             msg = "Unable to connect to the XMPP server."
 116             self.error(msg)
 117             raise RuntimeError(msg)
 118
 119     def check_ready(self, xmpp):
 120         delay = 1.0
 121         for i in xrange(4):
 122             if xmpp.ready:
 123                 break
 124             else:
 125                 time.sleep(delay)
 126                 delay = delay * 1.5
 127         else:
 128             msg = "XMPP Client is not ready after long time"
 129             self.error(msg, out, err)
 130             raise RuntimeError, msg
 131
 132     @property
 133     def _nepi_topic(self):
 134         msg = "nepi-" + self._exp_id
 135         self.debug(msg)
 136         return msg
 137
 138     def _enroll_nepi(self):
 139         """ Create and Subscribe to the Session Topic
 140
 141         """
 142         nepi_topic = self._nepi_topic
 143         self._client.create(nepi_topic)
 144         self._client.subscribe(nepi_topic)
 145
 146
 147     def enroll_topic(self, topic):
 148         """ Create and Subscribe to the session topic and the resources
 149             corresponding to the hostname
 150
 151         :param hostname: Full hrn of the node
 152         :type hostname: str
 153
 154         """
 155         if topic in self._topics:
 156             return
 157
 158         self._topics.append(topic)
 159
 160 #        try :
 161         self._client.create(topic)
 162 #        except:
 163 #            msg = "Topic already existing"
 164 #            self.info(msg)
 165         self._client.subscribe(topic)
 166
 167     def frcp_inform(self, topic, cid, itype):
 168         """ Configure attribute on the node
 169
 170         """
 171         msg_id = os.urandom(16).encode('hex')
 172         timestamp = tsformat()
 173         payload = self._message.inform_function(msg_id, self._src, timestamp, props = props ,guards = guards)
 174
 175         self._client.publish(payload, xmpp_node)
 176
 177     def frcp_configure(self, topic, props = None, guards = None ):
 178         """ Configure attribute on the node
 179
 180         """
 181         msg_id = os.urandom(16).encode('hex')
 182         timestamp = tsformat()
 183         payload = self._message.configure_function(msg_id, self._src, timestamp ,props = props ,guards = guards)
 184         self._client.publish(payload, topic)
 185
 186
 187     def frcp_create(self, topic, rtype, props = None, guards = None ):
 188         """ Send to the stdin of the application the value
 189
 190         """
 191         msg_id = os.urandom(16).encode('hex')
 192         timestamp = tsformat()
 193         payload = self._message.create_function(msg_id, self._src, rtype, timestamp , props = props ,guards = guards)
 194         self._client.publish(payload, topic)
 195
 196
 197     def frcp_request(self, topic, props = None, guards = None ):
 198         """ Execute command on the node
 199
 200         """
 201         msg_id = os.urandom(16).encode('hex')
 202         timestamp = tsformat()
 203         payload = self._message.request_function(msg_id, self._src, timestamp, props = props ,guards = guards)
 204         self._client.publish(payload, xmpp_node)
 205
 206     def frcp_release(self, parent, child, res_id = None, props = None, guards = None ):
 207         """ Delete the session and logger topics. Then disconnect
 208
 209         """
 210         msg_id = os.urandom(16).encode('hex')
 211         timestamp = tsformat()
 212         payload = self._message.release_function(msg_id, self._src, timestamp, res_id = res_id, props = props ,guards = guards)
 213         self._client.publish(payload, parent)
 214
 215         if child in self._topics:
 216             self._topics.remove(child)
 217
 218         self._client.delete(child)
 219
 220     def disconnect(self) :
 221         """ Delete the session and logger topics. Then disconnect
 222
 223         """
 224         self._client.delete(self._nepi_topic)
 225
 226         #XXX Why there is a sleep there ?
 227         time.sleep(1)
 228
 229         # Wait the send queue to be empty before disconnect
 230         self._client.disconnect(wait=True)
 231         msg = " Disconnected from XMPP Server"
 232         self.debug(msg)
 233
 234
 235 class OMF6APIFactory(object):
 236     """
 237     .. note::
 238
 239         It allows the different RM to use the same xmpp client if they use
 240         the same credentials.  For the moment, it is focused on XMPP.
 241
 242     """
 243     # use lock to avoid concurrent access to the Api list at the same times by 2
 244     # different threads
 245     lock = threading.Lock()
 246     _apis = dict()
 247
 248     @classmethod
 249     def get_api(cls, host, user, port, password, exp_id = None):
 250         """ Get an OMF Api
 251
 252         :param slice: Xmpp Slice Name
 253         :type slice: str
 254         :param host: Xmpp Server Adress
 255         :type host: str
 256         :param port: Xmpp Port (Default : 5222)
 257         :type port: str
 258         :param password: Xmpp Password
 259         :type password: str
 260
 261         """
 262         if host and user and port and password:
 263             key = cls._make_key(host, user, port, password, exp_id)
 264             cls.lock.acquire()
 265             if key in cls._apis:
 266                 #print "Api Counter : " + str(cls._apis[key]['cnt'])
 267                 cls._apis[key]['cnt'] += 1
 268                 cls.lock.release()
 269                 return cls._apis[key]['api']
 270             else :
 271                 omf_api = cls.create_api(host, user, port, password, exp_id)
 272                 cls.lock.release()
 273                 return omf_api
 274         return None
 275
 276     @classmethod
 277     def create_api(cls, host, user, port, password, exp_id):
 278         """ Create an OMF API if this one doesn't exist yet with this credentials
 279
 280         :param slice: Xmpp Slice Name
 281         :type slice: str
 282         :param host: Xmpp Server Adress
 283         :type host: str
 284         :param port: Xmpp Port (Default : 5222)
 285         :type port: str
 286         :param password: Xmpp Password
 287         :type password: str
 288
 289         """
 290         omf_api = OMF6API(host, user = user, port = port, password = password, exp_id = exp_id)
 291         key = cls._make_key(host, user, port, password, exp_id)
 292         cls._apis[key] = {}
 293         cls._apis[key]['api'] = omf_api
 294         cls._apis[key]['cnt'] = 1
 295         return omf_api
 296
 297     @classmethod
 298     def release_api(cls, host, user, port, password, exp_id = None):
 299         """ Release an OMF API with this credentials
 300
 301         :param slice: Xmpp Slice Name
 302         :type slice: str
 303         :param host: Xmpp Server Adress
 304         :type host: str
 305         :param port: Xmpp Port (Default : 5222)
 306         :type port: str
 307         :param password: Xmpp Password
 308         :type password: str
 309
 310         """
 311         if host and user and port and password:
 312             key = cls._make_key(host, user, port, password, exp_id)
 313             if key in cls._apis:
 314                 cls._apis[key]['cnt'] -= 1
 315                 #print "Api Counter : " + str(cls._apis[key]['cnt'])
 316                 if cls._apis[key]['cnt'] == 0:
 317                     omf_api = cls._apis[key]['api']
 318                     omf_api.disconnect()
 319
 320
 321     @classmethod
 322     def _make_key(cls, *args):
 323         """ Hash the credentials in order to create a key
 324
 325         :param args: list of arguments used to create the hash (user, host, port, ...)
 326         :type args: list of args
 327
 328         """
 329         skey = "".join(map(str, args))
 330         return hashlib.md5(skey).hexdigest()
 331
 332
 333