DLT connector: intermediate backup

#!/bin/bash -eu

mkdir -p src/common/proto

rm -rf src/common/proto/*.py

touch src/common/proto/__init__.py

python3 -m grpc_tools.protoc -I=./proto --python_out=src/common/proto/ --grpc_python_out=src/common/proto/ proto/*.proto

find src/common/proto -type f -iname *.py -exec sed -i -E 's/(import\ .*)_pb2/from . \1_pb2/g' {} \;

build slice:

  variables:

    IMAGE_NAME: 'slice' # name of the microservice

    IMAGE_NAME_TEST: 'slice-test' # name of the microservice

    IMAGE_TAG: 'latest' # tag of the container image (production, development, etc)

  stage: build

  before_script:

unit_test slice:

  variables:

    IMAGE_NAME: 'slice' # name of the microservice

    IMAGE_NAME_TEST: 'slice-test' # name of the microservice

    IMAGE_TAG: 'latest' # tag of the container image (production, development, etc)

  stage: unit_test

  needs:

# 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.

import grpc, logging, queue, threading

from common.proto.dlt_gateway_pb2 import DltRecordSubscription

from common.tools.grpc.Tools import grpc_message_to_json_string

from dlt.connector.client.DltGatewayClient import DltGatewayClient

LOGGER = logging.getLogger(__name__)

LOGGER.setLevel(logging.DEBUG)

class DltEventsCollector:

    def __init__(

        self, dltgateway_client : DltGatewayClient,

        log_events_received     : bool = False,

    ) -> None:

        self._events_queue = queue.Queue()

        self._log_events_received = log_events_received

        subscription = DltRecordSubscription() # bu default subscribe to all

        self._dltgateway_stream = dltgateway_client.SubscribeToDlt(subscription)

        self._dltgateway_thread = self._create_collector_thread(self._dltgateway_stream)

    def _create_collector_thread(self, stream, as_daemon : bool = False):

        return threading.Thread(target=self._collect, args=(stream,), daemon=as_daemon)

    def _collect(self, events_stream) -> None:

        try:

            for event in events_stream:

                if self._log_events_received:

                    LOGGER.info('[_collect] event: {:s}'.format(grpc_message_to_json_string(event)))

                self._events_queue.put_nowait(event)

        except grpc.RpcError as e:

            if e.code() != grpc.StatusCode.CANCELLED: # pylint: disable=no-member

                raise # pragma: no cover

    def start(self):

        if self._dltgateway_thread is not None: self._dltgateway_thread.start()

    def get_event(self, block : bool = True, timeout : float = 0.1):

        try:

            return self._events_queue.get(block=block, timeout=timeout)

        except queue.Empty: # pylint: disable=catching-non-exception

            return None

    def get_events(self, block : bool = True, timeout : float = 0.1, count : int = None):

        events = []

        if count is None:

            while True:

                event = self.get_event(block=block, timeout=timeout)

                if event is None: break

                events.append(event)

        else:

            for _ in range(count):

                event = self.get_event(block=block, timeout=timeout)

                if event is None: continue

                events.append(event)

        return sorted(events, key=lambda e: e.event.timestamp.timestamp)

    def stop(self):

        if self._dltgateway_stream is not None: self._dltgateway_stream.cancel()

        if self._dltgateway_thread is not None: self._dltgateway_thread.join()

import sys

from .client.DltGatewayClient import DltGatewayClient

from .client.DltEventsCollector import DltEventsCollector

def main():

    dltgateway_client_1 = DltGatewayClient(host='', port=0)

    dltgateway_client_2 = DltGatewayClient(host='', port=0)

    dltgateway_client_3 = DltGatewayClient(host='', port=0)

    dltgateway_collector_1 = DltEventsCollector(dltgateway_client_1, log_events_received=True)

    dltgateway_collector_2 = DltEventsCollector(dltgateway_client_2, log_events_received=True)

    dltgateway_collector_3 = DltEventsCollector(dltgateway_client_3, log_events_received=True)

if __name__ == '__main__':

    sys.exit(main())

# 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.

import os

from typing import Union

from common.Constants import ServiceNameEnum

from common.Settings import ENVVAR_SUFIX_SERVICE_HOST, ENVVAR_SUFIX_SERVICE_PORT_GRPC, get_env_var_name

from common.proto.dlt_gateway_pb2_grpc import add_DltGatewayServiceServicer_to_server

from common.tests.MockServicerImpl_DltGateway import MockServicerImpl_DltGateway

from common.tools.service.GenericGrpcService import GenericGrpcService

LOCAL_HOST = '127.0.0.1'

SERVICE_DLT = ServiceNameEnum.DLT

class MockService_Dependencies(GenericGrpcService):

    def __init__(self, bind_port: Union[str, int]) -> None:

        super().__init__(bind_port, LOCAL_HOST, enable_health_servicer=False, cls_name='MockService')

    # pylint: disable=attribute-defined-outside-init

    def install_servicers(self):

        self.dltgateway_servicer = MockServicerImpl_DltGateway()

        add_DltGatewayServiceServicer_to_server(self.dltgateway_servicer, self.server)

    def configure_env_vars(self):

        os.environ[get_env_var_name(SERVICE_DLT, ENVVAR_SUFIX_SERVICE_HOST     )] = str(self.bind_address)

        os.environ[get_env_var_name(SERVICE_DLT, ENVVAR_SUFIX_SERVICE_PORT_GRPC)] = str(self.bind_port)