src/common/method_wrappers/Decorator.py

@@ -12,20 +12,6 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-# Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
-#
-# 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.
-
 import grpc, json, logging, threading
 from enum import Enum
 from prettytable import PrettyTable

src/dlt/connector/client/DltConnectorClient.py

@@ -12,12 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-# DltConnectorClient.py
-
-import grpc
-import logging
-import asyncio
-
+import grpc, logging
 from common.Constants import ServiceNameEnum
 from common.Settings import get_service_host, get_service_port_grpc
 from common.proto.context_pb2 import Empty, TopologyId
@@ -40,90 +35,77 @@ class DltConnectorClient:
         LOGGER.debug('Creating channel to {:s}...'.format(self.endpoint))
         self.channel = None
         self.stub = None
-        #self.connect()
-        #LOGGER.debug('Channel created')
+        self.connect()
+        LOGGER.debug('Channel created')
 
-    async def connect(self):
-        self.channel = grpc.aio.insecure_channel(self.endpoint)
+    def connect(self):
+        self.channel = grpc.insecure_channel(self.endpoint)
         self.stub = DltConnectorServiceStub(self.channel)
-        LOGGER.debug('Channel created')
 
-    async def close(self):
-        if self.channel is not None:
-            await self.channel.close()
+    def close(self):
+        if self.channel is not None: self.channel.close()
         self.channel = None
         self.stub = None
 
     @RETRY_DECORATOR
-    async def RecordAll(self, request: TopologyId) -> Empty:
+    def RecordAll(self, request : TopologyId) -> Empty:
         LOGGER.debug('RecordAll request: {:s}'.format(grpc_message_to_json_string(request)))
-        response = await self.stub.RecordAll(request)
+        response = self.stub.RecordAll(request)
         LOGGER.debug('RecordAll result: {:s}'.format(grpc_message_to_json_string(response)))
         return response
 
     @RETRY_DECORATOR
-    async def RecordAllDevices(self, request: TopologyId) -> Empty:
+    def RecordAllDevices(self, request : TopologyId) -> Empty:
         LOGGER.debug('RecordAllDevices request: {:s}'.format(grpc_message_to_json_string(request)))
-        response = await self.stub.RecordAllDevices(request)
+        response = self.stub.RecordAllDevices(request)
         LOGGER.debug('RecordAllDevices result: {:s}'.format(grpc_message_to_json_string(response)))
         return response
 
     @RETRY_DECORATOR
-   # async def RecordDevice(self, request: DltDeviceId) -> Empty:
-       # LOGGER.debug('RECORD_DEVICE request received: {:s}'.format(grpc_message_to_json_string(request)))
-    
-        # Simulate some asynchronous processing delay
-       # await asyncio.sleep(2)  # Simulates processing time
-    
-        # Create a dummy response (Empty message)
-       # response = Empty()
-    
-       # LOGGER.debug('RECORD_DEVICE processing complete for request: {:s}'.format(grpc_message_to_json_string(request)))
-       # return response
-    async def RecordDevice(self, request: DltDeviceId) -> Empty:
-        LOGGER.debug('RECORD_DEVICE request: {:s}'.format(grpc_message_to_json_string(request)))
-        response = await self.stub.RecordDevice(request)
+    def RecordDevice(self, request : DltDeviceId) -> Empty:
+        LOGGER.debug('RecordDevice request: {:s}'.format(grpc_message_to_json_string(request)))
+        response = self.stub.RecordDevice(request)
         LOGGER.debug('RecordDevice result: {:s}'.format(grpc_message_to_json_string(response)))
         return response
 
     @RETRY_DECORATOR
-    async def RecordAllLinks(self, request: TopologyId) -> Empty:
+    def RecordAllLinks(self, request : TopologyId) -> Empty:
         LOGGER.debug('RecordAllLinks request: {:s}'.format(grpc_message_to_json_string(request)))
-        response = await self.stub.RecordAllLinks(request)
+        response = self.stub.RecordAllLinks(request)
         LOGGER.debug('RecordAllLinks result: {:s}'.format(grpc_message_to_json_string(response)))
         return response
 
     @RETRY_DECORATOR
-    async def RecordLink(self, request: DltLinkId) -> Empty:
+    def RecordLink(self, request : DltLinkId) -> Empty:
         LOGGER.debug('RecordLink request: {:s}'.format(grpc_message_to_json_string(request)))
-        response = await self.stub.RecordLink(request)
+        response = self.stub.RecordLink(request)
         LOGGER.debug('RecordLink result: {:s}'.format(grpc_message_to_json_string(response)))
         return response
 
     @RETRY_DECORATOR
