-- Efficient expandable one dimensional array. -- Copyright (C) 2015 Tristan Gingold -- -- GHDL is free software; you can redistribute it and/or modify it under -- the terms of the GNU General Public License as published by the Free -- Software Foundation; either version 2, or (at your option) any later -- version. -- -- GHDL is distributed in the hope that it will be useful, but WITHOUT ANY -- WARRANTY; without even the implied warranty of MERCHANTABILITY or -- FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License -- for more details. -- -- You should have received a copy of the GNU General Public License -- along with GHDL; see the file COPYING. If not, write to the Free -- Software Foundation, 59 Temple Place - Suite 330, Boston, MA -- 02111-1307, USA. -- This package mimics GNAT.Table, but: -- - the index type can be any discrete type (in particular a modular type) -- - the increment is not used -- - the interface is simplified. with Dyn_Tables; generic -- This package creates: -- array (Table_Index_Type range Table_Low_Bound .. <>) -- of Table_Component_Type; type Table_Component_Type is private; type Table_Index_Type is (<>); -- The lowest bound of the array. Note that Table_Low_Bound shouldn't be -- Table_Index_Type'First, as otherwise Last may raise constraint error -- when the table is empty. Table_Low_Bound : Table_Index_Type; -- Initial number of elements. Table_Initial : Positive; package Tables is package Dyn_Table is new Dyn_Tables (Table_Component_Type, Table_Index_Type, Table_Low_Bound); T : Dyn_Table.Instance; subtype Table_Type is Dyn_Table.Table_Type; -- Pointer to the table. Note that the use of a thin pointer to the -- largest array, this implementation bypasses Ada index checks. Table : Dyn_Table.Table_Thin_Ptr renames T.Table; -- Initialize the table. This is done automatically at elaboration. procedure Init; -- Logical bounds of the array. First : constant Table_Index_Type := Table_Low_Bound; function Last return Table_Index_Type; pragma Inline (Last); -- Deallocate all the memory. Makes the array unusable until the next -- call to Init. procedure Free; -- Increase by 1 the length of the array. This may allocate memory. procedure Increment_Last; pragma Inline (Increment_Last); -- Decrease by 1 the length of the array. procedure Decrement_Last; pragma Inline (Decrement_Last); -- Increase or decrease the length of the array by specifying the upper -- bound. procedure Set_Last (Index : Table_Index_Type); -- Append VAL to the array. This always increase the length of the array. procedure Append (Val : Table_Component_Type); pragma Inline (Append); -- Increase by NUM the length of the array, and returns the old value -- of Last + 1. function Allocate (Num : Natural := 1) return Table_Index_Type; pragma Inline (Allocate); end Tables; id='n3' href='#n3'>3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73
import threading, Queue, cStringIO
import tcp
class ServerThread(threading.Thread):
def __init__(self, server):
self.server = server
threading.Thread.__init__(self)
def run(self):
self.server.serve_forever()
def shutdown(self):
self.server.shutdown()
class ServerTestBase:
ssl = None
handler = None
@classmethod
def setupAll(cls):
cls.q = Queue.Queue()
s = cls.makeserver()
cls.port = s.port
cls.server = ServerThread(s)
cls.server.start()
@classmethod
def makeserver(cls):
return TServer(cls.ssl, cls.q, cls.handler)
@classmethod
def teardownAll(cls):
cls.server.shutdown()
@property
def last_handler(self):
return self.server.server.last_handler
class TServer(tcp.TCPServer):
def __init__(self, ssl, q, handler_klass, addr=("127.0.0.1", 0)):
"""
ssl: A {cert, key, v3_only} dict.
"""
tcp.TCPServer.__init__(self, addr)
self.ssl, self.q = ssl, q
self.handler_klass = handler_klass
self.last_handler = None
def handle_connection(self, request, client_address):
h = self.handler_klass(request, client_address, self)
self.last_handler = h
if self.ssl:
if self.ssl["v3_only"]:
method = tcp.SSLv3_METHOD
options = tcp.OP_NO_SSLv2|tcp.OP_NO_TLSv1
else:
method = tcp.SSLv23_METHOD
options = None
h.convert_to_ssl(
self.ssl["cert"],
self.ssl["key"],
method = method,
options = options,
handle_sni = getattr(h, "handle_sni", None)
)
h.handle()
h.finish()
def handle_error(self, request, client_address):
s = cStringIO.StringIO()
tcp.TCPServer.handle_error(self, request, client_address, s)
self.q.put(s.getvalue())