<s> from __future__ import absolute_import , division , print_function , unicode_literals <EOL> __metaclass__ = type <EOL> from threading import Lock <EOL> import sys <EOL> from collections import defaultdict , deque <EOL> import six <EOL> import logging <EOL> import socket <EOL> import errno <EOL> from . utils import get_errno <EOL> if six . PY2 : <EOL> from Queue import Queue <EOL> else : <EOL> from queue import Queue <EOL> from . concurrency import synchronized <EOL> from . exceptions import UnexpectedFrame , Timeout , METHOD_NAME_MAP <EOL> from . proto import Method <EOL> from . import spec <EOL> from . spec import FrameType <EOL> log = logging . getLogger ( '<STR_LIT>' ) <EOL> __all__ = [ '<STR_LIT>' ] <EOL> _CONTENT_METHODS = [ <EOL> spec . Basic . Return , <EOL> spec . Basic . Deliver , <EOL> spec . Basic . GetOk , <EOL> ] <EOL> class MethodReader : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , transport ) : <EOL> """<STR_LIT>""" <EOL> self . transport = transport <EOL> self . sock = transport . sock <EOL> self . method_queue = deque ( ) <EOL> self . partial_methods = { } <EOL> self . expected_types = defaultdict ( lambda : FrameType . METHOD ) <EOL> self . frames_recv = <NUM_LIT:0> <EOL> self . _method_read_lock = Lock ( ) <EOL> def _next_method ( self ) : <EOL> """<STR_LIT>""" <EOL> while not self . method_queue : <EOL> try : <EOL> frame = self . transport . read_frame ( ) <EOL> except Exception as exc : <EOL> if six . PY2 : <EOL> _ , _ , tb = sys . exc_info ( ) <EOL> exc . tb = tb <EOL> self . method_queue . append ( exc ) <EOL> break <EOL> self . frames_recv += <NUM_LIT:1> <EOL> if frame . frame_type not in ( self . expected_types [ frame . channel ] , <NUM_LIT:8> ) : <EOL> msg = '<STR_LIT>' . format ( frame . frame_type , self . expected_types [ frame . channel ] ) <EOL> self . method_queue . append ( UnexpectedFrame ( msg , channel_id = frame . channel ) ) <EOL> elif frame . frame_type == FrameType . METHOD : <EOL> self . _process_method_frame ( frame ) <EOL> elif frame . frame_type == FrameType . HEADER : <EOL> self . _process_content_header ( frame ) <EOL> elif frame . frame_type == FrameType . BODY : <EOL> self . _process_content_body ( frame ) <EOL> def _process_method_frame ( self , frame ) : <EOL> """<STR_LIT>""" <EOL> channel_id = frame . channel <EOL> method = Method ( ) <EOL> method . load_method_frame ( frame ) <EOL> if method . method_type in _CONTENT_METHODS : <EOL> self . partial_methods [ channel_id ] = method <EOL> self . expected_types [ channel_id ] = spec . FrameType . HEADER <EOL> else : <EOL> self . method_queue . append ( method ) <EOL> def _process_content_header ( self , frame ) : <EOL> """<STR_LIT>""" <EOL> method = self . partial_methods [ frame . channel ] <EOL> method . load_header_frame ( frame ) <EOL> if method . complete : <EOL> self . method_queue . append ( method ) <EOL> self . partial_methods . pop ( frame . channel , None ) <EOL> del self . expected_types [ frame . channel ] <EOL> else : <EOL> self . expected_types [ frame . channel ] = spec . FrameType . BODY <EOL> def _process_content_body ( self , frame ) : <EOL> """<STR_LIT>""" <EOL> method = self . partial_methods [ frame . channel ] <EOL> method . load_body_frame ( frame ) <EOL> if method . complete : <EOL> self . method_queue . append ( method ) <EOL> self . partial_methods . pop ( frame . channel , None ) <EOL> del self . expected_types [ frame . channel ] <EOL> @ synchronized ( '<STR_LIT>' ) <EOL> def _read_method ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _next_method ( ) <EOL> method = self . method_queue . popleft ( ) <EOL> if isinstance ( method , Exception ) : <EOL> raise method <EOL> log . debug ( '<STR_LIT>' <EOL> . format ( '<STR_LIT>' , method . channel_id , <EOL> method . method_type , METHOD_NAME_MAP [ method . method_type ] ) ) <EOL> return method <EOL> def read_method ( self , timeout = None ) : <EOL> """<STR_LIT>""" <EOL> if timeout is None : <EOL> return self . _read_method ( ) <EOL> orig_timeout = self . sock . gettimeout ( ) <EOL> if orig_timeout != timeout : <EOL> self . sock . settimeout ( timeout ) <EOL> try : <EOL> return self . _read_method ( ) <EOL> except socket . timeout : <EOL> raise Timeout ( ) <EOL> except socket . error as e : <EOL> if get_errno ( e ) == errno . EAGAIN : <EOL> raise Timeout ( ) <EOL> raise <EOL> finally : <EOL> if orig_timeout != timeout : <EOL> self . sock . settimeout ( orig_timeout ) <EOL> class MethodWriter : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , transport , frame_max ) : <EOL> """<STR_LIT>""" <EOL> self . transport = transport <EOL> self . frame_max = frame_max <EOL> self . methods_sent = <NUM_LIT:0> <EOL> def write_method ( self , method ) : <EOL> """<STR_LIT>""" <EOL> transport = self . transport <EOL> log . debug ( '<STR_LIT>' <EOL> . format ( '<STR_LIT>' , method . channel_id , <EOL> method . method_type , METHOD_NAME_MAP [ method . method_type ] ) ) <EOL> frames = Queue ( ) <EOL> frames . put ( method . dump_method_frame ( ) ) <EOL> if method . content : <EOL> frames . put ( method . dump_header_frame ( ) ) <EOL> chunk_size = self . frame_max - <NUM_LIT:8> <EOL> for frame in method . dump_body_frame ( chunk_size ) : <EOL> frames . put ( frame ) <EOL> while not frames . empty ( ) : <EOL> transport . write_frame ( frames . get ( ) ) <EOL> self . methods_sent += <NUM_LIT:1> </s>
