aiohttp_rpc.server.ws.WsJsonRpcServer.handle_websocket_request() - Code Metrics - Inspection of "Split protocol.py" - expert-m/aiohttp-rpc - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 1de29a...581576 )

by Michael

created 2020-05-28 11:50 UTC

WsJsonRpcServer.handle_websocket_request() A

↳ Parent: aiohttp_rpc.server.ws

Complexity

Conditions

Size

Total Lines	17
Code Lines	12

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
eloc	12
dl	0
loc	17
rs	9.8
c	0
b	0
f	0
cc	3
nop	2

import json

import aiohttp
from aiohttp import web, web_ws

from .base import BaseJsonRpcServer


__all__ = (
    'WsJsonRpcServer',
)


class WsJsonRpcServer(BaseJsonRpcServer):
    async def handle_http_request(self, http_request: web.Request) -> web.StreamResponse:
        if http_request.method == 'GET' and http_request.headers.get('upgrade', '').lower() == 'websocket':
            return await self.handle_websocket_request(http_request)
        else:
            return web.HTTPMethodNotAllowed(method=http_request.method, allowed_methods=('POST',))

    async def handle_websocket_request(self, http_request: web.Request) -> web_ws.WebSocketResponse:
        http_request.msg_id = 0
        http_request.pending = {}

        ws = web_ws.WebSocketResponse()
        await ws.prepare(http_request)
        http_request.ws = ws

        while not ws.closed:
            ws_msg = await ws.receive()

            if ws_msg.type != aiohttp.WSMsgType.TEXT:
                continue

            await self._handle_ws_msg(http_request, ws_msg)

        return ws

    async def _handle_ws_msg(self, http_request: web.Request, ws_msg: web_ws.WSMessage) -> None:
        input_data = json.loads(ws_msg.data)
        output_data = await self._process_input_data(input_data, http_request=http_request)

        if http_request.ws._writer.transport.is_closing():
            self.clients.remove(http_request)
            await http_request.ws.close()

        await http_request.ws.send_str(self.json_serialize(output_data))


1			import json
2
3			import aiohttp
4			from aiohttp import web, web_ws
5
6			from .base import BaseJsonRpcServer
7
8
9			__all__ = (
10			'WsJsonRpcServer',
11			)
12
13
14			class WsJsonRpcServer(BaseJsonRpcServer):
15			async def handle_http_request(self, http_request: web.Request) -> web.StreamResponse:
16			if http_request.method == 'GET' and http_request.headers.get('upgrade', '').lower() == 'websocket':
17			return await self.handle_websocket_request(http_request)
18			else:
19			return web.HTTPMethodNotAllowed(method=http_request.method, allowed_methods=('POST',))
20
21			async def handle_websocket_request(self, http_request: web.Request) -> web_ws.WebSocketResponse:
22			http_request.msg_id = 0
23			http_request.pending = {}
24
25			ws = web_ws.WebSocketResponse()
26			await ws.prepare(http_request)
27			http_request.ws = ws
28
29			while not ws.closed:
30			ws_msg = await ws.receive()
31
32			if ws_msg.type != aiohttp.WSMsgType.TEXT:
33			continue
34
35			await self._handle_ws_msg(http_request, ws_msg)
36
37			return ws
38
39			async def _handle_ws_msg(self, http_request: web.Request, ws_msg: web_ws.WSMessage) -> None:
40			input_data = json.loads(ws_msg.data)
41			output_data = await self._process_input_data(input_data, http_request=http_request)
42
43			if http_request.ws._writer.transport.is_closing():
44			self.clients.remove(http_request)
45			await http_request.ws.close()
46
47			await http_request.ws.send_str(self.json_serialize(output_data))
48

expert-m / aiohttp-rpc

Push — master ( 1de29a...581576 )

WsJsonRpcServer.handle_websocket_request() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like