-    async def RecordAllServices(self, request: TopologyId) -> Empty:
+    def RecordAllServices(self, request : TopologyId) -> Empty:
         LOGGER.debug('RecordAllServices request: {:s}'.format(grpc_message_to_json_string(request)))
-        response = await self.stub.RecordAllServices(request)
+        response = self.stub.RecordAllServices(request)
         LOGGER.debug('RecordAllServices result: {:s}'.format(grpc_message_to_json_string(response)))
         return response
 
     @RETRY_DECORATOR
-    async def RecordService(self, request: DltServiceId) -> Empty:
+    def RecordService(self, request : DltServiceId) -> Empty:
         LOGGER.debug('RecordService request: {:s}'.format(grpc_message_to_json_string(request)))
-        response = await self.stub.RecordService(request)
+        response = self.stub.RecordService(request)
         LOGGER.debug('RecordService result: {:s}'.format(grpc_message_to_json_string(response)))
         return response
 
     @RETRY_DECORATOR
-    async def RecordAllSlices(self, request: TopologyId) -> Empty:
+    def RecordAllSlices(self, request : TopologyId) -> Empty:
         LOGGER.debug('RecordAllSlices request: {:s}'.format(grpc_message_to_json_string(request)))
-        response = await self.stub.RecordAllSlices(request)
+        response = self.stub.RecordAllSlices(request)
         LOGGER.debug('RecordAllSlices result: {:s}'.format(grpc_message_to_json_string(response)))
         return response
 
     @RETRY_DECORATOR
-    async def RecordSlice(self, request: DltSliceId) -> Empty:
+    def RecordSlice(self, request : DltSliceId) -> Empty:
         LOGGER.debug('RecordSlice request: {:s}'.format(grpc_message_to_json_string(request)))
-        response = await self.stub.RecordSlice(request)
+        response = self.stub.RecordSlice(request)
         LOGGER.debug('RecordSlice result: {:s}'.format(grpc_message_to_json_string(response)))
         return response

src/dlt/connector/client/DltConnectorClientSync.py → src/dlt/connector/client/DltConnectorClientAsync.py

@@ -12,7 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import grpc, logging
+import grpc, logging, asyncio
+
 from common.Constants import ServiceNameEnum
 from common.Settings import get_service_host, get_service_port_grpc
 from common.proto.context_pb2 import Empty, TopologyId
@@ -27,7 +28,7 @@ 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 DltConnectorClientSync:
+class DltConnectorClientAsync:
     def __init__(self, host=None, port=None):
         if not host: host = get_service_host(ServiceNameEnum.DLT)
         if not port: port = get_service_port_grpc(ServiceNameEnum.DLT)
@@ -35,77 +36,79 @@ class DltConnectorClientSync:
         LOGGER.debug('Creating channel to {:s}...'.format(self.endpoint))
         self.channel = None
         self.stub = None
-        self.connect()
-        LOGGER.debug('Channel created')
+        #self.connect()
+        #LOGGER.debug('Channel created')
 
-    def connect(self):
-        self.channel = grpc.insecure_channel(self.endpoint)
+    async def connect(self):
+        self.channel = grpc.aio.insecure_channel(self.endpoint)
         self.stub = DltConnectorServiceStub(self.channel)
+        LOGGER.debug('Channel created')
 
-    def close(self):
-        if self.channel is not None: self.channel.close()
+    async def close(self):
+        if self.channel is not None:
+            await self.channel.close()
         self.channel = None
         self.stub = None
 
     @RETRY_DECORATOR
-    def RecordAll(self, request : TopologyId) -> Empty:
+    async def RecordAll(self, request: TopologyId) -> Empty:
         LOGGER.debug('RecordAll request: {:s}'.format(grpc_message_to_json_string(request)))
-        response = self.stub.RecordAll(request)
+        response = await self.stub.RecordAll(request)
         LOGGER.debug('RecordAll result: {:s}'.format(grpc_message_to_json_string(response)))
         return response
 
     @RETRY_DECORATOR
-    def RecordAllDevices(self, request : TopologyId) -> Empty:
+    async def RecordAllDevices(self, request: TopologyId) -> Empty:
         LOGGER.debug('RecordAllDevices request: {:s}'.format(grpc_message_to_json_string(request)))
-        response = self.stub.RecordAllDevices(request)
+        response = await self.stub.RecordAllDevices(request)
         LOGGER.debug('RecordAllDevices result: {:s}'.format(grpc_message_to_json_string(response)))
         return response
 
     @RETRY_DECORATOR
