# -*- coding: utf-8 -*-
"""
FileName: connection
Author: Tao Hao
@contact: taohaohust@outlook.com
Created time: 2019/5/29
Description:
Changelog:
"""
import socket
import errno
from httptools import HttpRequestParser
from httptools.parser.errors import HttpParserError
import traceback
from KBEDebug import *
from multidict import CIMultiDict
from httpserver.request import Request
from httpserver.response import HTTPResponse
from httpserver.exceptions import BaseException_
from httpserver.utils import ERRNO_WOULDBLOCK, errno_from_exception
import KBEngine
from Functor import Functor
class Connection(object):
def __init__(self, sock, addr, app):
self._sock = sock
# addr = (hostaddr, port)
self._addr = addr
self._fd = self._sock.fileno()
self.parser = None
self.url = None
self.headers = None
self._header_fragment = b""
self.request = None
self.app = app
self.has_sent_size = 0
self.send_body_size = 0
self._read_buffer = bytearray()
self._write_buffer = bytes()
# - on_message_begin()
# - on_url(url: bytes)
# - on_header(name: bytes, value: bytes)
# - on_headers_complete()
# - on_body(body: bytes)
# - on_message_complete()
# - on_chunk_header()
# - on_chunk_complete()
# - on_status(status: bytes)
def get_address(self):
return self._addr
def close(self):
KBEngine.deregisterReadFileDescriptor(self._fd)
self._sock.close()
self._fd = 0
self._sock = None
self.cleanup()
def cleanup(self):
self.request = None
self.parser = None
self.headers = []
self.app = None
self._read_buffer.clear()
self._write_buffer = None
def data_received(self, fd):
# DEBUG_MSG("data_received, fd: %s, id: %s" % (fd, id(self)))
if self.parser is None:
self.parser = HttpRequestParser(self)
self.headers = []
while True:
try:
data = self._sock.recv(4096)
# DEBUG_MSG("data_received, fd: %s, data len: %s" %
# (fd, len(data)))
#
if not data:
ERROR_MSG("data_received, data len is 0, close")
self.close()
return
self._read_buffer += data
except (socket.error, IOError, OSError) as e:
_errno = errno_from_exception(e)
# signal
if _errno == errno.EINTR:
continue
# recvrecv
# recv
# epoll
elif _errno in ERRNO_WOULDBLOCK:
DEBUG_MSG("data_received, done")
break
ERROR_MSG("socket recv error: %s" % str(e))
self.close()
return
except Exception as e:
ERROR_MSG("data_received exception, e: %s" % str(e))
return
if self._read_buffer:
try:
self.parser.feed_data(bytes(self._read_buffer))
# feedfeed feed
self._read_buffer.clear()
except HttpParserError as e:
ERROR_MSG(
"Connection::data_received feed_data error. error: %s"
" \n %s \n id: %s" % (
str(e), traceback.format_exc(), id(self)
)
)
# ERROR_MSG(
# "Connection::data_received feed_data error. read_buffer: %s"
# " \n" % str(bytes(self._read_buffer))
#
# )
# TODO
def on_url(self, url):
if not self.url:
self.url = url
else:
self.url += url
def on_header(self, key, value):
"""
:param key:
:param value: bytes
:return:
"""
self._header_fragment += key
if value is not None:
try:
value = value.decode()
except UnicodeDecodeError:
value = value.decode("latin_1")
self.headers.append(
(self._header_fragment.decode().casefold(), value)
)
self._header_fragment = b""
def on_headers_complete(self):
DEBUG_MSG("Connection::on_header_complete")
self.request = Request(
url_bytes=self.url,
headers=CIMultiDict(self.headers),
version=self.parser.get_http_version(),
method=self.parser.get_method().decode(),
connection=self,
)
def on_body(self, data):
DEBUG_MSG("Connection::on_body, data len: %s" % len(data))
# DEBUG_MSG("data: %s" % str(data))
self.request.body.append(data)
def on_message_complete(self):
INFO_MSG("Connection request: %s" % str(self.request))
self.request.body = b"".join(self.request.body)
self.handle_request()
def handle_request(self):
request_handler_class = self.app.router.get(self.request.path)
if not request_handler_class:
response = HTTPResponse(
"NotFound handler", status=500
)
self.write_response(response)
return
try:
handler = request_handler_class(self.request)
# TODO url
handler.execute_handler()
except BaseException_ as e:
response = HTTPResponse(
"An error occurred, error: %s" % str(e),
status=e.status_code or 500
)
self.write_response(response)
except Exception as e:
ERROR_MSG("Internal Server Error: %s" % str(e))
response = HTTPResponse(
"Internal Server Error", status=500
)
self.write_response(response)
def write_response(self, response):
self._write_buffer = response.output()
self.send_body_size = len(self._write_buffer)
self._write_to_fd(False, self._fd)
if self._check_send_finish():
self.write_completed()
else:
# fdepoll
DEBUG_MSG("Connection::register write fd: %s" % self._fd)
KBEngine.registerWriteFileDescriptor(
self._fd, Functor(self._write_to_fd, True)
)
def _write_to_fd(self, in_poller, fd):
try:
send_size = self._sock.send(self._write_buffer[self.has_sent_size:])
self.has_sent_size += send_size
DEBUG_MSG("_write_to_fd, in_poller: %s, fd: %s" % (in_poller, fd))
if in_poller and self._check_send_finish():
self.write_completed(in_poller)
except (socket.error, IOError, OSError) as e:
ERROR_MSG("write to fd error: %s" % str(e))
self.close()
def _check_send_finish(self):
return self.has_sent_size >= self.send_body_size
def write_completed(self, in_poller=False):
if in_poller:
KBEngine.deregisterWriteFileDescriptor(self._fd)
DEBUG_MSG("write_completed, in_poller: %s, fd: %s" %
(in_poller, self._fd))
self.close()