def thrash_tnetstring():
    for obj, tns, json, msh in TESTS:
        if TEST_DUMP_ONLY:
            tnetstring.dumps(obj)
        elif TEST_LOAD_ONLY:
            assert tnetstring.loads(tns) == obj
        else:
            assert tnetstring.loads(tnetstring.dumps(obj)) == obj

	def _send_to_zmq(self, content, channel):
		self._lock.acquire()
		if self._push_sock:
			content[_ensure_utf8('channel')] = channel
			self._push_sock.send(tnetstring.dumps(content))
		else:
			self._pub_controller.publish(channel,
					tnetstring.dumps(content))
		self._lock.release()

 def test_unicode_handling(self):
     self.assertRaises(ValueError,tnetstring.dumps,u"hello")
     self.assertEquals(tnetstring.dumps(u"hello","utf8"),"5:hello,")
     self.assertEquals(type(tnetstring.loads("5:hello,")),str)
     self.assertEquals(type(tnetstring.loads("5:hello,","utf8")),unicode)
     ALPHA = u"\N{GREEK CAPITAL LETTER ALPHA}lpha"
     self.assertEquals(tnetstring.dumps(ALPHA,"utf8"),"6:"+ALPHA.encode("utf8")+",")
     self.assertEquals(tnetstring.dumps(ALPHA,"utf16"),"12:"+ALPHA.encode("utf16")+",")
     self.assertEquals(tnetstring.loads("12:\xff\xfe\x91\x03l\x00p\x00h\x00a\x00,","utf16"),ALPHA)

 def __init__ (self, sip_comm, sip_data, rtp_flow):
     for uuid, cnt in sip_comm.iteritems ():
         path = "%s/%s/"%(main_path, uuid)
         print 'Dumping SIP uuid:\'%s\' (%d packets) in %s'%(uuid, cnt, path)
         dialog, arr = sip_data [uuid]
         i = 0
         filearr = []
         try:
             os.makedirs (path)
         except OSError, e:
             # directory might already exist
             pass
         listpath = "%s/list.txt"%path
         listfd = open (listpath, "w+")
         listfd.write (
             "# This file contains the listing of %d's dumped packets\n"%cnt)
         listfd_legs = []
         for timestamp, addr, pkt, legs_n in arr:
             if not listfd_legs:
                 timestamppath = "%s/timestamp.txt"%path
                 timestampfd = open (timestamppath, "w+")
                 timestampfd.write (str (timestamp))
                 timestampfd.close ()
                 
             if legs_n in listfd_legs:
                 continue
             listfd_legs += [legs_n]
             src, dst = (addr [0], addr [1]), (addr [2], addr [3])
             listfd.write ("# legs_%d from %s to %s\n"%(legs_n,
                                                      str (src),
                                                      str (dst)))
         for timestamp, addr, pkt, legs_n in arr:
             src, dst = (addr [0], addr [1]), (addr [2], addr [3])
             dirpath = "%s/legs_%d"%(path, legs_n)
             filename = "%d.txt"%i
             filepath = "%s/%s"%(dirpath, filename)
             filearr += [tnetstring.dumps ([filepath, timestamp, 
                                            list (addr)])]
             listfd.write ("legs_%d/%d.txt\n"%(legs_n, i))
             try:
                 os.makedirs (dirpath)
             except OSError, e:
                 # directory might already exist
                 pass
             fd = open (filepath, "w+")
             fd.write (tnetstring.dumps ([timestamp, list (addr), str (pkt)],
                                         'iso-8859-15'))
             fd.close ()
             i += 1

	def _respond_error(self, condition):
		resp = dict()
		if self.req_id:
			resp['id'] = self.req_id
		resp['success'] = False
		resp['condition'] = ensure_utf8(condition)
		self.rep_sock.send(tnetstring.dumps(resp))

 def server_rtp_create (_redis, ckey, s_id, port, bind_ip, rr):
     _redis.lpush (redis_server_inst%rr,
                   tnetstring.dumps (["CREATE_RTP_SOCK",
                                      [s_id,
                                       bind_ip, port], ckey],
                                     'iso-8859-15'))
     return

	def _respond(self, value):
		resp = dict()
		if self.req_id:
			resp['id'] = self.req_id
		resp['success'] = True
		resp['value'] = ensure_utf8(value)
		self.rep_sock.send(tnetstring.dumps(resp))

	def publish(self, channel, id, prev_id, rheaders=None, rbody=None, sbody=None, code=None, reason=None):
		self.lock.acquire()

		if self.pubs is None:
			self.pubs = list()
			for p in self.proxies:
				pub = gripcontrol.GripPubControl(p['control_uri'])
				if 'control_iss' in p:
					pub.set_auth_jwt({'iss': p['control_iss']}, p['key'])
				self.pubs.append(pub)

		formats = list()
		if rbody is not None:
			formats.append(gripcontrol.HttpResponseFormat(code=code, reason=reason, headers=rheaders, body=rbody))
		if sbody is not None:
			formats.append(gripcontrol.HttpStreamFormat(sbody))

		item = pubcontrol.Item(formats, id, prev_id)

		zitem = item.export()
		zitem['channel'] = channel
		zitem['http-response'] = convert_json_transport(zitem['http-response'])
		self.sock.send(tnetstring.dumps(zitem))

		for pub in self.pubs:
			pub.publish_async(channel, item)

		self.lock.release()

 def server_sip_create (_redis, ckey, s_id, bind_ip, port, rr, transport):
     _redis.lpush (redis_server_inst%rr,
                   tnetstring.dumps (["CREATE_SIP%s_SOCK"%transport,
                                      [s_id,
                                       bind_ip, port], ckey],
                                     'iso-8859-15'))
     return

	def _send_to_zmq(self, channel, item):
		self._lock.acquire()
		if self._zmq_pub_controller:
			channel = _ensure_utf8(channel)
			content = item.export(True, True)
			self._zmq_pub_controller.publish(channel,
					tnetstring.dumps(content))
		self._lock.release()

	def test_send_to_zmq(self):
		client = ZmqPubControlClientTestClass2('uri')
		client._push_sock = ZmqSocketTestClass()
		client._send_to_zmq({'content'.encode('utf-8'):
				'content'.encode('utf-8')}, 'chan'.encode('utf-8'))
		self.assertEqual(client._push_sock.send_data, tnetstring.dumps(
				{'content'.encode('utf-8'): 'content'.encode('utf-8'),
				'channel'.encode('utf-8'): 'chan'.encode('utf-8')}))
		client = ZmqPubControlClientTestClass2('uri')
		client._pub_controller = ZmqPubControllerTestClass()
		client._send_to_zmq({'content'.encode('utf-8'):
				'content'.encode('utf-8')}, 'chan'.encode('utf-8'))
		self.assertEqual(client._pub_controller.publish_channel,
				'chan'.encode('utf-8'))
		self.assertEqual(client._pub_controller.publish_content,
				tnetstring.dumps({'content'.encode('utf-8'):
				'content'.encode('utf-8')}))