-    def RecordDevice(self, request : DltDeviceId) -> Empty:
-        LOGGER.debug('RecordDevice request: {:s}'.format(grpc_message_to_json_string(request)))
-        response = self.stub.RecordDevice(request)
+    async def RecordDevice(self, request: DltDeviceId) -> Empty:
+        LOGGER.debug('RECORD_DEVICE request: {:s}'.format(grpc_message_to_json_string(request)))
+        response = await self.stub.RecordDevice(request)
         LOGGER.debug('RecordDevice result: {:s}'.format(grpc_message_to_json_string(response)))
         return response
 
     @RETRY_DECORATOR
-    def RecordAllLinks(self, request : TopologyId) -> Empty:
+    async def RecordAllLinks(self, request: TopologyId) -> Empty:
         LOGGER.debug('RecordAllLinks request: {:s}'.format(grpc_message_to_json_string(request)))
-        response = self.stub.RecordAllLinks(request)
+        response = await self.stub.RecordAllLinks(request)
         LOGGER.debug('RecordAllLinks result: {:s}'.format(grpc_message_to_json_string(response)))
         return response
 
     @RETRY_DECORATOR
-    def RecordLink(self, request : DltLinkId) -> Empty:
+    async def RecordLink(self, request: DltLinkId) -> Empty:
         LOGGER.debug('RecordLink request: {:s}'.format(grpc_message_to_json_string(request)))
-        response = self.stub.RecordLink(request)
+        response = await self.stub.RecordLink(request)
         LOGGER.debug('RecordLink result: {:s}'.format(grpc_message_to_json_string(response)))
         return response
 
     @RETRY_DECORATOR
-    def RecordAllServices(self, request : TopologyId) -> Empty:
+    async def RecordAllServices(self, request: TopologyId) -> Empty:
         LOGGER.debug('RecordAllServices request: {:s}'.format(grpc_message_to_json_string(request)))
-        response = self.stub.RecordAllServices(request)
+        response = await self.stub.RecordAllServices(request)
         LOGGER.debug('RecordAllServices result: {:s}'.format(grpc_message_to_json_string(response)))
         return response
 
     @RETRY_DECORATOR
-    def RecordService(self, request : DltServiceId) -> Empty:
+    async def RecordService(self, request: DltServiceId) -> Empty:
         LOGGER.debug('RecordService request: {:s}'.format(grpc_message_to_json_string(request)))
-        response = self.stub.RecordService(request)
+        response = await self.stub.RecordService(request)
         LOGGER.debug('RecordService result: {:s}'.format(grpc_message_to_json_string(response)))
         return response
 
     @RETRY_DECORATOR
-    def RecordAllSlices(self, request : TopologyId) -> Empty:
+    async def RecordAllSlices(self, request: TopologyId) -> Empty:
         LOGGER.debug('RecordAllSlices request: {:s}'.format(grpc_message_to_json_string(request)))
-        response = self.stub.RecordAllSlices(request)
+        response = await self.stub.RecordAllSlices(request)
         LOGGER.debug('RecordAllSlices result: {:s}'.format(grpc_message_to_json_string(response)))
         return response
 
     @RETRY_DECORATOR
-    def RecordSlice(self, request : DltSliceId) -> Empty:
+    async def RecordSlice(self, request: DltSliceId) -> Empty:
         LOGGER.debug('RecordSlice request: {:s}'.format(grpc_message_to_json_string(request)))
-        response = self.stub.RecordSlice(request)
+        response = await self.stub.RecordSlice(request)
         LOGGER.debug('RecordSlice result: {:s}'.format(grpc_message_to_json_string(response)))
         return response

src/dlt/connector/client/DltEventsCollector.py

@@ -16,7 +16,7 @@ from typing import Callable, Optional
 import grpc, logging, queue, threading, time
 from common.proto.dlt_gateway_pb2 import DltRecordEvent, DltRecordSubscription
 from common.tools.grpc.Tools import grpc_message_to_json_string
-from dlt.connector.client.DltGatewayClientEvent import DltGatewayClientEvent
+from src.dlt.connector.client.DltGatewayClient import DltGatewayClient
 
 LOGGER = logging.getLogger(__name__)
 LOGGER.setLevel(logging.DEBUG)
@@ -31,7 +31,7 @@ LOGGER.setLevel(logging.DEBUG)
 
 class DltEventsCollector(threading.Thread):
     def __init__(
-        self, dltgateway_client : DltGatewayClientEvent,
+        self, dltgateway_client : DltGatewayClient,
         log_events_received     : bool = False,
         event_handler           : Optional[Callable[[DltRecordEvent], Optional[DltRecordEvent]]] = None,
     ) -> None:

