gGddlmZddlZddlZddlZddlmZmZmZm Z m Z m Z ddl m Z mZmZmZmZmZmZmZddlmZddlmZmZmZdd lmZdd lmZerdd lm Z ejBe"Z#Gd d Z$y)) annotationsN) TYPE_CHECKINGAnyCallableDictListOptional)BindingCallback ChannelEvents ChannelStatesRealtimeChannelOptions"RealtimePostgresChangesListenEventRealtimePresenceStateRealtimeSubscribeStates)http_endpoint_url)AsyncRealtimePresencePresenceOnJoinCallbackPresenceOnLeaveCallback) AsyncPush) AsyncTimer)AsyncRealtimeClientceZdZdZ d d!dZedZedZedZedZ edZ ed Z d d"d Z d Z d d#d Zd$d Z d d%dZd&dZ d'dZ d( d)dZd*dZd+dZd,dZd-dZ d.dZ d/dZd0dZdZd+dZdZd0dZd d1dZd2dZ dZ!y)3AsyncRealtimeChannela `Channel` is an abstraction for a topic listener for an existing socket connection. Each Channel has its own topic and a list of event-callbacks that responds to messages. Should only be instantiated through `connection.RealtimeClient().channel(topic)`. Nc|_|xstdddddidd_|_d_i_t _tj_ g_ jj_ ttjj_t#j$d_j)_dfd }fd }j j-d |j-d |fd }fd }j/d|j/d|fd}j/tj0|y)z Initialize the Channel object. :param socket: RealtimeClient object :param topic: Topic that it subscribes to on the realtime server :param params: Optional parameters for connection. F)ackselfkey) broadcastpresenceprivate)configc d|zS)Nr)triess L/var/www/openai/venv/lib/python3.12/site-packages/realtime/_async/channel.pyz/AsyncRealtimeChannel.__init__..Hs5ctj_jj j D]%}t j|j'g_yN) r JOINEDstate rejoin_timerreset _push_bufferasyncio create_tasksend)payloadargspushrs r)on_join_push_okz6AsyncRealtimeChannel.__init__..on_join_push_okMsN&--DJ    # # %))##DIIK0* "D r+cjsytjdjtj _jjy)Nzjoin push timeout for channel ) is_joiningloggererrortopicr ERROREDr/r0schedule_timeoutr7rs r)on_join_push_timeoutz;AsyncRealtimeChannel.__init__..on_join_push_timeoutTsD?? LL9$**F G&..DJ    . . 0r+oktimeoutctjdjdjj t j _jjy)Nchannel z closed) r<infor>r0r1r CLOSEDr/socketremove_channelrAs r)on_closez/AsyncRealtimeChannel.__init__..on_close`sM KK(4::,g6 7    # # %&--DJ KK & &t ,r+cjs jrytjdjd|t j _jjy)NrFz error: ) is_leaving is_closedr<rGr>r r?r/r0r@)r6r7rs r)on_errorz/AsyncRealtimeChannel.__init__..on_errorfsO$.. KK(4::,hwi@ A&..DJ    . . 0r+closer=cHjj||yr-)_trigger_reply_event_name)r6refrs r)on_replyz/AsyncRealtimeChannel.__init__..on_replyqs MM$005w ?r+Nr6Dict[str, Any])rIrparamsr> _joined_oncebindingsrr#r rHr/r2rDrr join join_pushr_rejoin_until_connectedr0_broadcast_endpoint_urlbroadcast_endpoint_urlreceive_onreply) rrIr>rXr9rBrKrOrUs ` r)__init__zAsyncRealtimeChannel.__init__)s<  6%*E:"BK !   !24 -d3 ")) -/{{** "4););T[[I&  ( (*@ '+&B&B&D# # 1 t_5== +  -  1 (# (# @ $$h/r+c<|jtjk(Sr-)r/r rHrs r)rNzAsyncRealtimeChannel.is_closedwzz]1111r+c<|jtjk(Sr-)r/r JOININGres r)r;zAsyncRealtimeChannel.is_joining{zz]2222r+c<|jtjk(Sr-)r/r LEAVINGres r)rMzAsyncRealtimeChannel.is_leavingrir+c<|jtjk(Sr-)r/r r?res r) is_erroredzAsyncRealtimeChannel.is_erroredrir+c<|jtjk(Sr-)r/r r.res r) is_joinedzAsyncRealtimeChannel.is_joinedrfr+c.|jjSr-)r\rTres r)join_refzAsyncRealtimeChannel.join_refs~~!!!r+c FKjjs"jjd{jr t dj j di}|j di}|j di}|j dd}i}|||ttdjj d gd }jjrjj|d <jjid|i|d _dfd }dfd }fd} jjd|jd|jd| jd{S7d7 w)zg Subscribe to the channel. :return: The Channel instance for method chaining. NzdTried to subscribe multiple times. 'subscribe' can only be called a single time per channel instancer%r"r#r$Fc|jSr-)filter)xs r)r*z0AsyncRealtimeChannel.subscribe..s!((r+postgres_changes)r"r#r$rv access_tokenTc|jdg}t|dk(r xr tjdy jjdg}g}t|}t |D]K}||}|j jd}|j jd}|j jd} |j jd} |t|kr||nd} | ry| jd|k(re| jd|k(rQ| jd| k(r=| jd| k(r)| jd|_|j|tj j xr  tjtdy| jd< xr tjdyy) Nrvreventschematablertidz@mismatch between server and client bindings for postgres changes)getlenr SUBSCRIBEDrZrangertr|appendr3r4 unsubscribe CHANNEL_ERROR Exception)r6server_postgres_changesclient_postgres_changesnew_postgres_bindings bindings_leniclient_bindingryrzr{rtserver_bindingcallbackrs r)r9z7AsyncRealtimeChannel.subscribe..on_join_push_oks@G &A'./14S*A*L*Ld!S*.--*;*;E+2266x@F*1155g>E+2266x@Fs#:;;02!#'*..w75@*..x8FB*..w75@*..x8FB,:,>,>t,D)-44^D++D,<,<,>? X3AA% b& =-@5J 01OX&=&H&H$OOr+cxxr5tjttj|yyr-)rrrjsondumps)r6rs r)on_join_push_errorz:AsyncRealtimeChannel.subscribe..on_join_push_errors1X+99djj12r+c@xrtjdyyr-)r TIMED_OUT)r7rs r)rBz.on_join_push_timeoutsNX&=&G&GNNr+rCr=rDrV)rI is_connectedconnectrYrrXr}listmaprZrwr\update_payloadr`_rejoin) rrr%r"r#r$access_token_payloadr9rrBs `` r) subscribezAsyncRealtimeChannel.subscribes{{''++%%' ' '   v [[__Xr2F ;3Izz*b1HjjE2G#% &$"$(*DMM,=,=>PRT,UV% F{{''7;{{7O7O$^4 NN ) )>Hf%>)=> !%D / Pb   O NN " "4 9 A A+ gi!56,,.  { (v !s"4F!FEF!FF!F!cKtj_jj j j fd}ttji}|jd|jd||jd{js|jdiyy7(w)Nctjdjdjtj dy)NrFz leaveleave)r<rGr>rRr rPrAs r)_closez0AsyncRealtimeChannel.unsubscribe.._closes0 KK(4::,f5 6 MM---w 7r+rCrD)r rkr/r0r1r\destroyrr rr`r5 _can_pushtrigger)rr leave_pushs` r)rz AsyncRealtimeChannel.unsubscribes"**  !   8t]%8%8"= 4(00FCoo~~   tR (  sBC "C #)C cHK|jstd|d|jd|xs |j}t ||||}|j r|j d{|S|j|jj||S73w)Nztried to push 'z' to 'z?' before joining. Use channel.subscribe() before pushing events) rYrr>rDrrr5 start_timeoutr2r)rryr6rDr8s r)r8zAsyncRealtimeChannel.push s  !%tzzl:yz )T\\ugw7 >> ))+           $ $T * sA*B",B -4B"cK |jj|jdd|jiddd{y7#t$r}t ||cYd}~Sd}~wwxYww)zx Coroutine that attempts to join Phoenix Realtime server via a certain topic. :return: Channel phx_joinr%N)r>ryr6rT)rIr5r>rXrprint)res r)r[zAsyncRealtimeChannel.join sb ++""!ZZ' ($++6     !HK sDA(9AAAA(A A% A A%A( A%%A(c|j}t||xsi|}|jj|gj ||S)aA Set up a listener for a specific event. :param type: The type of the event to listen for. :param filter: Additional parameters for the event. :param callback: The callback function to execute when the event is received. :return: The Channel instance for method chaining. )typertr)lowerr rZ setdefaultr)rrrrttype_lowercasebindings r)razAsyncRealtimeChannel._on4sE~flXV   4;;GD r+c|j}||jvr9|j|Dcgc]}|j|k7r|c}|j|<|Scc}w)a Remove a listener for a specific event type and filter. :param type: The type of the event to remove the listener for. :param filter: The filter associated with the listener to remove. :return: The Channel instance for method chaining. This method removes all bindings for the specified event type that match the given filter. If no matching bindings are found, the method does nothing. )rrZrt)rrrtrrs r)_offzAsyncRealtimeChannel._offFsd T]] * $}}^<->V+<-DMM. )  -sAc6|jdd|ifdS)a Set up a listener for a specific broadcast event. :param event: The name of the broadcast event to listen for. :param callback: The callback function to execute when the event is received. :return: The Channel instance for method chaining. r"ryc|Sr-r'r6_rs r)r*z3AsyncRealtimeChannel.on_broadcast..h (9r+rtrra)rryrs `r) on_broadcastz!AsyncRealtimeChannel.on_broadcast[s)xx U#9  r+cL|||d}|r||d<|jd|fdS)a Set up a listener for a specific Postgres changes event. :param event: The name of the Postgres changes event to listen for. :param table: The table name for which changes should be monitored. :param callback: The callback function to execute when the event is received. :param schema: The database schema where the table exists. Default is 'public'. :return: The Channel instance for method chaining. )ryrzr{rtrvc|Sr-r'rs r)r*z:AsyncRealtimeChannel.on_postgres_changes..rr+rr)rryrr{rzrtbinding_filters ` r)on_postgres_changesz(AsyncRealtimeChannel.on_postgres_changesks=$$)FUK '-N8 $xx !9  r+cDK|jd|d{y7w)zu Track a user's presence. :param user_status: User's presence status. :return: None trackN send_presence)r user_statuss r)rzAsyncRealtimeChannel.tracks  +666   cDK|jdid{y7w)zC Untrack a user's presence. :return: None untrackNrres r)rzAsyncRealtimeChannel.untracks   B///rc.|jjSr-)r#r/res r)presence_statez#AsyncRealtimeChannel.presence_states}}"""r+c<|jj||S)z Register a callback for presence sync events. :param callback: The callback function to execute when a presence sync event occurs. :return: The Channel instance for method chaining. )r#on_syncrrs r)on_presence_syncz%AsyncRealtimeChannel.on_presence_syncs h' r+c<|jj||S)z Register a callback for presence join events. :param callback: The callback function to execute when a presence join event occurs. :return: The Channel instance for method chaining. )r#on_joinrs r)on_presence_joinz%AsyncRealtimeChannel.on_presence_joins h' r+c<|jj||S)z Register a callback for presence leave events. :param callback: The callback function to execute when a presence leave event occurs. :return: The Channel instance for method chaining. )r#on_leavers r)on_presence_leavez&AsyncRealtimeChannel.on_presence_leaves x( r+chK|jtjd||dd{y7w)a Sends a broadcast message to the current channel. :param event: The name of the broadcast event. :param data: The data to be sent with the message. :return: An asyncio.Future object representing the send operation. r")rryr6N)r8r r"rrydatas r)send_broadcastz#AsyncRealtimeChannel.send_broadcasts1ii  # # 5T B   s (202cFt|jjdS)Nz/api/broadcast)rrI http_endpointres r)r^z,AsyncRealtimeChannel._broadcast_endpoint_urls#DKK$=$=>?~NNr+cK|jry|jj|jd{tj |_|jjd{y7<7wr-) rMrI_leave_open_topicr>r rhr/r\resendres r)rzAsyncRealtimeChannel._rejoinsX ?? kk++DJJ777"** nn##%%% 8%s!6A9A56A9/A70A97A9cJ|jjxr |jSr-)rIrrYres r)rzAsyncRealtimeChannel._can_pushs{{''=D,=,==r+cfK|jtj||dd{y7w)N)ryr6)r8r r#rs r)rz"AsyncRealtimeChannel.send_presences%ii ..%D0QRRRs '1/1c\ |j tjtjtjtj g}| |vr||j jk7ry dvrCt fd|jjdg}|D]}|j||y|jj g}|D]T} dvr)|j}|jjdr*|jjddjnd} |jdr |jddjnd} |jdijdr0|jdijddjnd} |r2||jd gvr| | k(s| d k(r|j||| | d fvs|j||2|j k(sC|j||Wy) N)insertupdatedeletec`|jjddjdfvS)Nryr!*)rtr}r)rrs r)r*z/AsyncRealtimeChannel._trigger..s, 2 27B ? E E G"C(!)r+rv)r"rvr#ryr!rridsr)rr rPr=rr[r\rTrtrZr}rr|r) rrr6rTeventsrvrrZbind_id bind_event payload_event data_typers @r)rRzAsyncRealtimeChannel._triggers               ?~7C4>>CUCU>--g6 **7B7==?!#;;w/ GR0668!"#;;vr266v> FB/33FB?EEG!  #w{{5"'=='94 c8I((#6# s';;((#6\\^3$$Wc27$r+c d|S)N chan_reply_r')rrTs r)rSz&AsyncRealtimeChannel._reply_event_namesSE""r+cK|jj|jjr|j d{yy7wr-)r0r@rIrrres r)r]z,AsyncRealtimeChannel._rejoin_until_connecteds< **, ;; # #,,.  $ sAAA Ar-)rIrr>strrXz Optional[RealtimeChannelOptions]returnNone)rzHOptional[Callable[[RealtimeSubscribeStates, Optional[Exception]], None]]rr)ryrr6rWrDz Optional[int]rr)rr)rrrr rtzOptional[Dict[str, Any]]rr)rrrtrWrr)ryrr Callable[[Dict[str, Any]], None]rr)rpublicN) ryrrrr{rrzrrt Optional[str]rr)rrWrr)rr)rr)rzCallable[[], None]rr)rrrr)rrrr)ryrrrrr)rrr6z Optional[Any]rTr)rTr)"__name__ __module__ __qualname____doc__rcpropertyrNr;rMrmrorqrrr8r[rarrrrrrrrrrr^rrrrRrSr]r'r+r)rr"s48 K0#K0K01 K0  K0\2233333322""  i i  iV)&MQ#1S13f#!r+r)% __future__rr3rloggingtypingrrrrrr realtime.typesr r r r rrrr transformersrr#rrrr8rtimerrclientr getLoggerrr<rr'r+r)rsa" EE   - +   8 $t!t!r+