def add_test(v):
    #  These modules have a few round-tripping problems...
    try:
        assert cjson.decode(cjson.encode(v)) == v
        assert yajl.loads(yajl.dumps(v)) == v
    except Exception:
        pass
    else:
        TESTS.append((v,tnetstring.dumps(v),cjson.encode(v)))

 def makePickle(self, record):
     """ Create message in TNetStrings format.
     """
     msg = {}
     self.format(record) # render 'message' attribute and others
     for k in self.send_fields:
         msg[k] = record.__dict__[k]
     tnetstr = tnetstrings.dumps(msg)
     return tnetstr

	def call(self, method, args=dict()):
		req = dict()
		req['id'] = str(uuid.uuid4())
		req['method'] = ensure_utf8(method)
		req['args'] = ensure_utf8(args)
		req_raw = tnetstring.dumps(req)
		try:
			if not self.sock.poll(30000, zmq.POLLOUT):
				raise CallError('send-timeout')
			m_list = list()
			m_list.append('')
			m_list.append(req_raw)
			self.sock.send_multipart(m_list)
		except zmq.ZMQError as e:
			raise CallError('send-failed', e.message)
		start = int(time.clock() * 1000)
		while True:
			elapsed = max(int(time.clock() * 1000) - start, 0)
			try:
				if not self.sock.poll(max(30000 - elapsed, 0), zmq.POLLIN):
					raise CallError('receive-timeout')
				m_list = self.sock.recv_multipart()
			except zmq.ZMQError as e:
				raise CallError('receive-failed', e.message)
			if len(m_list) != 2:
				print "response has wrong number of parts, skipping"
				continue
			if len(m_list[0]) > 0:
				print "response first part is not empty, skipping"
				continue
			resp_raw = m_list[1]
			try:
				resp = tnetstring.loads(resp_raw)
			except:
				print "failed to parse response as tnetstring, skipping"
				continue
			if 'id' not in resp:
				print "response missing id field, skipping"
				continue
			if resp['id'] != req['id']:
				print "unexpected response id, skipping"
				continue
			break
		if 'success' not in resp:
			raise CallError('invalid-response', 'missing success field')
		if resp['success']:
			if 'value' not in resp:
				raise CallError('invalid-response', 'missing value field')
			return resp['value']
		else:
			if 'condition' not in resp:
				raise CallError('invalid-response', 'missing condition field')
			raise CallError(resp['condition'])