src/dlt/connector/client/DltGatewayClient.py

@@ -13,14 +13,9 @@
 # limitations under the License.
 
 
-
-import grpc
-import logging
-import asyncio
-from typing import Iterator, List
-from common.proto.context_pb2 import Empty, TeraFlowController
-from common.proto.dlt_gateway_pb2 import (
-    DltPeerStatus, DltPeerStatusList, DltRecord, DltRecordEvent, DltRecordId, DltRecordStatus, DltRecordSubscription)
+import grpc, logging
+from typing import Iterator
+from common.proto.dlt_gateway_pb2 import DltRecordEvent, DltRecordSubscription
 from common.proto.dlt_gateway_pb2_grpc import DltGatewayServiceStub
 from common.tools.client.RetryDecorator import retry, delay_exponential
 from common.tools.grpc.Tools import grpc_message_to_json_string
@@ -40,70 +35,22 @@ class DltGatewayClient:
         LOGGER.debug('Creating channel to {:s}...'.format(self.endpoint))
         self.channel = None
         self.stub = None
-        #self.connect()
-        self.message_queue: List[DltRecord] = []
-        #LOGGER.debug('Channel created')
-
+        self.connect()
+        LOGGER.debug('Channel created')
 
-    async def connect(self):
-        self.channel = grpc.aio.insecure_channel(self.endpoint)
+    def connect(self):
+        self.channel = grpc.insecure_channel(self.endpoint)
         self.stub = DltGatewayServiceStub(self.channel)
-        LOGGER.debug('Channel created')
 
-    async def close(self):
+    def close(self):
         if self.channel is not None:
-            await self.channel.close()
+            self.channel.close()
         self.channel = None
         self.stub = None
 
-    # async def dummy_process(self, request: DltRecord):
-    #     # Simulate processing delay
-    #     await asyncio.sleep(2)
-    #     return DltRecordStatus(status="DLTRECORDSTATUS_SUCCEEDED", record_id=request.record_id)
-
-
-    @RETRY_DECORATOR
-    async def RecordToDlt(self, request : DltRecord) -> DltRecordStatus:
-        LOGGER.debug('RecordToDlt request: {:s}'.format(grpc_message_to_json_string(request)))
-        response = await self.stub.RecordToDlt(request)
-        LOGGER.debug('RecordToDlt result: {:s}'.format(grpc_message_to_json_string(response)))
-        return response
-    
-    # @RETRY_DECORATOR
-    # async def RecordToDlt(self, request: DltRecord) -> DltRecordStatus:
-    #     self.message_queue.append(request)
-    #     LOGGER.debug(f'RecordToDlt request: {grpc_message_to_json_string(request)}')
-    #     LOGGER.debug(f'Queue length before processing: {len(self.message_queue)}')
-    #     response = await self.dummy_process(request)
-    #     LOGGER.debug(f'RecordToDlt result: {grpc_message_to_json_string(response)}')
-    #     self.message_queue.remove(request)
-    #     LOGGER.debug(f'Queue length after processing: {len(self.message_queue)}')
-    #     return response
-
     @RETRY_DECORATOR
-    async def GetFromDlt(self, request : DltRecordId) -> DltRecord:
-        LOGGER.debug('GetFromDlt request: {:s}'.format(grpc_message_to_json_string(request)))
-        response = await self.stub.GetFromDlt(request)
-        LOGGER.debug('GetFromDlt result: {:s}'.format(grpc_message_to_json_string(response)))
-        return response
-
-    @RETRY_DECORATOR
-    def SubscribeToDlt(self, request : DltRecordSubscription) -> Iterator[DltRecordEvent]:
+    def SubscribeToDlt(self, request: DltRecordSubscription) -> Iterator[DltRecordEvent]:
         LOGGER.debug('SubscribeToDlt request: {:s}'.format(grpc_message_to_json_string(request)))
         response = self.stub.SubscribeToDlt(request)
         LOGGER.debug('SubscribeToDlt result: {:s}'.format(grpc_message_to_json_string(response)))
         return response
