Raju, please replase the file rtphandler.py. Regards, Gonzalo. along lines up i is any through passing NAT_IP1:NAT_PORT1, PUBLIC experienced. changing arrives, IP "The

-- Gonzalo J. Sambucaro

''' Handle media stream relaying and provide control ports to receive and process requests from SER (SIP Express Router) or the proxydispatcher. Copyright 2003-2005 Dan Pascu ''' import asyncore import socket import sys, os, grp, errno, traceback, atexit import time import struct import re import weakref from utilities import * from datatypes import * from version import release from request import Request, InvalidRequestError from configuration import readSettings, ConfigSection, Boolean, dumpSection class ProxyConfig(ConfigSection): _dataTypes = {'socket': ControlSocket, 'proxyIP': IPAddress, 'portRange': PortRange, 'TOS': HexNumber, 'listen': NetworkAddress, 'allow': NetworkRangeList} socket = '/var/run/mediaproxy.sock' group = 'openser' proxyIP = thisHostIP ## defined in utilities portRange = (60000, 65000) TOS = 0xb8 listen = None allow = None idleTimeout = 60 holdTimeout = 60*60 forceClose = 0 ## We use this to overwrite some of the settings above on a local basis if needed readSettings('MediaProxy', ProxyConfig) #dumpSection(ProxyConfig) if ProxyConfig.allow and NetworkRange('Any') not in ProxyConfig.allow: ## Always add ourselves to the accept list myself = NetworkRange('127.0.0.1') if myself not in ProxyConfig.allow: ProxyConfig.allow.append(myself) maxDataSize = 8*1024 try: from accounting import accounting, StopRecord, StopRecordSerializer, QueuedItemProcessingThread except ImportError: warning("accounting is enabled but the accounting module is missing. accounting is not available!") accounting = Null() dispatcher = Null() class StopRecord(object): def __new__(typ, *args, **kwargs): return None else: class DispatcherNotifyThread(QueuedItemProcessingThread): ''' Notify the proxydispatcher of calls that did timeout and provide the dispatcher with the accounting information for them. ''' def __init__(self): QueuedItemProcessingThread.__init__(self, name='DispatcherNotify', file='.dispatcher-notify.dat') def process(self, record): address = record['dispatcher'] expires = record['expires'] del record['dispatcher'], record['expires'] msg = 'timeout %s %s\n' % (record['id'], StopRecordSerializer().dump(record)) s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) try: s.settimeout(5) except AttributeError: pass try: start = time.time() try: s.connect(address) s.send(msg) result = s.recv(maxDataSize).strip() except (socket.error, socket.gaierror, socket.herror, socket.timeout), why: try: motive = why[1] except IndexError: motive = why now = time.time() if now < expires: warning("couldn't notify dispatcher at %s (%s). will retry later." % (address[0], motive)) record['dispatcher'] = address record['expires'] = expires # eventually use a different queue only for failures delay = 1 duration = now - start if duration < delay: time.sleep(delay-duration) self.put(record) ## put back in the queue and process later else: error("couldn't notify dispatcher at %s (%s). giving up." % (address[0], motive)) # eventually log the record to syslog or a file return if result.lower() == 'ok': info("notified dispatcher at %s about expired session %s" % (address[0], record['id'])) elif result.lower() == 'not found': info("session %s already closed by the dispatcher at %s" % (record['id'], address[0])) else: error("failed to notify dispatcher at %s (returned `%s')." % (address[0], result)) # eventually log the record to syslog or a file finally: s.close() def notify(self, session): if not session.dispatcher: return ## no dispatcher to notify. record = StopRecord(session) record['dispatcher'] = session.dispatcher record['expires'] = time.time() + 3600 ## try for an hour self.put(record) dispatcher = DispatcherNotifyThread() dispatcher.start() ## Internal state of mediaproxy (DON'T TOUCH!!!) ## A few shortcuts, for faster access proxyIP = ProxyConfig.proxyIP minPort = ProxyConfig.portRange[0] maxPort = ProxyConfig.portRange[1] crtPort = minPort Sessions = {} ## For computing traffic through mediaproxy byteBucket = [0, 0, 0] Traffic = [0, 0, 0] ## Non public networks. This includes RFC1918 networks, localnet (127.0.0.0/8), ## broadcast networks (224.0.0.0/4) and 0.0.0.0/8 nonPublicNetworks = [ {'name': '0.0.0.0', 'value': 0x00000000L, 'mask': 0xff000000L}, {'name': '10.0.0.0', 'value': 0x0a000000L, 'mask': 0xff000000L}, {'name': '127.0.0.0', 'value': 0x7f000000L, 'mask': 0xff000000L}, {'name': '172.16.0.0', 'value': 0xac100000L, 'mask': 0xfff00000L}, {'name': '192.168.0.0', 'value': 0xc0a80000L, 'mask': 0xffff0000L}, {'name': '224.0.0.0', 'value': 0xe0000000L, 'mask': 0xf0000000L} ] rtpPayloads = {