kbengine-httpserver/httpserver/connection.py at master · blivenwei/kbengine-httpserver · GitHub

History

237 lines (205 loc) · 7.55 KB

Raw

100

101

102

103

104

105

106

107

108

109

110

111

112

113

114

115

116

117

118

119

120

121

122

123

124

125

126

127

128

129

130

131

132

133

134

135

136

137

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

# -*- 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

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

kbengine-httpserver/httpserver/connection.py at master · blivenwei/kbengine-httpserver · GitHub

Expand file tree

Breadcrumbs

connection.py

Latest commit

History

Breadcrumbs

connection.py

File metadata and controls