-
-    @RETRY_DECORATOR
-    async def GetDltStatus(self, request : TeraFlowController) -> DltPeerStatus:
-        LOGGER.debug('GetDltStatus request: {:s}'.format(grpc_message_to_json_string(request)))
-        response = await self.stub.GetDltStatus(request)
-        LOGGER.debug('GetDltStatus result: {:s}'.format(grpc_message_to_json_string(response)))
-        return response
-
-    @RETRY_DECORATOR
-    async def GetDltPeers(self, request : Empty) -> DltPeerStatusList:
-        LOGGER.debug('GetDltPeers request: {:s}'.format(grpc_message_to_json_string(request)))
-        response = await self.stub.GetDltPeers(request)
-        LOGGER.debug('GetDltPeers result: {:s}'.format(grpc_message_to_json_string(response)))
-        return response

src/dlt/connector/client/DltGatewayClientEvent.py → src/dlt/connector/client/DltGatewayClientAsync.py

@@ -12,11 +12,11 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-
-import grpc
-import logging
-from typing import Iterator
-from common.proto.dlt_gateway_pb2 import DltRecordEvent, DltRecordSubscription
+import grpc, logging, asyncio
+from typing import Iterator, List
+from common.proto.context_pb2 import Empty, TeraFlowController
+from common.proto.dlt_gateway_pb2 import (
+    DltPeerStatus, DltPeerStatusList, DltRecord, DltRecordEvent, DltRecordId, DltRecordStatus, DltRecordSubscription)
 from common.proto.dlt_gateway_pb2_grpc import DltGatewayServiceStub
 from common.tools.client.RetryDecorator import retry, delay_exponential
 from common.tools.grpc.Tools import grpc_message_to_json_string
@@ -28,7 +28,7 @@ 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 DltGatewayClientEvent:
+class DltGatewayClientAsync:
     def __init__(self, host=None, port=None):
         if not host: host = DLT_GATEWAY_HOST
         if not port: port = DLT_GATEWAY_PORT
@@ -36,22 +36,52 @@ class DltGatewayClientEvent:
         LOGGER.debug('Creating channel to {:s}...'.format(self.endpoint))
         self.channel = None
         self.stub = None
-        self.connect()
-        LOGGER.debug('Channel created')
+        #self.connect()
+        self.message_queue: List[DltRecord] = []
+        #LOGGER.debug('Channel created')
 
-    def connect(self):
-        self.channel = grpc.insecure_channel(self.endpoint)
+    async def connect(self):
+        self.channel = grpc.aio.insecure_channel(self.endpoint)
         self.stub = DltGatewayServiceStub(self.channel)
+        LOGGER.debug('Channel created')
 
-    def close(self):
+    async def close(self):
         if self.channel is not None:
-            self.channel.close()
+            await self.channel.close()
         self.channel = None
         self.stub = None
 
     @RETRY_DECORATOR
-    def SubscribeToDlt(self, request: DltRecordSubscription) -> Iterator[DltRecordEvent]:
+    async def RecordToDlt(self, request : DltRecord) -> DltRecordStatus:
+        LOGGER.debug('RecordToDlt request: {:s}'.format(grpc_message_to_json_string(request)))
+        response = await self.stub.RecordToDlt(request)
+        LOGGER.debug('RecordToDlt result: {:s}'.format(grpc_message_to_json_string(response)))
+        return response
+    
+    @RETRY_DECORATOR
+    async def GetFromDlt(self, request : DltRecordId) -> DltRecord:
+        LOGGER.debug('GetFromDlt request: {:s}'.format(grpc_message_to_json_string(request)))
+        response = await self.stub.GetFromDlt(request)
+        LOGGER.debug('GetFromDlt result: {:s}'.format(grpc_message_to_json_string(response)))
+        return response
+
+    @RETRY_DECORATOR
+    def SubscribeToDlt(self, request : DltRecordSubscription) -> Iterator[DltRecordEvent]:
         LOGGER.debug('SubscribeToDlt request: {:s}'.format(grpc_message_to_json_string(request)))
         response = self.stub.SubscribeToDlt(request)
         LOGGER.debug('SubscribeToDlt result: {:s}'.format(grpc_message_to_json_string(response)))
         return response
+
+    @RETRY_DECORATOR
+    async def GetDltStatus(self, request : TeraFlowController) -> DltPeerStatus:
+        LOGGER.debug('GetDltStatus request: {:s}'.format(grpc_message_to_json_string(request)))
+        response = await self.stub.GetDltStatus(request)
+        LOGGER.debug('GetDltStatus result: {:s}'.format(grpc_message_to_json_string(response)))
+        return response
+
+    @RETRY_DECORATOR
+    async def GetDltPeers(self, request : Empty) -> DltPeerStatusList:
+        LOGGER.debug('GetDltPeers request: {:s}'.format(grpc_message_to_json_string(request)))
+        response = await self.stub.GetDltPeers(request)
+        LOGGER.debug('GetDltPeers result: {:s}'.format(grpc_message_to_json_string(response)))
+        return response

