rwlaunchpad/plugins/rwpkgmgr/rift/tasklets/rwpkgmgr/rwpkgmgr.py

   1 """
   2 #
   3 #   Copyright 2016 RIFT.IO Inc
   4 #
   5 #   Licensed under the Apache License, Version 2.0 (the "License");
   6 #   you may not use this file except in compliance with the License.
   7 #   You may obtain a copy of the License at
   8 #
   9 #       http://www.apache.org/licenses/LICENSE-2.0
  10 #
  11 #   Unless required by applicable law or agreed to in writing, software
  12 #   distributed under the License is distributed on an "AS IS" BASIS,
  13 #   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 #   See the License for the specific language governing permissions and
  15 #   limitations under the License.
  16 #
  17
  18 @file rwpkgmgr.py
  19 @author Varun Prasad (varun.prasad@riftio.com)
  20 @date 18-Sep-2016
  21
  22 """
  23
  24 import asyncio
  25
  26 import gi
  27 gi.require_version('RwDts', '1.0')
  28 gi.require_version('RwPkgMgmtYang', '1.0')
  29
  30
  31 from gi.repository import (
  32         RwDts as rwdts,
  33         RwPkgMgmtYang)
  34 import rift.tasklets
  35
  36
  37 from . import rpc
  38 from .proxy import filesystem
  39 from . import publisher as pkg_publisher
  40 from . import subscriber
  41
  42 class PackageManagerTasklet(rift.tasklets.Tasklet):
  43     def __init__(self, *args, **kwargs):
  44         try:
  45             super().__init__(*args, **kwargs)
  46             self.rwlog.set_category("rw-mano-log")
  47             self.endpoint_rpc = None
  48             self.schema_rpc = None
  49         except Exception as e:
  50             self.log.exception(e)
  51
  52     def start(self):
  53
  54         self.log.debug("Registering with dts")
  55
  56         try:
  57             super().start()
  58             self.dts = rift.tasklets.DTS(
  59                 self.tasklet_info,
  60                 RwPkgMgmtYang.get_schema(),
  61                 self.loop,
  62                 self.on_dts_state_change
  63                 )
  64
  65             proxy = filesystem.FileSystemProxy(self.loop, self.log)
  66             args = [self.log, self.dts, self.loop]
  67
  68         # create catalog publishers
  69             self.job_handler = pkg_publisher.DownloadStatusPublisher(*args)
  70             self.copy_publisher = pkg_publisher.CopyStatusPublisher(*args +[self.tasklet_info])
  71
  72         # create catalog subscribers
  73             self.vnfd_catalog_sub = subscriber.VnfdStatusSubscriber(*args)
  74             self.nsd_catalog_sub = subscriber.NsdStatusSubscriber(*args)
  75
  76             args.append(proxy)
  77             self.endpoint_rpc = rpc.EndpointDiscoveryRpcHandler(*args)
  78             self.schema_rpc = rpc.SchemaRpcHandler(*args)
  79             self.delete_rpc = rpc.PackageDeleteOperationsRpcHandler(*args)
  80             self.copy_rpc = rpc.PackageCopyOperationsRpcHandler(*(args + [self.copy_publisher]))
  81
  82             args.append(self.job_handler)
  83             self.pkg_op = rpc.PackageOperationsRpcHandler(*args)
  84
  85         except Exception as e:
  86             self.log.error("Exception caught rwpkgmgr start: %s", str(e))
  87         else:
  88             self.log.debug("rwpkgmgr started successfully!")
  89
  90     def stop(self):
  91         try:
  92             self.dts.deinit()
  93         except Exception as e:
  94             self.log.exception(e)
  95
  96     @asyncio.coroutine
  97     def init(self):
  98         try:
  99             yield from self.endpoint_rpc.register()
 100             yield from self.schema_rpc.register()
 101             yield from self.pkg_op.register()
 102             yield from self.job_handler.register()
 103             yield from self.delete_rpc.register()
 104             yield from self.copy_rpc.register()
 105             yield from self.copy_publisher.register()
 106             yield from self.vnfd_catalog_sub.register()
 107             yield from self.nsd_catalog_sub.register()
 108         except Exception as e:
 109             self.log.error("Exception caught rwpkgmgr init %s", str(e))
 110
 111     @asyncio.coroutine
 112     def run(self):
 113         pass
 114
 115     @asyncio.coroutine
 116     def on_dts_state_change(self, state):
 117         """Handle DTS state change
 118
 119         Take action according to current DTS state to transition application
 120         into the corresponding application state
 121
 122         Arguments
 123             state - current dts state
 124
 125         """
 126         switch = {
 127             rwdts.State.INIT: rwdts.State.REGN_COMPLETE,
 128             rwdts.State.CONFIG: rwdts.State.RUN,
 129         }
 130
 131         handlers = {
 132             rwdts.State.INIT: self.init,
 133             rwdts.State.RUN: self.run,
 134         }
 135
 136         # Transition application to next state
 137         handler = handlers.get(state, None)
 138         if handler is not None:
 139             yield from handler()
 140
 141         # Transition dts to next state
 142         next_state = switch.get(state, None)
 143         if next_state is not None:
 144             self.dts.handle.set_state(next_state)