diff --git a/sts/__init__.py b/sts/__init__.py index fb945ac..5207d47 100755 --- a/sts/__init__.py +++ b/sts/__init__.py @@ -1,2 +1,2 @@ -# -*- coding: utf-8 -*- +# -*- coding: utf-8 -*- __all__ = ['ineterr', 'inetheaders', 'inetmgr', 'inetmsg', 'inetsocket'] \ No newline at end of file diff --git a/sts/ineterr.py b/sts/ineterr.py index 2d46b51..40e1eb2 100755 --- a/sts/ineterr.py +++ b/sts/ineterr.py @@ -1,37 +1,37 @@ -# -*- coding: utf-8 -*- -import collections -from xml.etree import ElementTree - -SUCCESS = 0 -PENDING = -1 -BAD_SERVER_DATA = 4 -TIMEOUT = 42 - -Error = collections.namedtuple('Error', 'code server module line text') - -def err(code, text=''): - import inspect - frame = inspect.stack()[1] - return Error(code, 0, 0, line=frame[2], text=text) - - -def parse_error(body): - try: - elem = ElementTree.fromstring(body) - except ElementTree.ParseError: - return err(BAD_SERVER_DATA, 'parse_error, invalid XML') - - try: - return Error( - int(elem.get('code', '0')), - int(elem.get('server', '0')), - int(elem.get('module', '0')), - int(elem.get('line', '0')), - elem.get('text', '') - ) - except ValueError: - return err(BAD_SERVER_DATA, 'parse_error, invalid attribute(s)') - - -def build_error(error): +# -*- coding: utf-8 -*- +import collections +from xml.etree import ElementTree + +SUCCESS = 0 +PENDING = -1 +BAD_SERVER_DATA = 4 +TIMEOUT = 42 + +Error = collections.namedtuple('Error', 'code server module line text') + +def err(code, text=''): + import inspect + frame = inspect.stack()[1] + return Error(code, 0, 0, line=frame[2], text=text) + + +def parse_error(body): + try: + elem = ElementTree.fromstring(body) + except ElementTree.ParseError: + return err(BAD_SERVER_DATA, 'parse_error, invalid XML') + + try: + return Error( + int(elem.get('code', '0')), + int(elem.get('server', '0')), + int(elem.get('module', '0')), + int(elem.get('line', '0')), + elem.get('text', '') + ) + except ValueError: + return err(BAD_SERVER_DATA, 'parse_error, invalid attribute(s)') + + +def build_error(error): return ElementTree.Element('Error', error._asdict()).tostring() \ No newline at end of file diff --git a/sts/inetheaders.py b/sts/inetheaders.py index 10fd5f0..a29fad0 100755 --- a/sts/inetheaders.py +++ b/sts/inetheaders.py @@ -1,70 +1,70 @@ -# -*- coding: utf-8 -*- - -class Header(object): - - def __init__(self, name): - - if (isinstance(name, Header)): - name = name.normalized - - name = name.strip() - self.normalized = name.lower() - - def __hash__(self): - return hash(self.normalized) - - def __eq__(self, right): - assert isinstance(right, Header), 'Invalid Comparison' - return self.normalized == right.normalized - - def __str__(self): - return self.normalized - -ACCEPT = Header('a') -CONTENT_ENCODING = Header('e') -CONTENT_LENGTH = Header('l') -CONTENT_RANGE = Header('n') -CONTENT_TYPE = Header('c') -FROM = Header('f') -FROM_EX = Header('g') -FROM_RIGHTS = Header('h') -REFER_TO = Header('r') -REPLY_TO = Header('p') -SEQUENCE = Header('q') -STREAM = Header('m') -SUBJECT = Header('s') -TIMESTAMP = Header('z') -TO = Header('t') -TRACE = Header('i') -TRANSFER_ENCODING = Header('x') -VIA = Header('v') - - -COMPACT_HEADERS = dict([(Header(key), value) for key, value in { - 'Accept': ACCEPT, - 'Content-Encoding': CONTENT_ENCODING, - 'Content-Length': CONTENT_LENGTH, - 'Content-Range': CONTENT_RANGE, - 'Content-Type': CONTENT_TYPE, - 'From': FROM, - 'X-From-Game': FROM_EX, - 'X-From-Rights': FROM_RIGHTS, - 'Refer-To': REFER_TO, - 'Reply-To': REPLY_TO, - 'X-Sequence': SEQUENCE, - 'Stream': STREAM, - 'Subject': SUBJECT, - 'Timestamp': TIMESTAMP, - 'To': TO, - 'X-Trace-ID': TRACE, - 'Transfer-Encoding': TRANSFER_ENCODING, - 'Via': VIA -}.items()]) - - -MULTI_HEADERS = frozenset([Header(name) for name in [ - ACCEPT, 'Accept-Charset', 'Accept-Encoding', 'Accept-Language', 'Accept-Ranges', 'Allow', 'Cache-Control', - 'Connection', CONTENT_ENCODING, 'Content-Language', 'Expect', 'If-Match', 'If-None-Match', 'Pragma', - 'Proxy-Authenticate', 'Set-Cookie', 'TE', 'Trailer', TRANSFER_ENCODING, 'Upgrade', 'User-Agent', 'Vary', VIA, - 'Warning', 'WWW-Authenticate', 'X-Forwarded-For' -]]) +# -*- coding: utf-8 -*- + +class Header(object): + + def __init__(self, name): + + if (isinstance(name, Header)): + name = name.normalized + + name = name.strip() + self.normalized = name.lower() + + def __hash__(self): + return hash(self.normalized) + + def __eq__(self, right): + assert isinstance(right, Header), 'Invalid Comparison' + return self.normalized == right.normalized + + def __str__(self): + return self.normalized + +ACCEPT = Header('a') +CONTENT_ENCODING = Header('e') +CONTENT_LENGTH = Header('l') +CONTENT_RANGE = Header('n') +CONTENT_TYPE = Header('c') +FROM = Header('f') +FROM_EX = Header('g') +FROM_RIGHTS = Header('h') +REFER_TO = Header('r') +REPLY_TO = Header('p') +SEQUENCE = Header('q') +STREAM = Header('m') +SUBJECT = Header('s') +TIMESTAMP = Header('z') +TO = Header('t') +TRACE = Header('i') +TRANSFER_ENCODING = Header('x') +VIA = Header('v') + + +COMPACT_HEADERS = dict([(Header(key), value) for key, value in { + 'Accept': ACCEPT, + 'Content-Encoding': CONTENT_ENCODING, + 'Content-Length': CONTENT_LENGTH, + 'Content-Range': CONTENT_RANGE, + 'Content-Type': CONTENT_TYPE, + 'From': FROM, + 'X-From-Game': FROM_EX, + 'X-From-Rights': FROM_RIGHTS, + 'Refer-To': REFER_TO, + 'Reply-To': REPLY_TO, + 'X-Sequence': SEQUENCE, + 'Stream': STREAM, + 'Subject': SUBJECT, + 'Timestamp': TIMESTAMP, + 'To': TO, + 'X-Trace-ID': TRACE, + 'Transfer-Encoding': TRANSFER_ENCODING, + 'Via': VIA +}.items()]) + + +MULTI_HEADERS = frozenset([Header(name) for name in [ + ACCEPT, 'Accept-Charset', 'Accept-Encoding', 'Accept-Language', 'Accept-Ranges', 'Allow', 'Cache-Control', + 'Connection', CONTENT_ENCODING, 'Content-Language', 'Expect', 'If-Match', 'If-None-Match', 'Pragma', + 'Proxy-Authenticate', 'Set-Cookie', 'TE', 'Trailer', TRANSFER_ENCODING, 'Upgrade', 'User-Agent', 'Vary', VIA, + 'Warning', 'WWW-Authenticate', 'X-Forwarded-For' +]]) diff --git a/sts/inetmgr.py b/sts/inetmgr.py index 2ddb8e9..ed0a37c 100755 --- a/sts/inetmgr.py +++ b/sts/inetmgr.py @@ -1,19 +1,19 @@ -# -*- coding: utf-8 -*- -import os -from xml.etree import ElementTree -from . import inetsocket - -def connect(socket, conn_type, program, build): - s = inetsocket.Socket(socket) - - root = ElementTree.Element('Connect') - ElementTree.SubElement(root, 'ConnType').text = str(conn_type) - ElementTree.SubElement(root, 'Program').text = str(program) - ElementTree.SubElement(root, 'Build').text = str(build) - ElementTree.SubElement(root, 'Process').text = str(os.getpid()) - ElementTree.SubElement(root, 'ProductType').text = '1000' - ElementTree.SubElement(root, 'AppIndex').text = '1' - ElementTree.SubElement(root, 'Address').text = socket.getsockname()[0] - - s.send('Sts', 'Connect', body=ElementTree.tostring(root)) - return s +# -*- coding: utf-8 -*- +import os +from xml.etree import ElementTree +from . import inetsocket + +def connect(socket, conn_type, program, build): + s = inetsocket.Socket(socket) + + root = ElementTree.Element('Connect') + ElementTree.SubElement(root, 'ConnType').text = str(conn_type) + ElementTree.SubElement(root, 'Program').text = str(program) + ElementTree.SubElement(root, 'Build').text = str(build) + ElementTree.SubElement(root, 'Process').text = str(os.getpid()) + ElementTree.SubElement(root, 'ProductType').text = '1000' + ElementTree.SubElement(root, 'AppIndex').text = '1' + ElementTree.SubElement(root, 'Address').text = socket.getsockname()[0] + + s.send('Sts', 'Connect', body=ElementTree.tostring(root)) + return s diff --git a/sts/inetmsg.py b/sts/inetmsg.py index 2cb91fe..2ed0819 100755 --- a/sts/inetmsg.py +++ b/sts/inetmsg.py @@ -1,98 +1,98 @@ -# -*- coding: utf-8 -*- -from . import inetheaders -import collections - -PROTOCOL_VERSION = 'STS/1.0' - - -class ParseError(Exception): - """Invalid message content.""" - - -RequestLine = collections.namedtuple('RequestLine', 'method uri') -StatusLine = collections.namedtuple('StatusLine', 'code reason') - - -def _read_start_line(fp): - line = fp.readline().strip().split(' ') - if len(line) != 3: - raise ParseError('start line, invalid format') - - if line[0] == PROTOCOL_VERSION: - return StatusLine(code=int(line[1]), reason=line[2]) - elif line[2] == PROTOCOL_VERSION: - return RequestLine(method=line[0], uri=line[1]) - else: - raise ParseError('start line, invalid protocol version') - - -def _parse_header(line): - name, value = line.split(':', 1) - name = inetheaders.Header(name.strip()) - - if name in inetheaders.COMPACT_HEADERS: - name = inetheaders.COMPACT_HEADERS[name] - - value = value.strip() - if name in inetheaders.MULTI_HEADERS: - value = [value.strip() for value in value.split(',')] - - return name, value - - -def _read_headers(fp): - headers = {} - - while True: - line = fp.readline() - if not line: - raise ParseError('headers, EOF during parsing') - if line == '\r\n': - break - - name, value = _parse_header(line) - if isinstance(value, list): - headers.get(name, []).extend(value) - else: - headers[name] = value - - return headers - - -def _read_body(fp, headers): - return fp.read(int(headers.get(inetheaders.CONTENT_LENGTH, 0))) - - -def read_message(fp): - """Read an STS message from a file-like object and return the start line, headers, and body.""" - start_line = _read_start_line(fp) - headers = _read_headers(fp) - body = _read_body(fp, headers) - - return start_line, headers, body - - -def _build_start_line(start_line): - if isinstance(start_line, StatusLine): - return '%s %d %s\r\n' % (PROTOCOL_VERSION, start_line.code, start_line.reason) - elif isinstance(start_line, RequestLine): - return '%s %s %s\r\n' % (start_line.method, start_line.uri, PROTOCOL_VERSION) - else: - raise TypeError('start_line must be a StatusLine or RequestLine') - - -def _build_header_line(header): - name, value = header - if isinstance(value, list): - value = ','.join(value) - - return '%s: %s\r\n' % (name, value) - - -def _build_header_lines(headers): - return [_build_header_line(header) for header in headers.items()] + ['\r\n'] - - -def build_message(start_line, headers, body): - """Return a string containing an STS message built using the provided start line, headers, and body.""" - return ''.join([_build_start_line(start_line)] + _build_header_lines(headers) + [body]) +# -*- coding: utf-8 -*- +from . import inetheaders +import collections + +PROTOCOL_VERSION = 'STS/1.0' + + +class ParseError(Exception): + """Invalid message content.""" + + +RequestLine = collections.namedtuple('RequestLine', 'method uri') +StatusLine = collections.namedtuple('StatusLine', 'code reason') + + +def _read_start_line(fp): + line = fp.readline().strip().split(' ') + if len(line) != 3: + raise ParseError('start line, invalid format') + + if line[0] == PROTOCOL_VERSION: + return StatusLine(code=int(line[1]), reason=line[2]) + elif line[2] == PROTOCOL_VERSION: + return RequestLine(method=line[0], uri=line[1]) + else: + raise ParseError('start line, invalid protocol version') + + +def _parse_header(line): + name, value = line.split(':', 1) + name = inetheaders.Header(name.strip()) + + if name in inetheaders.COMPACT_HEADERS: + name = inetheaders.COMPACT_HEADERS[name] + + value = value.strip() + if name in inetheaders.MULTI_HEADERS: + value = [value.strip() for value in value.split(',')] + + return name, value + + +def _read_headers(fp): + headers = {} + + while True: + line = fp.readline() + if not line: + raise ParseError('headers, EOF during parsing') + if line == '\r\n': + break + + name, value = _parse_header(line) + if isinstance(value, list): + headers.get(name, []).extend(value) + else: + headers[name] = value + + return headers + + +def _read_body(fp, headers): + return fp.read(int(headers.get(inetheaders.CONTENT_LENGTH, 0))) + + +def read_message(fp): + """Read an STS message from a file-like object and return the start line, headers, and body.""" + start_line = _read_start_line(fp) + headers = _read_headers(fp) + body = _read_body(fp, headers) + + return start_line, headers, body + + +def _build_start_line(start_line): + if isinstance(start_line, StatusLine): + return '%s %d %s\r\n' % (PROTOCOL_VERSION, start_line.code, start_line.reason) + elif isinstance(start_line, RequestLine): + return '%s %s %s\r\n' % (start_line.method, start_line.uri, PROTOCOL_VERSION) + else: + raise TypeError('start_line must be a StatusLine or RequestLine') + + +def _build_header_line(header): + name, value = header + if isinstance(value, list): + value = ','.join(value) + + return '%s: %s\r\n' % (name, value) + + +def _build_header_lines(headers): + return [_build_header_line(header) for header in headers.items()] + ['\r\n'] + + +def build_message(start_line, headers, body): + """Return a string containing an STS message built using the provided start line, headers, and body.""" + return ''.join([_build_start_line(start_line)] + _build_header_lines(headers) + [body]) diff --git a/sts/inetsocket.py b/sts/inetsocket.py index 91ee418..74c0cb6 100755 --- a/sts/inetsocket.py +++ b/sts/inetsocket.py @@ -1,287 +1,287 @@ -# -*- coding: utf-8 -*- - -import Queue as pqueue -import socket as psocket -import threading as pthread -import itertools as it -from . import inetheaders, inetmsg, ineterr - -AUTO_PING_SECONDS = 5.0 - -class Socket(object): - - def __init__(self, socket): - - self._txn_id = it.count(start = 1) - self._txn_queues = {} - - self._socket = None - self.attach_socket(socket) - - def is_closed(self): - return self._cancel.is_set() - - def detach_socket(self): - assert self._socket, 'Invalid Socket State' - - self._cancel.set() - self._read_thread.join() - self._send_thread.join() - - socket, self._socket = self._socket, None - return socket - - def attach_socket(self, socket): - assert not self._socket, 'Invalid Socket State' - - self._socket = socket - self._socket.settimeout(AUTO_PING_SECONDS) - - self._cancel = pthread.Event() - self._send_queue = pqueue.Queue() - self._send_thread = pthread.Thread(target = self._send_worker, args = (socket, self._send_queue, self._cancel)) - self._send_thread.daemon = True; - self._send_thread.start() - - self._read_queue = pqueue.Queue() - self._read_thread = pthread.Thread(target = self._read_worker, args = (socket, self._read_queue, self._cancel)) - self._read_thread.daemon = True; - self._read_thread.start() - - def _read_worker(self, socket, queue, cancel): - - fd = socket.makefile('rb') - - while (not cancel.is_set()): - - start_line = False - headers = False - body = False - - try: - start_line, headers, body = inetmsg.read_message(fd) - - except psocket.timeout: - continue - - except: - cancel.set() - continue - - if (isinstance(start_line, inetmsg.StatusLine)): - - if (not self._handle_response(start_line, headers, body, cancel)): - queue.put((start_line, headers, body), False) - - def _handle_response(self, start_line, headers, body, cancel): - - try: - subject = headers.get(inetheaders.SUBJECT, 0) - subject = subject.lower() - txn_id = subject.split(';')[0] - txn_id = int(txn_id.split('r')[0]) - - except ValueError: - cancel.set() - - if (txn_id in self._txn_queues): - self._txn_queues[txn_id].put((start_line, headers, body), False) - return True - - return False - - def _read(self, queue, timeout): - - start_line, headers, body = queue.get(timeout = timeout) - - if (isinstance(start_line, inetmsg.StatusLine)): - - if (start_line.code >= 400): - err = ineterr.parse_error(body) - - if err.code == ineterr.PENDING: - err = ineterr.err(ineterr.BAD_SERVER_DATA, '_read, received PENDING') - - body = None - - else: - err = ineterr.err(ineterr.SUCCESS) - - return err, start_line, headers, body - - def read(self, timeout = None): - - while True: - - try: - err, start_line, headers, body = self._read(self._read_queue, timeout) - - except pqueue.Empty: - return None - - if (isinstance(start_line, inetmsg.StatusLine)): - err = ineterr.err(ineterr.BAD_SERVER_DATA, 'read, received status') - - if (body is None): - continue - - try: - __, protocol, command = start_line.uri.split('/') - - except ValueError: - continue - - return (protocol, command, headers, body) - - def _send_worker(self, socket, queue, cancel): - - while (not cancel.is_set()): - - try: - start_line, headers, body = queue.get(timeout = AUTO_PING_SECONDS) - socket.sendall(inetmsg.build_message(start_line, headers, body)) - - except pqueue.Empty: - self._send(queue, inetmsg.RequestLine(method = 'P', uri='/Sts/Ping')) - - def _send(self, queue, start_line, headers = False, body = False): - - headers = headers if headers else {} - body = body if body else '' - content_type = headers.get(inetheaders.CONTENT_TYPE, 'application/xml') - - if (content_type == 'application/xml'): - body += '\n' - - if (len(body) > 0): - headers[inetheaders.CONTENT_LENGTH] = str(len(body)) - - elif (inetheaders.CONTENT_LENGTH in headers): - del headers[inetheaders.CONTENT_LENGTH] - - queue.put((start_line, headers, body)) - - def send(self, protocol, command, headers = False, body = False): - - request_line = inetmsg.RequestLine(method='P', uri='/%s/%s' % (protocol, command)) - self._send(self._send_queue, request_line, headers, body) - - def _request(self, protocol, command, headers = False, body = False, timeout = None): - - txn_id = self._txn_id.next() - self._txn_queues[txn_id] = pqueue.Queue() - - headers = headers if headers else {} - headers[inetheaders.SUBJECT] = str(txn_id) - - self.send(protocol, command, headers, body) - - try: - - body_result = False - next_seq = False - while True: - - try: - err, start_line, headers, body = self._read(self._txn_queues[txn_id], timeout) - - except pqueue.Empty: - yield ineterr.err(ineterr.TIMEOUT), None - break - - if (isinstance(start_line, inetmsg.RequestLine)): - err = ineterr.err(ineterr.BAD_SERVER_DATA, '_request, received request') - - subject = headers[inetheaders.SUBJECT].split(';')[0] - - try: - __, seq = subject.lower().split('r') - seq = int(seq) if seq else False - - except ValueError: - err = ineterr.err(ineterr.BAD_SERVER_DATA, '_request, malformed subject') - - if (not seq and next_seq): - err = ineterr.err(ineterr.BAD_SERVER_DATA, '_request, out of sequence') - elif (next_seq and seq != next_seq): - err = ineterr.err(ineterr.BAD_SERVER_DATA, '_request, out of sequence') - elif (seq and seq != 1 and not next_seq): - err = ineterr.err(ineterr.BAD_SERVER_DATA, '_request, out of sequence') - elif (seq): - next_seq = seq + 1 - - complete = 'R' in subject - if (not seq and not complete): - err = ineterr.err(ineterr.BAD_SERVER_DATA, '_request, invalid seq') - - if (err.code != ineterr.SUCCESS and not complete): - err = ineterr.err(ineterr.BAD_SERVER_DATA, '_request, pending error') - - if (inetheaders.CONTENT_RANGE in headers): - - raw_range = headers[inetheaders.CONTENT_RANGE] - type_segments = raw_range.split(' ') - range_segments = type_segments[1].split('/') - numerator_range = range_segments[0].split('-') - start_range = int(numerator_range[0]) - end_range = int(numerator_range[1]) - denominator = int(range_segments[1]) - - if (start_range == 0): - body_result = body - else: - body_result += body - - if ((end_range + 1) == denominator): - - if (not len(body_result) == denominator): - err = ineterr.err(ineterr.BAD_SERVER_DATA, '_request, invalid chunked result') - - yield err, (headers, body_result) - - continue - - if (body): - yield err, (headers, body) - else: - yield err, None - - if (err.code != ineterr.SUCCESS or complete): - break - - finally: - del self._txn_queues[txn_id] - - - def _request_one(self, protocol, command, headers, body, timeout): - - iter = self._request(protocol, command, headers, body, timeout) - err, msg = iter.next() - - if (err.code == ineterr.PENDING): - iter.close() - return ineterr.err(ineterr.BAD_SERVER_DATA, '_request_one, multi-part response'), None - - return err, msg - - def request_many(self, protocol, command, headers = False, body = False, timeout = None): - - return self._request(protocol, command, headers, body, timeout) - - def request_one(self, protocol, command, headers = False, body = False, timeout = None): - - err, msg = self._request_one(protocol, command, headers, body, timeout) - - if (err.code == ineterr.SUCCESS and not msg): - return ineterr.err(ineterr.BAD_SERVER_DATA, 'request_one, reply not message'), None - - return err, msg - - def request_none(self, protocol, command, headers = False, body = False, timeout = None): - - err, msg = self._request_one(protocol, command, headers, body, timeout) - - if (msg): - return ineterr.err(ineterr.BAD_SERVER_DATA, 'request_none, message response'), None - - return err +# -*- coding: utf-8 -*- + +import Queue as pqueue +import socket as psocket +import threading as pthread +import itertools as it +from . import inetheaders, inetmsg, ineterr + +AUTO_PING_SECONDS = 5.0 + +class Socket(object): + + def __init__(self, socket): + + self._txn_id = it.count(start = 1) + self._txn_queues = {} + + self._socket = None + self.attach_socket(socket) + + def is_closed(self): + return self._cancel.is_set() + + def detach_socket(self): + assert self._socket, 'Invalid Socket State' + + self._cancel.set() + self._read_thread.join() + self._send_thread.join() + + socket, self._socket = self._socket, None + return socket + + def attach_socket(self, socket): + assert not self._socket, 'Invalid Socket State' + + self._socket = socket + self._socket.settimeout(AUTO_PING_SECONDS) + + self._cancel = pthread.Event() + self._send_queue = pqueue.Queue() + self._send_thread = pthread.Thread(target = self._send_worker, args = (socket, self._send_queue, self._cancel)) + self._send_thread.daemon = True; + self._send_thread.start() + + self._read_queue = pqueue.Queue() + self._read_thread = pthread.Thread(target = self._read_worker, args = (socket, self._read_queue, self._cancel)) + self._read_thread.daemon = True; + self._read_thread.start() + + def _read_worker(self, socket, queue, cancel): + + fd = socket.makefile('rb') + + while (not cancel.is_set()): + + start_line = False + headers = False + body = False + + try: + start_line, headers, body = inetmsg.read_message(fd) + + except psocket.timeout: + continue + + except: + cancel.set() + continue + + if (isinstance(start_line, inetmsg.StatusLine)): + + if (not self._handle_response(start_line, headers, body, cancel)): + queue.put((start_line, headers, body), False) + + def _handle_response(self, start_line, headers, body, cancel): + + try: + subject = headers.get(inetheaders.SUBJECT, 0) + subject = subject.lower() + txn_id = subject.split(';')[0] + txn_id = int(txn_id.split('r')[0]) + + except ValueError: + cancel.set() + + if (txn_id in self._txn_queues): + self._txn_queues[txn_id].put((start_line, headers, body), False) + return True + + return False + + def _read(self, queue, timeout): + + start_line, headers, body = queue.get(timeout = timeout) + + if (isinstance(start_line, inetmsg.StatusLine)): + + if (start_line.code >= 400): + err = ineterr.parse_error(body) + + if err.code == ineterr.PENDING: + err = ineterr.err(ineterr.BAD_SERVER_DATA, '_read, received PENDING') + + body = None + + else: + err = ineterr.err(ineterr.SUCCESS) + + return err, start_line, headers, body + + def read(self, timeout = None): + + while True: + + try: + err, start_line, headers, body = self._read(self._read_queue, timeout) + + except pqueue.Empty: + return None + + if (isinstance(start_line, inetmsg.StatusLine)): + err = ineterr.err(ineterr.BAD_SERVER_DATA, 'read, received status') + + if (body is None): + continue + + try: + __, protocol, command = start_line.uri.split('/') + + except ValueError: + continue + + return (protocol, command, headers, body) + + def _send_worker(self, socket, queue, cancel): + + while (not cancel.is_set()): + + try: + start_line, headers, body = queue.get(timeout = AUTO_PING_SECONDS) + socket.sendall(inetmsg.build_message(start_line, headers, body)) + + except pqueue.Empty: + self._send(queue, inetmsg.RequestLine(method = 'P', uri='/Sts/Ping')) + + def _send(self, queue, start_line, headers = False, body = False): + + headers = headers if headers else {} + body = body if body else '' + content_type = headers.get(inetheaders.CONTENT_TYPE, 'application/xml') + + if (content_type == 'application/xml'): + body += '\n' + + if (len(body) > 0): + headers[inetheaders.CONTENT_LENGTH] = str(len(body)) + + elif (inetheaders.CONTENT_LENGTH in headers): + del headers[inetheaders.CONTENT_LENGTH] + + queue.put((start_line, headers, body)) + + def send(self, protocol, command, headers = False, body = False): + + request_line = inetmsg.RequestLine(method='P', uri='/%s/%s' % (protocol, command)) + self._send(self._send_queue, request_line, headers, body) + + def _request(self, protocol, command, headers = False, body = False, timeout = None): + + txn_id = self._txn_id.next() + self._txn_queues[txn_id] = pqueue.Queue() + + headers = headers if headers else {} + headers[inetheaders.SUBJECT] = str(txn_id) + + self.send(protocol, command, headers, body) + + try: + + body_result = False + next_seq = False + while True: + + try: + err, start_line, headers, body = self._read(self._txn_queues[txn_id], timeout) + + except pqueue.Empty: + yield ineterr.err(ineterr.TIMEOUT), None + break + + if (isinstance(start_line, inetmsg.RequestLine)): + err = ineterr.err(ineterr.BAD_SERVER_DATA, '_request, received request') + + subject = headers[inetheaders.SUBJECT].split(';')[0] + + try: + __, seq = subject.lower().split('r') + seq = int(seq) if seq else False + + except ValueError: + err = ineterr.err(ineterr.BAD_SERVER_DATA, '_request, malformed subject') + + if (not seq and next_seq): + err = ineterr.err(ineterr.BAD_SERVER_DATA, '_request, out of sequence') + elif (next_seq and seq != next_seq): + err = ineterr.err(ineterr.BAD_SERVER_DATA, '_request, out of sequence') + elif (seq and seq != 1 and not next_seq): + err = ineterr.err(ineterr.BAD_SERVER_DATA, '_request, out of sequence') + elif (seq): + next_seq = seq + 1 + + complete = 'R' in subject + if (not seq and not complete): + err = ineterr.err(ineterr.BAD_SERVER_DATA, '_request, invalid seq') + + if (err.code != ineterr.SUCCESS and not complete): + err = ineterr.err(ineterr.BAD_SERVER_DATA, '_request, pending error') + + if (inetheaders.CONTENT_RANGE in headers): + + raw_range = headers[inetheaders.CONTENT_RANGE] + type_segments = raw_range.split(' ') + range_segments = type_segments[1].split('/') + numerator_range = range_segments[0].split('-') + start_range = int(numerator_range[0]) + end_range = int(numerator_range[1]) + denominator = int(range_segments[1]) + + if (start_range == 0): + body_result = body + else: + body_result += body + + if ((end_range + 1) == denominator): + + if (not len(body_result) == denominator): + err = ineterr.err(ineterr.BAD_SERVER_DATA, '_request, invalid chunked result') + + yield err, (headers, body_result) + + continue + + if (body): + yield err, (headers, body) + else: + yield err, None + + if (err.code != ineterr.SUCCESS or complete): + break + + finally: + del self._txn_queues[txn_id] + + + def _request_one(self, protocol, command, headers, body, timeout): + + iter = self._request(protocol, command, headers, body, timeout) + err, msg = iter.next() + + if (err.code == ineterr.PENDING): + iter.close() + return ineterr.err(ineterr.BAD_SERVER_DATA, '_request_one, multi-part response'), None + + return err, msg + + def request_many(self, protocol, command, headers = False, body = False, timeout = None): + + return self._request(protocol, command, headers, body, timeout) + + def request_one(self, protocol, command, headers = False, body = False, timeout = None): + + err, msg = self._request_one(protocol, command, headers, body, timeout) + + if (err.code == ineterr.SUCCESS and not msg): + return ineterr.err(ineterr.BAD_SERVER_DATA, 'request_one, reply not message'), None + + return err, msg + + def request_none(self, protocol, command, headers = False, body = False, timeout = None): + + err, msg = self._request_one(protocol, command, headers, body, timeout) + + if (msg): + return ineterr.err(ineterr.BAD_SERVER_DATA, 'request_none, message response'), None + + return err