# -*- coding: utf-8 -*-

"""

FileName: connection

Author: Tao Hao

@contact: [email protected]

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

# 此次的recv数据读完了，recv抛出下面的异常

# 此次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))

# 注意要在feed之后清，这次请求可能会被feed 多次，因此每次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:

# 一次没有发完，注册fd，交给epoll

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()