diff --git a/callcontrol/controller.py b/callcontrol/controller.py index 452def5..c31c4d4 100644 --- a/callcontrol/controller.py +++ b/callcontrol/controller.py @@ -1,514 +1,513 @@ """Implementation of a call control server for OpenSIPS.""" import os import grp import re import pickle import time from application import log from application.configuration import ConfigSection, ConfigSetting from application.process import process from application.system import unlink from twisted.internet.protocol import Factory from twisted.protocols.basic import LineOnlyReceiver from twisted.internet import reactor, defer from twisted.python import failure from callcontrol.scheduler import RecurrentCall, KeepRunning from callcontrol.raddb import RadiusDatabase, RadiusDatabaseError from callcontrol.sip import Call from callcontrol.rating import RatingEngineConnections from callcontrol import configuration_file, backup_calls_file class TimeLimit(int): """A positive time limit (in seconds) or None""" def __new__(typ, value): if value.lower() == 'none': return None try: limit = int(value) except: raise ValueError("invalid time limit value: %r" % value) if limit < 0: raise ValueError("invalid time limit value: %r. should be positive." % value) return limit class CallControlConfig(ConfigSection): __cfgfile__ = configuration_file __section__ = 'CallControl' socket = process.runtime.file('socket') group = 'opensips' limit = ConfigSetting(type=TimeLimit, value=None) timeout = 24*60*60 # timeout calls that are stale for more than 24 hours. setupTime = 90 # timeout calls that take more than 1'30" to setup. checkInterval = 60 # check for staled calls and calls that did timeout at every minute. class CommandError(Exception): pass class InvalidRequestError(Exception): pass class CallsMonitor(object): """Check for staled calls""" def __init__(self, period, application): self.application = application self.reccall = RecurrentCall(period, self.run) def run(self): now = time.time() staled = [] nosetup = [] for callid, call in list(self.application.calls.items()): if not call.complete and (now - call.created >= CallControlConfig.setupTime): self.application.clean_call(callid) nosetup.append(call) elif call.inprogress and call.timer is not None: continue # this call will be expired by its own timer elif now - call.created >= CallControlConfig.timeout: self.application.clean_call(callid) staled.append(call) # Terminate staled for call in staled: call.end(reason='calls monitor as staled', sendbye=True) # Terminate calls that didn't setup in setupTime for call in nosetup: call.end(reason="calls monitor as it didn't setup in %d seconds" % CallControlConfig.setupTime) return KeepRunning def shutdown(self): self.reccall.cancel() class CallControlProtocol(LineOnlyReceiver): def lineReceived(self, line): line = line.decode('utf-8') if line.strip() == "": if self.line_buf: self._process() self.line_buf = [] else: self.line_buf.append(line.strip()) def _process(self): try: req = Request(self.line_buf[0], self.line_buf[1:]) except InvalidRequestError as e: log.info("Invalid OpenSIPS request: %s" % str(e)) self._send_error_reply(failure.Failure(e)) else: log.debug('Received request from OpenSIPS %s', req) def _unknown_handler(req): req.deferred.errback(failure.Failure(CommandError(req))) try: getattr(self, '_CC_%s' % req.cmd, _unknown_handler)(req) except Exception as e: self._send_error_reply(failure.Failure(e)) else: req.deferred.addCallbacks(callback=self._send_reply, errback=self._send_error_reply) def connectionMade(self): self.line_buf = [] def _send_reply(self, msg): log.debug('Send response to OpenSIPS: %s', msg) self.sendLine(msg.encode('utf-8')) def _send_error_reply(self, fail): log.error(fail.value) log.info("Sent 'Error' response to OpenSIPS") self.sendLine(b'Error') def _CC_init(self, req): - #import pdb; pdb.set_trace() try: call = self.factory.application.calls[req.callid] except KeyError: call = Call(req, self.factory.application) if call.billingParty is None: req.deferred.callback('Error') return self.factory.application.calls[req.callid] = call # log.debug('Call id %s added to list of controlled calls', call.callid) else: if call.token != req.call_token: log.error("Call id %s is duplicated" % call.callid) req.deferred.callback('Duplicated callid') return # The call was previously setup which means it could be in the the users table try: user_calls = self.factory.application.users[call.billingParty] user_calls.remove(call.callid) if len(user_calls) == 0: del self.factory.application.users[call.billingParty] self.factory.application.engines.remove_user(call.billingParty) except (ValueError, KeyError): pass deferred = call.setup(req) deferred.addCallbacks(callback=self._CC_finish_init, errback=self._CC_init_failed, callbackArgs=[req], errbackArgs=[req]) def _CC_finish_init(self, value, req): try: call = self.factory.application.calls[req.callid] except KeyError: log.error("Call id %s disappeared before we could finish initializing it" % req.callid) req.deferred.callback('Error') else: if req.call_limit is not None and len(self.factory.application.users.get(call.billingParty, ())) >= req.call_limit: log.info("Call id %s of %s to %s forbidden because limit has been reached" % (req.callid, call.user, call.ruri)) self.factory.application.clean_call(req.callid) call.end() req.deferred.callback('Call limit reached') elif call.locked: # prepaid account already locked by another call log.info("Call from %s to %s is forbidden because account is locked" % (call.user, call.ruri, req.callid)) self.factory.application.clean_call(req.callid) call.end() req.deferred.callback('Locked') elif call.timelimit == 0: # prepaid account with no credit log.info("Call from %s to %s is forbidden because of low credit (%s)" % (call.user, call.ruri, req.callid)) self.factory.application.clean_call(req.callid) call.end() req.deferred.callback('No credit') elif req.call_limit is not None or call.timelimit is not None: # call limited by credit value, a global time limit or number of calls log.info("%s can make %s concurrent calls (%s)" % (call.billingParty, req.call_limit or "unlimited", req.callid)) self.factory.application.users.setdefault(call.billingParty, []).append(call.callid) req.deferred.callback('Limited') else: # no limit for call log.info("Call from %s to %s is postpaid without limits (%s)" % (call.user, call.ruri, req.callid)) self.factory.application.clean_call(req.callid) call.end() req.deferred.callback('No limit') def _CC_init_failed(self, fail, req): self._send_error_reply(fail) self.factory.application.clean_call(req.callid) def _CC_start(self, req): try: call = self.factory.application.calls[req.callid] except KeyError: req.deferred.callback('Not found') else: call.start(req) req.deferred.callback('Ok') def _CC_stop(self, req): try: call = self.factory.application.calls[req.callid] except KeyError: req.deferred.callback('Not found') else: self.factory.application.clean_call(req.callid) call.end(reason='user') req.deferred.callback('Ok') def _CC_debug(self, req): debuglines = [] if req.show == 'sessions': for callid, call in list(self.factory.application.calls.items()): if not req.user or call.user.startswith(req.user): debuglines.append('Call id %s of %s to %s: %s' % (callid, call.user, call.ruri, call.status)) elif req.show == 'session': try: call = self.factory.application.calls[req.callid] except KeyError: debuglines.append('Call id %s does not exist' % req.callid) else: for key, value in list(call.items()): debuglines.append('%s: %s' % (key, value)) req.deferred.callback('\r\n'.join(debuglines)+'\r\n') def _CC_terminate(self, req): try: call = self.factory.application.calls[req.callid] except KeyError: req.deferred.callback('Call id %s does not exist\r\n' % req.callid) else: self.factory.application.clean_call(req.callid) call.end(reason='admin', sendbye=True) req.deferred.callback('Ok\r\n') class CallControlFactory(Factory): protocol = CallControlProtocol def __init__(self, application): self.application = application class CallControlServer(object): def __init__(self): unlink(CallControlConfig.socket) self.path = CallControlConfig.socket self.group = CallControlConfig.group self.listening = None self.engines = None self.monitor = None self.calls = {} self.users = {} self._restore_calls() def clean_call(self, callid): try: call = self.calls[callid] except KeyError: return else: del self.calls[callid] user_calls = self.users.get(call.billingParty, []) try: user_calls.remove(callid) except ValueError: pass if not user_calls: self.users.pop(call.billingParty, None) self.engines.remove_user(call.billingParty) # log.debug('Call id %s removed from the list of controlled calls', callid) def run(self): reactor.addSystemEventTrigger('before', 'startup', self.on_startup) reactor.addSystemEventTrigger('before', 'shutdown', self.on_shutdown) reactor.run() def stop(self): reactor.stop() def on_startup(self): # First set up listening on the unix socket try: gid = grp.getgrnam(self.group)[2] mode = 0o660 except (KeyError, IndexError): gid = -1 mode = 0o666 self.listening = reactor.listenUNIX(address=self.path, factory=CallControlFactory(self)) # Make it writable only to the SIP proxy group members try: os.chown(self.path, -1, gid) os.chmod(self.path, mode) except OSError: log.warning("Couldn't set access rights for %s" % self.path) log.warning("OpenSIPS may not be able to communicate with us!") # Then setup the CallsMonitor self.monitor = CallsMonitor(CallControlConfig.checkInterval, self) # Open the connection to the rating engines self.engines = RatingEngineConnections() def on_shutdown(self): should_close = [] if self.listening is not None: self.listening.stopListening() if self.engines is not None: should_close.append(self.engines.shutdown()) if self.monitor is not None: self.monitor.shutdown() d = defer.DeferredList(should_close) d.addBoth(self._save_calls) return d def _save_calls(self, result): if self.calls: log.info('Saving calls') calls_file = process.runtime.file(backup_calls_file) try: f = open(calls_file, 'wb') except: pass else: for call in list(self.calls.values()): call.application = None # we will mark timers with 'running' or 'idle', depending on their current state, # to be able to correctly restore them later (Timer objects cannot be pickled) if call.timer is not None: if call.inprogress: call.timer.cancel() call.timer = 'running' # temporary mark that this timer was running else: call.timer = 'idle' # temporary mark that this timer was not running failed_dump = False try: try: pickle.dump(self.calls, f) except Exception as e: log.warning('Failed to dump call list: %s', e) failed_dump = True finally: f.close() if failed_dump: unlink(calls_file) else: log.info("Saved calls: %s" % str(list(self.calls.keys()))) self.calls = {} def _restore_calls(self): calls_file = process.runtime.file(backup_calls_file) try: f = open(calls_file, 'r') except: pass else: try: self.calls = pickle.load(f) except Exception as e: log.warning('Failed to load calls saved in the previous session: %s', e) f.close() unlink(calls_file) if self.calls: log.info("Restoring calls saved previously: %s" % str(list(self.calls.keys()))) # the calls in the 2 sets below are never overlapping because closed and terminated # calls have different database fingerprints. so the dictionary update below is safe try: db = RadiusDatabase() try: terminated = db.query(RadiusDatabase.RadiusTask(None, 'terminated', calls=self.calls)) # calls terminated by caller/called didtimeout = db.query(RadiusDatabase.RadiusTask(None, 'timedout', calls=self.calls)) # calls closed by mediaproxy after a media timeout finally: db.close() except RadiusDatabaseError as e: log.error("Could not query database: %s" % e) else: for callid, call in list(self.calls.items()): callinfo = terminated.get(callid) or didtimeout.get(callid) if callinfo: # call already terminated or did timeout in mediaproxy del self.calls[callid] callinfo['call'] = call call.timer = None continue # close all calls that were already terminated or did timeout count = 0 for callinfo in list(terminated.values()): call = callinfo.get('call') if call is not None: call.end(calltime=callinfo['duration']) count += 1 for callinfo in list(didtimeout.values()): call = callinfo.get('call') if call is not None: call.end(sendbye=True) count += 1 if count > 0: log.info("Removed %d already terminated call%s" % (count, 's'*(count!=1))) for callid, call in list(self.calls.items()): call.application = self if call.timer == 'running': now = time.time() remain = call.starttime + call.timelimit - now if remain < 0: call.timelimit = int(round(now - call.starttime)) remain = 0 call._setup_timer(remain) call.timer.start() elif call.timer == 'idle': call._setup_timer() # also restore users table self.users.setdefault(call.billingParty, []).append(callid) class Request(object): """A request parsed into a structure based on request type""" __methods = {'init': ('callid', 'diverter', 'ruri', 'sourceip', 'from'), 'start': ('callid', 'dialogid'), 'stop': ('callid',), 'debug': ('show',), 'terminate': ('callid',)} def __init__(self, cmd, params): if cmd not in list(self.__methods.keys()): raise InvalidRequestError("Unknown request: %s" % cmd) try: parameters = dict(re.split(r':\s+', l, 1) for l in params) except ValueError: raise InvalidRequestError("Badly formatted request") for p in self.__methods[cmd]: try: parameters[p] except KeyError: raise InvalidRequestError("Missing %s from request" % p) self.cmd = cmd self.deferred = defer.Deferred() self.__dict__.update(parameters) try: getattr(self, '_RE_%s' % self.cmd)() except AttributeError: pass def _RE_init(self): self.from_ = self.__dict__['from'] if self.cmd=='init' and self.diverter.lower()=='none': self.diverter = None try: self.prepaid except AttributeError: self.prepaid = None else: if self.prepaid.lower() == 'true': self.prepaid = True elif self.prepaid.lower() == 'false': self.prepaid = False else: self.prepaid = None try: self.call_limit = int(self.call_limit) except (AttributeError, ValueError): self.call_limit = None else: if self.call_limit <= 0: self.call_limit = None try: self.call_token except AttributeError: self.call_token = None else: if not self.call_token or self.call_token.lower() == 'none': self.call_token = None try: self.sip_application except AttributeError: self.sip_application = '' def _RE_debug(self): if self.show == 'session': try: if not self.callid: raise InvalidRequestError("Missing callid from request") except AttributeError: raise InvalidRequestError("Missing callid from request") elif self.show == 'sessions': try: self.user except AttributeError: self.user = None else: raise InvalidRequestError("Illegal value for 'show' attribute in request") def __str__(self): if self.cmd == 'init': return "%(cmd)s: callid=%(callid)s from=%(from_)s ruri=%(ruri)s diverter=%(diverter)s sourceip=%(sourceip)s prepaid=%(prepaid)s call_limit=%(call_limit)s" % self.__dict__ elif self.cmd == 'start': return "%(cmd)s: callid=%(callid)s dialogid=%(dialogid)s" % self.__dict__ elif self.cmd == 'stop': return "%(cmd)s: callid=%(callid)s" % self.__dict__ elif self.cmd == 'debug': return "%(cmd)s: show=%(show)s" % self.__dict__ elif self.cmd == 'terminate': return "%(cmd)s: callid=%(callid)s" % self.__dict__ else: return object.__str__(self) diff --git a/callcontrol/rating/__init__.py b/callcontrol/rating/__init__.py index 224f786..6f7c7e1 100644 --- a/callcontrol/rating/__init__.py +++ b/callcontrol/rating/__init__.py @@ -1,385 +1,364 @@ """Rating engine interface implementation.""" import random from collections import deque from application.configuration import ConfigSection, ConfigSetting from application.configuration.datatypes import EndpointAddress from application import log from application.python.types import Singleton from twisted.internet.protocol import ReconnectingClientFactory from twisted.internet.error import TimeoutError from twisted.internet import reactor, defer, protocol #from twisted.protocols.basic import LineOnlyReceiver from twisted.python import failure from callcontrol import configuration_file ## ## Rating engine configuration ## class ThorNodeConfig(ConfigSection): __cfgfile__ = configuration_file __section__ = 'ThorNetwork' enabled = False class RatingConfig(ConfigSection): __cfgfile__ = configuration_file __section__ = 'CDRTool' timeout = 500 class TimeLimit(int): """A positive time limit (in seconds) or None""" def __new__(typ, value): if value.lower() == 'none': return None try: limit = int(value) except: raise ValueError("invalid time limit value: %r" % value) if limit < 0: raise ValueError("invalid time limit value: %r. should be positive." % value) return limit class CallControlConfig(ConfigSection): __cfgfile__ = configuration_file __section__ = 'CallControl' prepaid_limit = ConfigSetting(type=TimeLimit, value=None) limit = ConfigSetting(type=TimeLimit, value=None) class RatingError(Exception): pass class RatingEngineError(RatingError): pass class RatingEngineTimeoutError(TimeoutError): pass -# class RatingRequest(str): class RatingRequest(bytes): - """ - def __init__(self, command, reliable=True, **kwargs): - self.command = command.encode() - self.reliable = reliable - self.kwargs = kwargs - self.deferred = defer.Deferred() - - def __new__(cls, command, reliable=True, **kwargs): - reqstr = command + (kwargs and (' ' + ' '.join("%s=%s" % (name, value) for name, value in list(kwargs.items()))) or '') - obj = str.__new__(cls, reqstr) - obj = super().__new__(cls, reqstr.encode('utf-8')) - print('RatingRequest object type: ', type(obj)) - return obj - """ def __new__(cls, command, reliable=True, **kwargs): reqstr = command + (kwargs and (' ' + ' '.join("%s=%s" % (name, value) for name, value in list(kwargs.items()))) or '') obj = super().__new__(cls, reqstr.encode('utf-8')) obj.command = command obj.reliable = reliable obj.kwargs = kwargs obj.deferred = defer.Deferred() return obj def __init__(self, *args, **kwargs): super(RatingRequest, self).__init__() -#class RatingEngineProtocol(LineOnlyReceiver): -class RatingEngineProtocol(protocol.Protocol): -# delimiter = '\n\n' +class RatingEngineProtocol(protocol.Protocol, TimeoutMixin): delimiter = b'\r\n' def __init__(self): self.connected = False self.__request = None self.__timeout_call = None self._request_queue = deque() def connectionMade(self): log.info("Connected to Rating Engine at %s:%d" % (self.transport.getPeer().host, self.transport.getPeer().port)) self.connected = True self.factory.application.connectionMade(self.transport.connector) if self._request_queue: self._send_next_request() def connectionLost(self, reason=None): log.info("Disconnected from Rating Engine at %s:%d" % (self.transport.getPeer().host, self.transport.getPeer().port)) self.connected = False if self.__request is not None: if self.__request.reliable: self._request_queue.appendleft(self.__request) self.__request = None else: self._respond("Connection with the Rating Engine is down: %s" % reason, success=False) self.factory.application.connectionLost(self.transport.connector, reason, self) # def timeoutConnection(self): # log.info("Connection to Rating Engine at %s:%d timed out" % (self.transport.getPeer().host, self.transport.getPeer().port)) # self.transport.loseConnection() def dataReceived(self, data): #log.debug('Rating response from %s:%S: %s' % (self.transport.getPeer().host, self.transport.getPeer().port, data)) self.lineReceived(data.decode()) def lineReceived(self, line): log.debug('Received response from rating engine %s:%s: %s' % (self.transport.getPeer().host, self.transport.getPeer().port, line.strip().replace("\n", " "))) if not line: return if self.__timeout_call is not None: self.__timeout_call.cancel() if self.__request is None: log.warning('Got reply for non-existing request: %s' % line) return try: self._respond(getattr(self, '_PE_%s' % self.__request.command.lower())(line)) except AttributeError: self._respond("Unknown command in request. Cannot handle reply. Reply is: %s" % line, success=False) except Exception as e: self._respond(str(e), success=False) def _PE_maxsessiontime(self, line): -# import pdb; pdb.set_trace() lines = line.splitlines() try: limit = lines[0].strip().capitalize() except IndexError: raise ValueError("Empty reply from rating engine") try: limit = int(limit) except: if limit == 'None': limit = None elif limit == 'Locked': pass else: raise ValueError("rating engine limit must be a positive number, None or Locked: got '%s' from %s:%s" % (limit, self.transport.getPeer().host, self.transport.getPeer().port)) else: if limit < 0: raise ValueError("rating engine limit must be a positive number, None or Locked: got '%s' from %s:%s" % (limit, self.transport.getPeer().host, self.transport.getPeer().port)) data_list = [line.split('=', 1) for line in lines[1:]] info = {} for elem in data_list: try: info[elem[0]] = elem[1] except IndexError: pass if 'type' in info: type = info['type'].lower() if type == 'prepaid': prepaid = True elif type == 'postpaid': prepaid = False else: raise ValueError("prepaid must be either True or False: got '%s' from %s:%s" % (type, self.transport.getPeer().host, self.transport.getPeer().port)) else: prepaid = limit is not None return limit, prepaid def _PE_debitbalance(self, line): -# import pdb; pdb.set_trace() valid_answers = ('Ok', 'Failed', 'Not prepaid') lines = line.splitlines() try: result = lines[0].strip().capitalize() except IndexError: raise ValueError("Empty reply from rating engine %s:%s", (self.transport.getPeer().host, self.transport.getPeer().port)) if result not in valid_answers: log.error("Invalid reply from rating engine: got '%s' from %s:%s" % (lines[0].strip(), self.transport.getPeer().host, self.transport.getPeer().port)) log.warning('Rating engine possible failed query: %s', self.__request) raise RatingEngineError('Invalid rating engine response') elif result == 'Failed': log.warning('Rating engine failed query: %s', self.__request) raise RatingEngineError('Rating engine failed query') else: try: timelimit = int(lines[1].split('=', 1)[1].strip()) totalcost = lines[2].strip() except: log.error("Invalid reply from rating engine for DebitBalance on lines 2, 3: got '%s' from %s:%s" % ("', `".join(lines[1:3]), self.transport.getPeer().host, self.transport.getPeer().port)) timelimit = None totalcost = 0 return timelimit, totalcost def _send_next_request(self): if self.connected: self.__request = self._request_queue.popleft() self.delimiter = b'\r\n' log.debug("Send request to rating engine %s:%s: %s" % (self.transport.getPeer().host, self.transport.getPeer().port, self.__request.decode())) -# self.sendLine(self.__request) self.transport.write(self.__request) #self._set_timeout() self._set_timeout(self.factory.timeout) log.debug('Sent request to rating engine: %s', self.__request.decode()) else: self.__request = None def _respond(self, result, success=True): if self.__request is not None: req = self.__request self.__request = None try: if success: req.deferred.callback(result) else: req.deferred.errback(failure.Failure(RatingEngineError(result))) except defer.AlreadyCalledError: log.debug('Request %s was already responded to', req) if self._request_queue: self._send_next_request() def _set_timeout(self, timeout=None): if timeout is None: timeout = self.factory.timeout # self.__timeout_call = reactor.callLater(timeout/1000, self.timeoutConnection) def send_request(self, request): if not request.reliable and not self.connected: request.deferred.errback(failure.Failure(RatingEngineError("Connection with the Rating Engine is down"))) return request self._request_queue.append(request) if self.__request is None: self._send_next_request() return request class RatingEngineFactory(ReconnectingClientFactory): protocol = RatingEngineProtocol timeout = RatingConfig.timeout # reconnect parameters maxDelay = 15 factor = maxDelay initialDelay = 1.0/factor delay = initialDelay def __init__(self, application): self.application = application def buildProtocol(self, addr): self.resetDelay() return ReconnectingClientFactory.buildProtocol(self, addr) def clientConnectionFailed(self, connector, reason): if self.application.disconnecting: return ReconnectingClientFactory.clientConnectionFailed(self, connector, reason) def clientConnectionLost(self, connector, reason): if self.application.disconnecting: return ReconnectingClientFactory.clientConnectionLost(self, connector, reason) class DummyRatingEngine(object): def getCallLimit(self, call, max_duration=CallControlConfig.prepaid_limit, reliable=True): return defer.fail(failure.Failure(RatingEngineError("Connection with the Rating Engine not yet established"))) def debitBalance(self, call, reliable=True): return defer.fail(failure.Failure(RatingEngineError("Connection with the Rating Engine not yet established"))) class RatingEngine(object): def __init__(self, address): self.address = address self.disconnecting = False self.connector = reactor.connectTCP(self.address[0], self.address[1], factory=RatingEngineFactory(self)) self.connection = None self.__unsent_req = deque() def shutdown(self): self.disconnecting = True self.connector.disconnect() def connectionMade(self, connector): self.connection = connector.transport self.connection.protocol._request_queue.extend(self.__unsent_req) for req in self.__unsent_req: log.debug('Re-queueing request for the rating engine: %s', req) self.__unsent_req.clear() def connectionLost(self, connector, reason, protocol): while protocol._request_queue: req = protocol._request_queue.pop() if not req.reliable: log.debug('Request is considered failed: %s', req) req.deferred.errback(failure.Failure(RatingEngineError("Connection with the Rating Engine is down"))) else: log.debug('Saving request to be requeued later: %s', req) self.__unsent_req.appendleft(req) self.connection = None def getCallLimit(self, call, max_duration=CallControlConfig.prepaid_limit, reliable=True): - #import pdb; pdb.set_trace() max_duration = max_duration or CallControlConfig.limit or 36000 args = {} if call.inprogress: args['State'] = 'Connected' req = RatingRequest('MaxSessionTime', reliable=reliable, CallId=call.callid, From=call.billingParty, To=call.ruri, Gateway=call.sourceip, Duration=max_duration, Application=call.sip_application, **args) if self.connection is not None: return self.connection.protocol.send_request(req).deferred else: self.__unsent_req.append(req) return req.deferred def debitBalance(self, call, reliable=True): req = RatingRequest('DebitBalance', reliable=reliable, CallId=call.callid, From=call.billingParty, To=call.ruri, Gateway=call.sourceip, Duration=call.duration, Application=call.sip_application) if self.connection is not None: return self.connection.protocol.send_request(req).deferred else: self.__unsent_req.append(req) return req.deferred class RatingEngineAddress(EndpointAddress): default_port = 9024 name = 'rating engine address' class RatingEngineConnections(object): __metaclass__ = Singleton def __init__(self): self.user_connections = {} if not ThorNodeConfig.enabled: from callcontrol.rating.backends.opensips import OpensipsBackend self.backend = OpensipsBackend() else: from callcontrol.rating.backends.sipthor import SipthorBackend self.backend = SipthorBackend() @staticmethod def getConnection(call=None): engines = RatingEngineConnections() try: conn = random.choice(engines.backend.connections) except IndexError: return DummyRatingEngine() if call is None: return conn return engines.user_connections.setdefault(call.billingParty, conn) def remove_user(self, user): try: del self.user_connections[user] except KeyError: pass def shutdown(self): return defer.maybeDeferred(self.backend.shutdown)