o }jhI@sDddlmZddlZddlZddlZddlmZmZmZm Z m Z m Z m Z ddl mZddlmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZmZm Z ddl!m"Z"m#Z#m$Z$m%Z%m&Z&m'Z'm(Z(m)Z)m*Z*m+Z+m,Z,m-Z-m.Z.m/Z/m0Z0m1Z1ddl2m3Z3d d l4m5Z5d d l6m7Z7d d l8m9Z9erd d l:m;Z;eGdddZ?dS)) annotationsN) TYPE_CHECKINGAnyCallableDictListMappingOptional) assert_never)BroadcastCallbackBroadcastPayloadCallback ChannelEvents ChannelStatesPostgresChangesCallbackPostgresChangesDataPresenceOnJoinCallbackPresenceOnLeaveCallbackRealtimeAcknowledgementStatusRealtimeChannelBroadcastConfigRealtimeChannelConfigRealtimeChannelOptionsRealtimeChannelPresenceConfig"RealtimePostgresChangesListenEventRealtimePresenceStateRealtimeSubscribeStates)BroadcastMessageChannelCloseMessageChannelErrorMessageHeartbeatMessageMessagePostgresChangesMessagePostgresChangesPayloadPostgresRowChangePresenceDiffMessagePresenceStateMessage ReplyMessageReplyPostgresChanges ServerMessageSuccessReplyMessageSuccessSystemPayload SystemMessage)http_endpoint_url)AsyncRealtimePresence) AsyncPush) AsyncTimer)AsyncRealtimeClientc@s>eZdZdZ d]d^d d Zd dZd_ddZeddZeddZ eddZ eddZ eddZ d]d`dd Z dad!d"Z d]dbd(d)Zdcd*d+Zddd-d.Z   dedfd5d6Zdgd8d9Zdhd;d<Zdad=d>Zdid@dAZdjdCdDZdkdFdGZdldIdJZdmdMdNZdOdPZdadQdRZdSdTZdmdUdVZdndYdZZd[d\Z dS)oAsyncRealtimeChannela Channel is an abstraction for a topic subscription on an existing socket connection. Each Channel has its own topic and a list of event-callbacks that respond to messages. Should only be instantiated through `AsyncRealtimeClient.channel(topic)`. Nsocketr2topicstrparams Optional[RealtimeChannelOptions]returnNonecs|_|r|n ddddddiddi_|_d_t_tj_g_ jj _ t t j j_i_g_g_g_d_tjdd __dfd d }fdd}jtj|tj|dS)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. configF)ackselfkey) broadcastpresenceprivateNcSsd|S)Nr)triesrCrCM/var/www/html/bot/env/lib/python3.10/site-packages/realtime/_async/channel.pynsz/AsyncRealtimeChannel.__init__..payloadr(cs6tj_jjD] }t|q g_dSN) rJOINEDstate rejoin_timerreset _push_bufferasyncio create_tasksend)rGpushr=rCrEon_join_push_okss    z6AsyncRealtimeChannel.__init__..on_join_push_okcs2jsdStdjtj_jdS)Nzjoin push timeout for channel ) is_joiningloggererrorr5rERROREDrJrKschedule_timeoutrCrRrCrEon_join_push_timeoutzs z;AsyncRealtimeChannel.__init__..on_join_push_timeoutrGr()r4r7r5 _joined_oncer/rArCLOSEDrJrMtimeoutr0rjoin join_pushmessages_waiting_for_ackbroadcast_callbackssystem_callbackspostgres_changes_callbackssubscribe_callbackr1_rejoin_until_connectedrK_broadcast_endpoint_urlbroadcast_endpoint_urlreceiverOkTimeout)r=r4r5r7rSrYrCrRrE__init__DsB      zAsyncRealtimeChannel.__init__cCs6td|jd|jtj|_|j |dS)Nchannel z closed) rUinfor5rKrLrr\rJr4_remove_channelrRrCrCrEon_closes zAsyncRealtimeChannel.on_closerGdict[str, Any]cCs>|js|jrdStd|jd|tj|_|j dS)Nrlz error: ) is_leaving is_closedrUrmr5rrWrJrKrX)r=rGrCrCrEon_errors zAsyncRealtimeChannel.on_errorcC |jtjkSrH)rJrr\rRrCrCrErr zAsyncRealtimeChannel.is_closedcCrtrH)rJrJOININGrRrCrCrErTruzAsyncRealtimeChannel.is_joiningcCrtrH)rJrLEAVINGrRrCrCrErqruzAsyncRealtimeChannel.is_leavingcCrtrH)rJrrWrRrCrCrE is_erroredruzAsyncRealtimeChannel.is_erroredcCrtrH)rJrrIrRrCrCrE is_joinedruzAsyncRealtimeChannel.is_joinedcallbackHOptional[Callable[[RealtimeSubscribeStates, Optional[Exception]], None]]c sjjs jIdHjrtdjd}|d}|d}|dd}d|||dd jDd i}jjrBjj|d <j |d _dfdd }dfdd }fdd} j t j | t j| t j| IdHS)a Subscribe to the channel. Can only be called once per channel instance. :param callback: Optional callback function that receives subscription state updates and any errors that occur during subscription :return: The Channel instance for method chaining :raises: Exception if called multiple times on the same channel instance NzdTried to subscribe multiple times. 'subscribe' can only be called a single time per channel instancer;r@rArBFcSg|]}|jqSrCbinding_filter.0crCrCrE z2AsyncRealtimeChannel.subscribe..r@rArBpostgres_changes access_tokenTrGr(cs|j}g}|rbtjD]U\}}|t|kr||nd}t|d||rJ|j|jkrJ|j|j krJ|j |j krJ|j |j krJ|j |_ | |q to^tjtddSdS|_ootjddSdS)Nz, z@mismatch between server and client bindings for postgres changes)r enumeraterclenrUrmeventseventschema_schematablefilteridappendrNrO unsubscriber CHANNEL_ERROR Exception SUBSCRIBED)rGserver_postgres_changesnew_postgres_bindingsipostgres_callbackserver_bindingrzr=rCrErSs>      z7AsyncRealtimeChannel.subscribe..on_join_push_okDict[str, Any]cs$otjtt|dSdSrH)rrrjsondumps)rGrzrCrEon_join_push_errors z:AsyncRealtimeChannel.subscribe..on_join_push_errorcso tjddSdSrH)r TIMED_OUTargsrrCrErYsz.on_join_push_timeoutrZ)rGr)r4 is_connectedconnectr[rr7getrcrr_update_payloadrhrriErrorrj_rejoin) r=rzr;r@rArBconfig_payloadrSrrYrCrrE subscribesF      &  zAsyncRealtimeChannel.subscribecsdtj_jjdfdd }ttj i}| t j | t j ||IdHdS)z Unsubscribe from the channel and leave the topic. Sets channel state to LEAVING and cleans up timers and pushes. r9r:cs tdjddS)Nrlz leave)rUrmr5rorrRrCrE_closes z0AsyncRealtimeChannel.unsubscribe.._closeNr9r:)rrwrJrKrLr_destroyr0rleaverhrrirrP)r=r leave_pushrCrRrErs  z AsyncRealtimeChannel.unsubscriberrr] Optional[int]r0cs||jstd|d|jd|p|j}t||||}|r2|IdH|jdus0Jd|S||j ||S)aN Push a message to the channel. :param event: The event name to push :param payload: The payload to send :param timeout: Optional timeout in milliseconds :return: AsyncPush instance representing the push operation :raises: Exception if called before subscribing to the channel ztried to push 'z' to 'z?' before joining. Use channel.subscribe() before pushing eventsNz Sent AsyncPush should have a ref) r[rr5r]r0 _can_pushrPref start_timeoutrMr)r=rrGr]rQrCrCrErQ#s   zAsyncRealtimeChannel.pushcs~|jd}|d}|d}|dd}d|||dd|jDdi}t|jtjd|i|jd }|j |Id H|S) zx Coroutine that attempts to join Phoenix Realtime server via a certain topic. :return: Channel r;r@rArBFcSr|rCr}rrCrCrErPrz-AsyncRealtimeChannel.join..r)r5rrGrN) r7rrcr!r5rr^r4 _make_refrP)r=r;r@rArBrmessagerCrCrEr^@s,     zAsyncRealtimeChannel.join"Callable[[BroadcastPayload], None]cCs|jt||d|S)a" Set up a listener for a specific broadcast event. :param event: The name of the broadcast event to listen for :param callback: Function called with the payload when a matching broadcast is received :return: The Channel instance for method chaining )rzr)rarr )r=rrzrCrCrE on_broadcast^s z!AsyncRealtimeChannel.on_broadcastr(Callable[[PostgresChangesPayload], None]r Optional[str]rrcCs"t|||||d}|j||S)a  Set up a listener for Postgres database changes. :param event: The type of database event to listen for (INSERT, UPDATE, DELETE, or *) :param callback: Function called with the payload when a matching change is detected :param table: The table name to monitor. Defaults to "*" for all tables :param schema: The database schema to monitor. Defaults to "public" :param filter: Optional filter string to apply :return: The Channel instance for method chaining )rzrrrr)rrcr)r=rrzrrrrCrCrEon_postgres_changesms   z(AsyncRealtimeChannel.on_postgres_changes&Callable[[SuccessSystemPayload], None]cC|j||S)z Set up a listener for system events. :param callback: The callback function to execute when a system event is received. :return: The Channel instance for method chaining. )rbrr=rzrCrCrE on_system zAsyncRealtimeChannel.on_system user_statuscs|d|IdHdS)z Track presence status for the current user. :param user_status: Dictionary containing the user's presence information trackN send_presence)r=rrCrCrErszAsyncRealtimeChannel.trackcs|diIdHdS)z> Stop tracking presence for the current user. untrackNrrRrCrCrErszAsyncRealtimeChannel.untrackrcCs|jjS)z Get the current state of presence on this channel. :return: Dictionary mapping presence keys to lists of presence payloads )rArJrRrCrCrEpresence_statesz#AsyncRealtimeChannel.presence_stateCallable[[], None]cCr)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. )rAon_syncrrCrCrEon_presence_syncs z%AsyncRealtimeChannel.on_presence_syncrcCr)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. )rAon_joinrrCrCrEon_presence_joinrz%AsyncRealtimeChannel.on_presence_joinrcCr)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. )rAon_leaverrCrCrEon_presence_leaverz&AsyncRealtimeChannel.on_presence_leavedatarcs"|tjd||dIdHdS)z Send a broadcast message through this channel. :param event: The name of the broadcast event :param data: The payload to broadcast r@)typerrGN)rQrr@r=rrrCrCrEsend_broadcasts  z#AsyncRealtimeChannel.send_broadcastcCst|jjdS)Nz/api/broadcast)r-r4 http_endpointrRrCrCrErfsz,AsyncRealtimeChannel._broadcast_endpoint_urlcs<|jrdS|j|jIdHtj|_|jIdHdSrH) rqr4_leave_open_topicr5rrvrJr_resendrRrCrCrErs zAsyncRealtimeChannel._rejoincCs|jjo|jSrH)r4rr[rRrCrCrErszAsyncRealtimeChannel._can_pushcs |tj||dIdHdS)N)rrG)rQrrArrCrCrErsz"AsyncRealtimeChannel.send_presencerr)c Cst|jd|t|tr-t|jtr#|jD]}||jqdS|t |jdSt|t r^|j}|j rZ|j |j }r\|jdkrP|tj|jdS|tj|jdSdSdSt|trr|j}|jD]}||qidSt|tr|j|jdSt|tr|j|jdSt|tr|j}|jD]}||qdSt|tr||jdSt|tr|dSt|t rdSt!|dS)Nz : ok)"rUrmr5 isinstancer,rGr+rbrsdictr'rr`popstatustriggerrriresponserrrar&rA_on_state_eventr%_on_diff_eventr"rcrrror r ) r=rrz reply_payloadrQbroadcast_payloadbroadcast_callbackrGrrCrCrE_handle_messagesN                   z$AsyncRealtimeChannel._handle_messagecs*|j|jjr|IdHdSdSrH)rKrXr4rrrRrCrCrEres  z,AsyncRealtimeChannel._rejoin_until_connectedrH)r4r2r5r6r7r8r9r:)rGrp)rzr{r9r3r)rr6rGrr]rr9r0)r9r3)rr6rzrr9r3)NNN) rrrzrrrrrrrr9r3)rzrr9r3)rrr9r:)r9r)rzrr9r3)rzrr9r3)rzrr9r3)rr6rrr9r:)rr))!__name__ __module__ __qualname____doc__rkrorspropertyrrrTrqrxryrrrQr^rrrrrrrrrrrfrrrrrerCrCrCrEr3=sR B       e           (r3)@ __future__rrNrloggingtypingrrrrrrr typing_extensionsr realtime.typesr r r rrrrrrrrrrrrrrrrrrr r!r"r#r$r%r&r'r(r)r*r+r, transformersr-rAr/rQr0timerr1clientr2 getLoggerrrUr3rCrCrCrEs $ LH