<s> """<STR_LIT>""" <EOL> import logging <EOL> import struct <EOL> from collections import deque <EOL> from snakemq . exceptions import NoConnection <EOL> from snakemq . buffers import StreamBuffer <EOL> from snakemq . exceptions import SnakeMQBrokenPacket <EOL> from snakemq . callbacks import Callback <EOL> SEND_BLOCK_SIZE = <NUM_LIT:64> * <NUM_LIT> <EOL> BIN_SIZE_FORMAT = "<STR_LIT>" <EOL> SIZEOF_BIN_SIZE = struct . calcsize ( BIN_SIZE_FORMAT ) <EOL> def size_to_bin ( size ) : <EOL> return struct . pack ( BIN_SIZE_FORMAT , size ) <EOL> def bin_to_size ( buf ) : <EOL> return struct . unpack ( BIN_SIZE_FORMAT , buf ) [ <NUM_LIT:0> ] <EOL> class ReceiveBuffer ( StreamBuffer ) : <EOL> def __init__ ( self ) : <EOL> StreamBuffer . __init__ ( self ) <EOL> self . packet_size = None <EOL> def get_packets ( self ) : <EOL> """<STR_LIT>""" <EOL> packets = [ ] <EOL> while self . size : <EOL> if self . packet_size is None : <EOL> if self . size < SIZEOF_BIN_SIZE : <EOL> break <EOL> header = self . get ( SIZEOF_BIN_SIZE , True ) <EOL> self . packet_size = bin_to_size ( header ) <EOL> if self . packet_size < <NUM_LIT:0> : <EOL> raise SnakeMQBrokenPacket ( "<STR_LIT>" ) <EOL> else : <EOL> if self . size < self . packet_size : <EOL> break <EOL> packets . append ( self . get ( self . packet_size , True ) ) <EOL> self . packet_size = None <EOL> return packets <EOL> class ConnectionInfo ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . send_buffer = StreamBuffer ( ) <EOL> self . recv_buffer = ReceiveBuffer ( ) <EOL> self . send_in_progress = False <EOL> self . queued_packet_ids = deque ( ) <EOL> class Packeter ( object ) : <EOL> def __init__ ( self , link ) : <EOL> """<STR_LIT>""" <EOL> self . link = link <EOL> self . log = logging . getLogger ( "<STR_LIT>" ) <EOL> self . on_connect = Callback ( ) <EOL> self . on_disconnect = Callback ( ) <EOL> self . on_packet_recv = Callback ( ) <EOL> self . on_packet_sent = Callback ( ) <EOL> self . on_error = Callback ( ) <EOL> self . _connections = { } <EOL> self . _last_packet_id = <NUM_LIT:0> <EOL> self . link . on_connect . add ( self . _on_connect ) <EOL> self . link . on_disconnect . add ( self . _on_disconnect ) <EOL> self . link . on_recv . add ( self . _on_recv ) <EOL> self . link . on_ready_to_send . add ( self . _on_ready_to_send ) <EOL> def send_packet ( self , conn_id , buf ) : <EOL> """<STR_LIT>""" <EOL> assert type ( buf ) == bytes <EOL> try : <EOL> conn = self . _connections [ conn_id ] <EOL> except KeyError : <EOL> raise NoConnection <EOL> self . _last_packet_id += <NUM_LIT:1> <EOL> packet_id = self . _last_packet_id <EOL> buf = size_to_bin ( len ( buf ) ) + buf <EOL> conn . send_buffer . put ( buf ) <EOL> conn . queued_packet_ids . append ( ( len ( buf ) , packet_id ) ) <EOL> self . _send_to_link ( conn_id , conn ) <EOL> return packet_id <EOL> def _on_connect ( self , conn_id ) : <EOL> self . _connections [ conn_id ] = ConnectionInfo ( ) <EOL> self . on_connect ( conn_id ) <EOL> def _on_disconnect ( self , conn_id ) : <EOL> del self . _connections [ conn_id ] <EOL> self . on_disconnect ( conn_id ) <EOL> def _on_recv ( self , conn_id , buf ) : <EOL> recv_buffer = self . _connections [ conn_id ] . recv_buffer <EOL> recv_buffer . put ( buf ) <EOL> try : <EOL> packets = recv_buffer . get_packets ( ) <EOL> except SnakeMQBrokenPacket as exc : <EOL> self . log . error ( "<STR_LIT>" % ( conn_id , exc ) ) <EOL> self . on_error ( conn_id , exc ) <EOL> self . link . close ( conn_id ) <EOL> return <EOL> for packet in packets : <EOL> self . log . debug ( "<STR_LIT>" % ( conn_id , len ( packet ) ) ) <EOL> self . on_packet_recv ( conn_id , packet ) <EOL> def _on_ready_to_send ( self , conn_id , sent_length ) : <EOL> conn = self . _connections [ conn_id ] <EOL> conn . send_in_progress = False <EOL> conn . send_buffer . cut ( sent_length ) <EOL> while sent_length > <NUM_LIT:0> : <EOL> first , packet_id = conn . queued_packet_ids . popleft ( ) <EOL> if first <= sent_length : <EOL> self . on_packet_sent ( conn_id , packet_id ) <EOL> else : <EOL> conn . queued_packet_ids . appendleft ( ( first - sent_length , <EOL> packet_id ) ) <EOL> sent_length -= first <EOL> self . _send_to_link ( conn_id , conn ) <EOL> def _send_to_link ( self , conn_id , conn ) : <EOL> if conn . send_in_progress : <EOL> return <EOL> buf = conn . send_buffer . get ( SEND_BLOCK_SIZE , False ) <EOL> if buf : <EOL> self . link . send ( conn_id , buf ) <EOL> conn . send_in_progress = True </s>