src/dlt/connector/service/DltConnectorServiceServicerImpl.py

@@ -12,10 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-import grpc
-import asyncio
+import grpc, asyncio, logging
 from grpc.aio import ServicerContext
-import logging
 from typing import Optional
 from common.method_wrappers.Decorator import MetricsPool, safe_and_metered_rpc_method_async
 from common.proto.context_pb2 import Empty, TopologyId
@@ -24,7 +22,7 @@ from common.proto.dlt_connector_pb2_grpc import DltConnectorServiceServicer
 from common.proto.dlt_gateway_pb2 import DltRecord, DltRecordId, DltRecordOperationEnum, DltRecordTypeEnum
 from common.tools.grpc.Tools import grpc_message_to_json_string
 from context.client.ContextClient import ContextClient
-from dlt.connector.client.DltGatewayClient import DltGatewayClient
+from src.dlt.connector.client.DltGatewayClientAsync import DltGatewayClientAsync
 from .tools.Checkers import record_exists
 
 LOGGER = logging.getLogger(__name__)
@@ -35,7 +33,7 @@ METRICS_POOL = MetricsPool('DltConnector', 'RPC')
 class DltConnectorServiceServicerImpl(DltConnectorServiceServicer):
     def __init__(self):
         LOGGER.debug('Creating Servicer...')
-        self.dltgateway_client = DltGatewayClient()
+        self.dltgateway_client = DltGatewayClientAsync()
         LOGGER.debug('Servicer Created')
 
     async def initialize(self):
@@ -50,16 +48,6 @@ class DltConnectorServiceServicerImpl(DltConnectorServiceServicer):
         return Empty()
 
     @safe_and_metered_rpc_method_async(METRICS_POOL, LOGGER)
-    # async def RecordDevice(self, request: DltDeviceId, context: ServicerContext) -> Empty:
-    #     LOGGER.debug('Received RecordDevice request: {:s}'.format(grpc_message_to_json_string(request)))
-    #     try:
-    #         if not request.delete:
-    #             LOGGER.debug('Processing RecordDevice request: {:s}'.format(grpc_message_to_json_string(request)))
-    #             # Perform any dummy operation or simply log the request
-    #         LOGGER.debug('Processed RecordDevice request: {:s}'.format(grpc_message_to_json_string(request)))
-    #     except Exception as e:
-    #         LOGGER.error(f"Error processing RecordDevice: {e}")
-    #     return Empty()
     async def RecordDevice(self, request: DltDeviceId, context: ServicerContext) -> Empty:
         data_json = None
         LOGGER.debug('RECORD_DEVICE = {:s}'.format(grpc_message_to_json_string(request)))

src/dlt/connector/service/__main__.py

@@ -13,11 +13,7 @@
 # limitations under the License.
 
 
