<s> """<STR_LIT>""" <EOL> import socket <EOL> import struct <EOL> import uuid <EOL> from stomp . connect import BaseConnection <EOL> from stomp . protocol import * <EOL> from stomp . transport import * <EOL> from stomp . utils import * <EOL> MCAST_GRP = '<STR_LIT>' <EOL> MCAST_PORT = <NUM_LIT> <EOL> class MulticastTransport ( Transport ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> Transport . __init__ ( self , [ ] , False , False , <NUM_LIT:0.0> , <NUM_LIT:0.0> , <NUM_LIT:0.0> , <NUM_LIT:0.0> , <NUM_LIT:0> , False , None , None , None , None , False , DEFAULT_SSL_VERSION , None , None , None ) <EOL> self . subscriptions = { } <EOL> self . current_host_and_port = ( MCAST_GRP , MCAST_PORT ) <EOL> def attempt_connection ( self ) : <EOL> """<STR_LIT>""" <EOL> self . socket = socket . socket ( socket . AF_INET , socket . SOCK_DGRAM , socket . IPPROTO_UDP ) <EOL> self . socket . setsockopt ( socket . IPPROTO_IP , socket . IP_MULTICAST_TTL , <NUM_LIT:2> ) <EOL> self . receiver_socket = socket . socket ( socket . AF_INET , socket . SOCK_DGRAM , socket . IPPROTO_UDP ) <EOL> self . receiver_socket . setsockopt ( socket . SOL_SOCKET , socket . SO_REUSEADDR , <NUM_LIT:1> ) <EOL> self . receiver_socket . setsockopt ( socket . SOL_SOCKET , socket . SO_REUSEPORT , <NUM_LIT:1> ) <EOL> self . receiver_socket . bind ( ( '<STR_LIT>' , MCAST_PORT ) ) <EOL> mreq = struct . pack ( "<STR_LIT>" , socket . inet_aton ( MCAST_GRP ) , socket . INADDR_ANY ) <EOL> self . receiver_socket . setsockopt ( socket . IPPROTO_IP , socket . IP_ADD_MEMBERSHIP , mreq ) <EOL> if not self . socket or not self . receiver_socket : <EOL> raise exception . ConnectFailedException ( ) <EOL> def send ( self , encoded_frame ) : <EOL> """<STR_LIT>""" <EOL> self . socket . sendto ( encoded_frame , ( MCAST_GRP , MCAST_PORT ) ) <EOL> def receive ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . receiver_socket . recv ( <NUM_LIT> ) <EOL> def process_frame ( self , f , frame_str ) : <EOL> frame_type = f . cmd . lower ( ) <EOL> if frame_type in [ '<STR_LIT>' ] : <EOL> return <EOL> if frame_type == '<STR_LIT>' : <EOL> frame_type = '<STR_LIT:message>' <EOL> f . cmd = '<STR_LIT>' <EOL> if frame_type in [ '<STR_LIT>' , '<STR_LIT:message>' , '<STR_LIT>' , '<STR_LIT:error>' , '<STR_LIT>' ] : <EOL> if frame_type == '<STR_LIT:message>' : <EOL> if f . headers [ '<STR_LIT>' ] not in self . subscriptions . values ( ) : <EOL> return <EOL> ( f . headers , f . body ) = self . notify ( '<STR_LIT>' , f . headers , f . body ) <EOL> self . notify ( frame_type , f . headers , f . body ) <EOL> if '<STR_LIT>' in f . headers : <EOL> receipt_frame = Frame ( '<STR_LIT>' , { '<STR_LIT>' : f . headers [ '<STR_LIT>' ] } ) <EOL> lines = convert_frame_to_lines ( receipt_frame ) <EOL> self . send ( encode ( pack ( lines ) ) ) <EOL> log . debug ( "<STR_LIT>" , f . cmd , f . headers , f . body ) <EOL> def stop ( self ) : <EOL> self . running = False <EOL> if hasattr ( self . receiver_socket , '<STR_LIT>' ) : <EOL> self . receiver_socket . shutdown ( socket . SHUT_RDWR ) <EOL> self . receiver_socket . close ( ) <EOL> self . disconnect_socket ( ) <EOL> Transport . stop ( self ) <EOL> class MulticastConnection ( BaseConnection , Protocol12 ) : <EOL> def __init__ ( self , wait_on_receipt = False ) : <EOL> self . transport = MulticastTransport ( ) <EOL> self . transport . set_listener ( '<STR_LIT>' , self ) <EOL> self . transactions = { } <EOL> Protocol12 . __init__ ( self , self . transport , ( <NUM_LIT:0> , <NUM_LIT:0> ) ) <EOL> def connect ( self , username = None , passcode = None , wait = False , headers = { } , ** keyword_headers ) : <EOL> pass <EOL> def subscribe ( self , destination , id , ack = '<STR_LIT>' , headers = { } , ** keyword_headers ) : <EOL> self . transport . subscriptions [ id ] = destination <EOL> def unsubscribe ( self , id , headers = { } , ** keyword_headers ) : <EOL> del self . transport . subscriptions [ id ] <EOL> def disconnect ( self , receipt = str ( uuid . uuid4 ( ) ) , headers = { } , ** keyword_headers ) : <EOL> Protocol12 . disconnect ( self , receipt , headers , ** keyword_headers ) <EOL> self . transport . stop ( ) <EOL> def send_frame ( self , cmd , headers = { } , body = '<STR_LIT>' ) : <EOL> frame = utils . Frame ( cmd , headers , body ) <EOL> if cmd == CMD_BEGIN : <EOL> trans = headers [ HDR_TRANSACTION ] <EOL> if trans in self . transactions : <EOL> self . notify ( '<STR_LIT:error>' , { } , '<STR_LIT>' % trans ) <EOL> else : <EOL> self . transactions [ trans ] = [ ] <EOL> elif cmd == CMD_COMMIT : <EOL> trans = headers [ HDR_TRANSACTION ] <EOL> if trans not in self . transactions : <EOL> self . notify ( '<STR_LIT:error>' , { } , '<STR_LIT>' % trans ) <EOL> else : <EOL> for f in self . transactions [ trans ] : <EOL> self . transport . transmit ( f ) <EOL> del self . transactions [ trans ] <EOL> elif cmd == CMD_ABORT : <EOL> trans = headers [ '<STR_LIT>' ] <EOL> del self . transactions [ trans ] <EOL> else : <EOL> if '<STR_LIT>' in headers : <EOL> trans = headers [ '<STR_LIT>' ] <EOL> if trans not in self . transactions : <EOL> self . transport . notify ( '<STR_LIT:error>' , { } , '<STR_LIT>' % trans ) <EOL> return <EOL> else : <EOL> self . transactions [ trans ] . append ( frame ) <EOL> else : <EOL> self . transport . transmit ( frame ) </s>
