aiohttp_rpc.server.websocket.WsJsonRpcServer.handle_websocket_request() - Code Metrics - Inspection of "0.5.1 -> 0.5.2" - expert-m/aiohttp-rpc - Measure and Improve Code Quality continuously with Scrutinizer

Passed

Push — master ( 239fba...767d9a )

by Michael

created 2020-05-28 17:10 UTC

WsJsonRpcServer.handle_websocket_request() A

↳ Parent: aiohttp_rpc.server.websocket

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

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

expert-m / aiohttp-rpc

Push — master ( 239fba...767d9a )

WsJsonRpcServer.handle_websocket_request() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like