-import logging
-import signal
-import sys
-import threading
-import asyncio
+import logging, signal, sys, threading, asyncio
 from prometheus_client import start_http_server
 from common.Constants import ServiceNameEnum
 from common.Settings import (

src/dlt/connector/service/event_dispatcher/DltEventDispatcher.py

@@ -26,9 +26,9 @@ from common.tools.grpc.Tools import grpc_message_to_json_string
 from common.tools.object_factory.Context import json_context_id
 from common.tools.object_factory.Topology import json_topology_id
 from context.client.ContextClient import ContextClient
-from dlt.connector.client.DltConnectorClientSync import DltConnectorClientSync
+from src.dlt.connector.client.DltConnectorClient import DltConnectorClient
 from dlt.connector.client.DltEventsCollector import DltEventsCollector
-from dlt.connector.client.DltGatewayClientEvent import DltGatewayClientEvent
+from src.dlt.connector.client.DltGatewayClient import DltGatewayClient
 from interdomain.client.InterdomainClient import InterdomainClient
 
 LOGGER = logging.getLogger(__name__)
@@ -40,8 +40,8 @@ ADMIN_CONTEXT_ID = ContextId(**json_context_id(DEFAULT_CONTEXT_NAME))
 class Clients:
     def __init__(self) -> None:
         self.context_client = ContextClient()
-        self.dlt_connector_client = DltConnectorClientSync()
-        self.dlt_gateway_client = DltGatewayClientEvent()
+        self.dlt_connector_client = DltConnectorClient()
+        self.dlt_gateway_client = DltGatewayClient()
         self.interdomain_client = InterdomainClient()
 
     def close(self) -> None:

src/dlt/gateway/Dockerfile

@@ -10,7 +10,8 @@ COPY src/dlt/gateway/dltApp/package*.json ./
 # Copy tsconfig.json
 COPY src/dlt/gateway/dltApp/tsconfig*.json ./
 # Copy the proto folder
-COPY src/dlt/gateway/dltApp/proto/ ./proto
+COPY proto/context.proto ./proto
+COPY proto/dlt_gateway.proto ./proto
 
 # Copy the src folder
 COPY src/dlt/gateway/dltApp/src/ ./src

src/dlt/gateway/automation/k8sconfig/configmap.yaml

-apiVersion: v1
-kind: ConfigMap
-metadata:
-  name: dlt-config
-  namespace: dlt
-#NEED to find a better way to setup ENV variables for Kubernetes services
-##Modify the PEER_ENDPOINT IP according to your deployment
-data:
-  CHANNEL_NAME: "channel1"
-  CHAINCODE_NAME: "adrenalineDLT"
-  MSP_ID: "Org1MSP"
-  PEER_ENDPOINT: "PEER_IP:PORT" #USE THE IP address and Ports of your Fabric deployment peers.
-  PEER_HOST_ALIAS: "peer0.org1.adrenaline.com"
-  CRYPTO_PATH: "/test-network/organizations/peerOrganizations/org1.adrenaline.com"
-  KEY_DIRECTORY_PATH: "/test-network/organizations/peerOrganizations/org1.adrenaline.com/users/User1@org1.adrenaline.com/msp/keystore"
-  CERT_DIRECTORY_PATH: "/test-network/organizations/peerOrganizations/org1.adrenaline.com/users/User1@org1.adrenaline.com/msp/signcerts"
-  TLS_CERT_PATH: "/test-network/organizations/peerOrganizations/org1.adrenaline.com/peers/peer0.org1.adrenaline.com/tls/ca.crt"

src/dlt/gateway/automation/k8sconfig/deploy_dlt_gateway.sh

-#!/bin/bash
-
-# Namespace
-NAMESPACE="dlt"
-
-# Configurations
-CONFIGMAP="configmap.yaml"
-PV="persistent-volume.yaml"
-PVC="persistent-volume-claim.yaml"
-DEPLOYMENT="deployment.yaml"
-SERVICE="service.yaml"
-
-# Apply Configurations
-echo "Applying ConfigMap..."
-kubectl apply -f $CONFIGMAP
-
-echo "Applying PersistentVolume..."
-kubectl apply -f $PV
-
-echo "Applying PersistentVolumeClaim..."
-kubectl apply -f $PVC
-
-echo "Applying Deployment..."
-kubectl apply -f $DEPLOYMENT
-
-echo "Applying Service..."
-kubectl apply -f $SERVICE
-
-# Verify Deployment
-echo "Verifying Deployment..."
-kubectl get pods -n $NAMESPACE
-kubectl get services -n $NAMESPACE
-
-echo "Deployment Completed."

src/dlt/gateway/automation/k8sconfig/deployment.yaml

-apiVersion: apps/v1
-kind: Deployment
-metadata:
-  name: dlt-gateway
-  namespace: dlt
-spec:
-  replicas: 3
-  selector:
-    matchLabels:
-      app: dlt-gateway
-  template:
-    metadata:
-      labels:
-        app: dlt-gateway
-    spec:
-      containers:
-      - name: dlt-gateway
-        image: shaifvier/dltgateway:v1.0.0
-        ports:
-        - containerPort: 50051
-        volumeMounts:
-        - mountPath: /test-network
-          name: dlt-volume
-          readOnly: true  # Mount the volume as read-only
-        env:
-        - name: CHANNEL_NAME
-          valueFrom:
-            configMapKeyRef:
-              name: dlt-config
-              key: CHANNEL_NAME
-        - name: CHAINCODE_NAME
-          valueFrom:
-            configMapKeyRef:
-              name: dlt-config
-              key: CHAINCODE_NAME
-        - name: MSP_ID
-          valueFrom:
-            configMapKeyRef:
-              name: dlt-config
-              key: MSP_ID
-        - name: PEER_ENDPOINT
-          valueFrom:
-            configMapKeyRef:
-              name: dlt-config
-              key: PEER_ENDPOINT
-        - name: PEER_HOST_ALIAS
-          valueFrom:
-            configMapKeyRef:
-              name: dlt-config
-              key: PEER_HOST_ALIAS
-        - name: CRYPTO_PATH
-          valueFrom:
-            configMapKeyRef:
-              name: dlt-config
-              key: CRYPTO_PATH
-        - name: KEY_DIRECTORY_PATH
-          valueFrom:
-            configMapKeyRef:
-              name: dlt-config
-              key: KEY_DIRECTORY_PATH
-        - name: CERT_DIRECTORY_PATH
-          valueFrom:
-            configMapKeyRef:
-              name: dlt-config
-              key: CERT_DIRECTORY_PATH
-        - name: TLS_CERT_PATH
-          valueFrom:
-            configMapKeyRef:
-              name: dlt-config
-              key: TLS_CERT_PATH
-      volumes:
-      - name: dlt-volume
-        persistentVolumeClaim:
-          claimName: dlt-pvc

src/dlt/gateway/automation/k8sconfig/persistent-volume-claim.yaml

-apiVersion: v1
-kind: PersistentVolumeClaim
-metadata:
-  name: dlt-pvc
-  namespace: dlt
-spec:
-  accessModes:
-    - ReadOnlyMany
-  resources:
-    requests:
-      storage: 1Gi

src/dlt/gateway/automation/k8sconfig/persistent-volume.yaml

-apiVersion: v1
-kind: PersistentVolume
-metadata:
-  name: dlt-pv
-spec:
-  capacity:
-    storage: 1Gi
-  accessModes:
-    - ReadOnlyMany
-  persistentVolumeReclaimPolicy: Retain
-  hostPath:
-    path: "/home/cttc/test-network"  # Update this path to the actual path on the host machine where the Kubernetes Pod will be deployed.
-  claimRef:
-    namespace: dlt
-    name: dlt-pvc
\ No newline at end of file

src/dlt/gateway/automation/k8sconfig/remove_dlt_gateway.sh

-#!/bin/bash
-
-# Namespace
-NAMESPACE="dlt"
-
-# Configurations
-CONFIGMAP="configmap.yaml"
-PV="persistent-volume.yaml"
-PVC="persistent-volume-claim.yaml"
-DEPLOYMENT="deployment.yaml"
-SERVICE="service.yaml"
-
-# Delete Configurations
-echo "Deleting Deployment..."
-kubectl delete -f $DEPLOYMENT || echo "Deployment not found."
-
-echo "Deleting Service..."
-kubectl delete -f $SERVICE || echo "Service not found."
-
-echo "Deleting PersistentVolumeClaim..."
-kubectl delete -f $PVC || echo "PersistentVolumeClaim not found."
-
-echo "Deleting PersistentVolume..."
-kubectl delete -f $PV || echo "PersistentVolume not found."
-
-echo "Deleting ConfigMap..."
-kubectl delete -f $CONFIGMAP || echo "ConfigMap not found."
-
-# Verify Deletion
-echo "Verifying Deletion..."
-kubectl get pods -n $NAMESPACE
-kubectl get services -n $NAMESPACE
-kubectl get pvc -n $NAMESPACE
-kubectl get pv
-kubectl get configmap -n $NAMESPACE
-
-echo "Deletion Completed."

src/dlt/gateway/automation/k8sconfig/service.yaml

-apiVersion: v1
-kind: Service
-metadata:
-  name: dlt-gateway
-  namespace: dlt
-spec:
-  selector:
-    app: dlt-gateway
-  ports:
-    - protocol: TCP
-      port: 50051       # External port
-      targetPort: 50051  # Internal port of the service
-      nodePort: 32001    # A high port number for external access
-  type: NodePort

src/dlt/gateway/automation/k8sconfig/simpletest.yaml

-apiVersion: v1
-kind: Pod
-metadata:
-  name: dlt-gateway
-  namespace: dlt
-spec:
-  containers:
-  - name: test-container
-    image: busybox
-    command: ["sh", "-c", "sleep 3600"]  # Keep the container running for testing
-    volumeMounts:
-    - mountPath: /mnt/test-network
-      name: dlt-volume
-      readOnly: true
-  volumes:
-  - name: dlt-volume
-    hostPath:
-      path: /home/cttc/test-network
-      type: Directory

src/dlt/gateway/chaincode/adrenalineTopo.go → src/dlt/gateway/chaincode/chaincode.go

+//  Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
+
+//  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.
+
 package main
 
 import (

src/dlt/gateway/chaincode/go.mod

-module adrenalineTopo
+<!--   Copyright 2022-2024 ETSI OSG/SDG TeraFlowSDN (TFS) (https://tfs.etsi.org/)
+
+  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. -->
+
+module chaincode
 
 go 1.17