ContextClient.py

# Copyright 2021-2023 H2020 TeraFlow (https://www.teraflow-h2020.eu/)
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

from typing import Iterator
import grpc, logging
from common.tools.client.RetryDecorator import retry, delay_exponential
from context.proto.context_pb2 import (
    Connection, ConnectionEvent, ConnectionId, ConnectionIdList, ConnectionList, Context, ContextEvent, ContextId,
    ContextIdList, ContextList, Device, DeviceEvent, DeviceId, DeviceIdList, DeviceList, Empty, Link, LinkEvent,
    LinkId, LinkIdList, LinkList, Service, ServiceEvent, ServiceId, ServiceIdList, ServiceList, Topology,
    TopologyEvent, TopologyId, TopologyIdList, TopologyList)
from context.proto.context_pb2_grpc import ContextServiceStub

LOGGER = logging.getLogger(__name__)
MAX_RETRIES = 15
DELAY_FUNCTION = delay_exponential(initial=0.01, increment=2.0, maximum=5.0)
RETRY_DECORATOR = retry(max_retries=MAX_RETRIES, delay_function=DELAY_FUNCTION, prepare_method_name='connect')

class ContextClient:
    def __init__(self, address, port):
        self.endpoint = '{:s}:{:s}'.format(str(address), str(port))
        LOGGER.debug('Creating channel to {:s}...'.format(self.endpoint))
        self.channel = None
        self.stub = None
        self.connect()
        LOGGER.debug('Channel created')

    def connect(self):
        self.channel = grpc.insecure_channel(self.endpoint)
        self.stub = ContextServiceStub(self.channel)

    def close(self):
        if self.channel is not None: self.channel.close()
        self.channel = None
        self.stub = None

    @RETRY_DECORATOR
    def ListContextIds(self, request: Empty) -> ContextIdList:
        LOGGER.debug('ListContextIds request: {:s}'.format(str(request)))
        response = self.stub.ListContextIds(request)
        LOGGER.debug('ListContextIds result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def ListContexts(self, request: Empty) -> ContextList:
        LOGGER.debug('ListContexts request: {:s}'.format(str(request)))
        response = self.stub.ListContexts(request)
        LOGGER.debug('ListContexts result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def GetContext(self, request: ContextId) -> Context:
        LOGGER.debug('GetContext request: {:s}'.format(str(request)))
        response = self.stub.GetContext(request)
        LOGGER.debug('GetContext result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def SetContext(self, request: Context) -> ContextId:
        LOGGER.debug('SetContext request: {:s}'.format(str(request)))
        response = self.stub.SetContext(request)
        LOGGER.debug('SetContext result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def RemoveContext(self, request: ContextId) -> Empty:
        LOGGER.debug('RemoveContext request: {:s}'.format(str(request)))
        response = self.stub.RemoveContext(request)
        LOGGER.debug('RemoveContext result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def GetContextEvents(self, request: Empty) -> Iterator[ContextEvent]:
        LOGGER.debug('GetContextEvents request: {:s}'.format(str(request)))
        response = self.stub.GetContextEvents(request)
        LOGGER.debug('GetContextEvents result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def ListTopologyIds(self, request: ContextId) -> TopologyIdList:
        LOGGER.debug('ListTopologyIds request: {:s}'.format(str(request)))
        response = self.stub.ListTopologyIds(request)
        LOGGER.debug('ListTopologyIds result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def ListTopologies(self, request: ContextId) -> TopologyList:
        LOGGER.debug('ListTopologies request: {:s}'.format(str(request)))
        response = self.stub.ListTopologies(request)
        LOGGER.debug('ListTopologies result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def GetTopology(self, request: TopologyId) -> Topology:
        LOGGER.debug('GetTopology request: {:s}'.format(str(request)))
        response = self.stub.GetTopology(request)
        LOGGER.debug('GetTopology result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def SetTopology(self, request: Topology) -> TopologyId:
        LOGGER.debug('SetTopology request: {:s}'.format(str(request)))
        response = self.stub.SetTopology(request)
        LOGGER.debug('SetTopology result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def RemoveTopology(self, request: TopologyId) -> Empty:
        LOGGER.debug('RemoveTopology request: {:s}'.format(str(request)))
        response = self.stub.RemoveTopology(request)
        LOGGER.debug('RemoveTopology result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def GetTopologyEvents(self, request: Empty) -> Iterator[TopologyEvent]:
        LOGGER.debug('GetTopologyEvents request: {:s}'.format(str(request)))
        response = self.stub.GetTopologyEvents(request)
        LOGGER.debug('GetTopologyEvents result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def ListDeviceIds(self, request: Empty) -> DeviceIdList:
        LOGGER.debug('ListDeviceIds request: {:s}'.format(str(request)))
        response = self.stub.ListDeviceIds(request)
        LOGGER.debug('ListDeviceIds result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def ListDevices(self, request: Empty) -> DeviceList:
        LOGGER.debug('ListDevices request: {:s}'.format(str(request)))
        response = self.stub.ListDevices(request)
        LOGGER.debug('ListDevices result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def GetDevice(self, request: DeviceId) -> Device:
        LOGGER.debug('GetDevice request: {:s}'.format(str(request)))
        response = self.stub.GetDevice(request)
        LOGGER.debug('GetDevice result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def SetDevice(self, request: Device) -> DeviceId:
        LOGGER.debug('SetDevice request: {:s}'.format(str(request)))
        response = self.stub.SetDevice(request)
        LOGGER.debug('SetDevice result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def RemoveDevice(self, request: DeviceId) -> Empty:
        LOGGER.debug('RemoveDevice request: {:s}'.format(str(request)))
        response = self.stub.RemoveDevice(request)
        LOGGER.debug('RemoveDevice result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def GetDeviceEvents(self, request: Empty) -> Iterator[DeviceEvent]:
        LOGGER.debug('GetDeviceEvents request: {:s}'.format(str(request)))
        response = self.stub.GetDeviceEvents(request)
        LOGGER.debug('GetDeviceEvents result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def ListLinkIds(self, request: Empty) -> LinkIdList:
        LOGGER.debug('ListLinkIds request: {:s}'.format(str(request)))
        response = self.stub.ListLinkIds(request)
        LOGGER.debug('ListLinkIds result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def ListLinks(self, request: Empty) -> LinkList:
        LOGGER.debug('ListLinks request: {:s}'.format(str(request)))
        response = self.stub.ListLinks(request)
        LOGGER.debug('ListLinks result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def GetLink(self, request: LinkId) -> Link:
        LOGGER.debug('GetLink request: {:s}'.format(str(request)))
        response = self.stub.GetLink(request)
        LOGGER.debug('GetLink result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def SetLink(self, request: Link) -> LinkId:
        LOGGER.debug('SetLink request: {:s}'.format(str(request)))
        response = self.stub.SetLink(request)
        LOGGER.debug('SetLink result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def RemoveLink(self, request: LinkId) -> Empty:
        LOGGER.debug('RemoveLink request: {:s}'.format(str(request)))
        response = self.stub.RemoveLink(request)
        LOGGER.debug('RemoveLink result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def GetLinkEvents(self, request: Empty) -> Iterator[LinkEvent]:
        LOGGER.debug('GetLinkEvents request: {:s}'.format(str(request)))
        response = self.stub.GetLinkEvents(request)
        LOGGER.debug('GetLinkEvents result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def ListServiceIds(self, request: ContextId) -> ServiceIdList:
        LOGGER.debug('ListServiceIds request: {:s}'.format(str(request)))
        response = self.stub.ListServiceIds(request)
        LOGGER.debug('ListServiceIds result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def ListServices(self, request: ContextId) -> ServiceList:
        LOGGER.debug('ListServices request: {:s}'.format(str(request)))
        response = self.stub.ListServices(request)
        LOGGER.debug('ListServices result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def GetService(self, request: ServiceId) -> Service:
        LOGGER.debug('GetService request: {:s}'.format(str(request)))
        response = self.stub.GetService(request)
        LOGGER.debug('GetService result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def SetService(self, request: Service) -> ServiceId:
        LOGGER.debug('SetService request: {:s}'.format(str(request)))
        response = self.stub.SetService(request)
        LOGGER.debug('SetService result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def RemoveService(self, request: ServiceId) -> Empty:
        LOGGER.debug('RemoveService request: {:s}'.format(str(request)))
        response = self.stub.RemoveService(request)
        LOGGER.debug('RemoveService result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def GetServiceEvents(self, request: Empty) -> Iterator[ServiceEvent]:
        LOGGER.debug('GetServiceEvents request: {:s}'.format(str(request)))
        response = self.stub.GetServiceEvents(request)
        LOGGER.debug('GetServiceEvents result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def ListConnectionIds(self, request: ServiceId) -> ConnectionIdList:
        LOGGER.debug('ListConnectionIds request: {:s}'.format(str(request)))
        response = self.stub.ListConnectionIds(request)
        LOGGER.debug('ListConnectionIds result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def ListConnections(self, request: ServiceId) -> ConnectionList:
        LOGGER.debug('ListConnections request: {:s}'.format(str(request)))
        response = self.stub.ListConnections(request)
        LOGGER.debug('ListConnections result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def GetConnection(self, request: ConnectionId) -> Connection:
        LOGGER.debug('GetConnection request: {:s}'.format(str(request)))
        response = self.stub.GetConnection(request)
        LOGGER.debug('GetConnection result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def SetConnection(self, request: Connection) -> ConnectionId:
        LOGGER.debug('SetConnection request: {:s}'.format(str(request)))
        response = self.stub.SetConnection(request)
        LOGGER.debug('SetConnection result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def RemoveConnection(self, request: ConnectionId) -> Empty:
        LOGGER.debug('RemoveConnection request: {:s}'.format(str(request)))
        response = self.stub.RemoveConnection(request)
        LOGGER.debug('RemoveConnection result: {:s}'.format(str(response)))
        return response

    @RETRY_DECORATOR
    def GetConnectionEvents(self, request: Empty) -> Iterator[ConnectionEvent]:
        LOGGER.debug('GetConnectionEvents request: {:s}'.format(str(request)))
        response = self.stub.GetConnectionEvents(request)
        LOGGER.debug('GetConnectionEvents result: {:s}'.format(str(response)))
        return response