1 # -*- coding: utf-8 -*-
3 # Copyright 2018 Telefonica S.A.
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
9 # http://www.apache.org/licenses/LICENSE-2.0
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
15 # See the License for the specific language governing permissions and
16 # limitations under the License.
18 from http
import HTTPStatus
20 from threading
import Lock
22 from osm_common
.common_utils
import FakeLock
24 __author__
= "Alfonso Tierno <alfonso.tiernosepulveda@telefonica.com>"
27 class MsgException(Exception):
29 Base Exception class for all msgXXXX exceptions
32 def __init__(self
, message
, http_code
=HTTPStatus
.SERVICE_UNAVAILABLE
):
35 :param message: descriptive text
36 :param http_code: <http.HTTPStatus> type. It contains ".value" (http error code) and ".name" (http error name
38 self
.http_code
= http_code
39 Exception.__init
__(self
, "messaging exception " + message
)
42 class MsgBase(object):
44 Base class for all msgXXXX classes
47 def __init__(self
, logger_name
="msg", lock
=False):
50 :param logger_name: logging name
51 :param lock: Used to protect simultaneous access to the same instance class by several threads:
52 False, None: Do not protect, this object will only be accessed by one thread
53 True: This object needs to be protected by several threads accessing.
54 Lock object. Use thi Lock for the threads access protection
56 self
.logger
= logging
.getLogger(logger_name
)
58 self
.lock
= FakeLock()
61 elif isinstance(lock
, Lock
):
64 raise ValueError("lock parameter must be a Lock class or boolean")
66 def connect(self
, config
):
72 def write(self
, topic
, key
, msg
):
74 "Method 'write' not implemented", http_code
=HTTPStatus
.INTERNAL_SERVER_ERROR
77 def read(self
, topic
):
79 "Method 'read' not implemented", http_code
=HTTPStatus
.INTERNAL_SERVER_ERROR
82 async def aiowrite(self
, topic
, key
, msg
):
84 "Method 'aiowrite' not implemented",
85 http_code
=HTTPStatus
.INTERNAL_SERVER_ERROR
,
89 self
, topic
, callback
=None, aiocallback
=None, group_id
=None, **kwargs
92 "Method 'aioread' not implemented",
93 http_code
=HTTPStatus
.INTERNAL_SERVER_ERROR
,