def keepalive_worker():
	lock.acquire()
	for id, s in sessions.iteritems():
		resp = dict()
		resp['from'] = client_id
		resp['id'] = id
		resp['seq'] = s.out_seq
		s.out_seq += 1
		resp['type'] = 'credit'
		resp['credits'] = 0

		print 'OUT %s' % resp
		out_sock.send(s.to_address + ' T' + tnetstring.dumps(resp))
	lock.release()

 def pkt_send (self, _id, data, host, port, rtp = False):
     if not rtp:
         self._last_packet ['_id'] = _id
         self._last_packet ['data'] = data
         self._last_packet ['host'] = host
         self._last_packet ['port'] = port
     try:
         self._redis.lpush ('send_%s'%str (_id),
                            tnetstring.dumps ([host, port, data],
                                              'iso-8859-15'))
         if self._debug:
             print "===> sending to %s:%d at %s"%(host, int (port),
                                                  time.ctime (time.time ()))
             print str (data)
     except:
         return True
     return False

	def _discover_uris(self):
		self._thread_cond.acquire()
		if self._discovery_in_progress:
			self._thread_cond.wait()
			self._thread_cond.release()
			self._verify_discovered_uris()
			return
		else:
			self._discovery_in_progress = True
			self._thread_cond.release()
		if (self.uri is None or self._discovery_completed or
				(self.pub_uri and self.push_uri)):
			self._end_discovery(False)
			return
		sock = self._context.socket(zmq.REQ)
		sock.linger = 0
		sock.connect(self.uri)
		start = int(timeit.default_timer() * 1000)
		if not sock.poll(3000, zmq.POLLOUT):
			sock.close()
			self._end_discovery(False)
			raise ValueError('uri discovery request failed: pollout timeout')
		req = {_ensure_utf8('method'): _ensure_utf8('get-zmq-uris')}
		sock.send(tnetstring.dumps(req))
		elapsed = max(int(timeit.default_timer() * 1000) - start, 0)
		if not sock.poll(max(3000 - elapsed, 0), zmq.POLLIN):
			sock.close()
			self._end_discovery(False)
			raise ValueError('uri discovery request failed: pollin timeout')
		resp = tnetstring.loads(sock.recv())
		sock.close()
		if (not resp.get(_ensure_utf8('success')) or
				not resp.get(_ensure_utf8('value'))):
			self._end_discovery(False)
			raise ValueError('uri discovery request failed: %s' % resp)
		self._set_discovered_uris(resp[_ensure_utf8('value')])
		self._end_discovery(True)
		self._verify_discovered_uris()

 def _handle_sip (self, date, data, addr, s_id):
     try:
         sip_pkt = SIPpacket ()
         sip_pkt.fromString (data)
         self._logger.debug (
             "%s: received SIP message (%d bytes)"% (addr, len (data)))
         redis_data = tnetstring.dumps ([s_id, date, list (addr), data],
                                        'iso-8859-15')
         valid = True
         try:
             # This is the SIP parser routine
             # There is no totag and it is a request : push in 'method's list
             # or compute session_id : 'fromtag:callid:branch'
             # when method is BYE 'fromtag' from 'To' and branch from 'CSeq'
             sip_pkt.to.enable ()
             tp = sip_pkt.tp
             if not tp and None is not sip_pkt.rm:
                 return [sip_pkt.rm, redis_data]
             if "BYE" == sip_pkt.rm:
                 sip_pkt.to.enable ()
                 fp = sip_pkt.tp
                 branch = sip_pkt.cseq.seq
             else:
                 sip_pkt._from.enable ()
                 fp = sip_pkt.fp
                 sip_pkt.via.enable ()
                 branch = sip_pkt.vb
             callid = sip_pkt.callid
             sess_id = "%s:%s:%s"% (fp, callid, branch)
             return [sess_id, redis_data]
         except KeyError, h:
             self._logger.warn (
                 "Missing header '%s', droping packet"%str (h))
     except Exception, e:
         self._logger.exception (e)
         self._logger.error (
             "Unable to parse SIP request, dumping")
         self._dump (data, suffix = "error")

		headers = list()
		headers.append(['Content-Type', 'text/plain'])
		headers.append(['Content-Length', str(os.stat(filename).st_size)])
		resp['headers'] = headers
		if len(body) > 0:
			resp['body'] = body
		if not eof:
			resp['more'] = True

		s.out_seq += 1

		if eof:
			del sessions[id]

		print 'OUT %s' % resp
		out_sock.send(req['from'] + ' ' + tnetstring.dumps(resp))
	elif socks.get(in_stream_sock) == zmq.POLLIN:
		parts = in_stream_sock.recv_multipart()
		req = tnetstring.loads(parts[1])
		print 'IN stream %s' % req

		# we are only streaming output, so subsequent input messages must be credits
		if 'type' not in req or req['type'] != 'credit':
			print 'wrong packet type'
			continue

		id = req['id']

		s = sessions.get(id)
		if s is None:
			continue

 def server_rtp_create_reply (_redis, ckey, s_id):
     _redis.lpush (redis_client_reply%ckey,
                   tnetstring.dumps (["CREATE_RTP_REPLY",
                                      [s_id]],
                                     'iso-8859-15'))
     return