o i3N@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@sHeZdZdZ d_d`d d Zd dZdaddZeddZeddZ eddZ eddZ eddZ d_dbdd Z dcd!d"Z d_ddd(d)Zded*d+Zdfd-d.Z   dgdhd5d6Zdid8d9Zdjd;d<Zdcd=d>Zdkd@dAZdldCdDZdmdFdGZdndIdJZdodMdNZdcdOdPZdQdRZdcdSdTZdUdVZdodWdXZdpd[d\Z d]d^Z!dS)qAsyncRealtimeChannela 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 dddddddddi_|_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)ackselfkeyenabled) broadcastpresenceprivateNcSsd|S)Nr)triesrErEU/home/ubuntu/transcripts/venv/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)rIpushr=rErGon_join_push_okss    z6AsyncRealtimeChannel.__init__..on_join_push_okcs2jsdStdjtj_jdS)Nzjoin push timeout for channel ) is_joiningloggererrorr5rERROREDrLrMschedule_timeoutrErTrErGon_join_push_timeoutzs z;AsyncRealtimeChannel.__init__..on_join_push_timeoutrIr()r4r7r5 _joined_oncer/rCrCLOSEDrLrOtimeoutr0rjoin join_pushmessages_waiting_for_ackbroadcast_callbackssystem_callbackspostgres_changes_callbackssubscribe_callbackr1_rejoin_until_connectedrM_broadcast_endpoint_urlbroadcast_endpoint_urlreceiverOkTimeout)r=r4r5r7rUr[rErTrG__init__DsB      zAsyncRealtimeChannel.__init__cCs6td|jd|jtj|_|j |dS)Nchannel z closed) rWinfor5rMrNrr^rLr4_remove_channelrTrErErGon_closes zAsyncRealtimeChannel.on_closerIdict[str, Any]cCs>|js|jrdStd|jd|tj|_|j dS)Nrnz error: ) is_leaving is_closedrWror5rrYrLrMrZ)r=rIrErErGon_errors zAsyncRealtimeChannel.on_errorcC |jtjkSrJ)rLrr^rTrErErGrt zAsyncRealtimeChannel.is_closedcCrvrJ)rLrJOININGrTrErErGrVrwzAsyncRealtimeChannel.is_joiningcCrvrJ)rLrLEAVINGrTrErErGrsrwzAsyncRealtimeChannel.is_leavingcCrvrJ)rLrrYrTrErErG is_erroredrwzAsyncRealtimeChannel.is_erroredcCrvrJ)rLrrKrTrErErG is_joinedrwzAsyncRealtimeChannel.is_joinedcallbackHOptional[Callable[[RealtimeSubscribeStates, Optional[Exception]], None]]c sjjs jIdHjrtdjd}|d}|dp(tddd}|d d}jj p8|d d}||d <d|||d d j Dd i}jj rVjj |d<j |d_dfdd }dfdd } fdd} j tj|tj| tj| 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;rBrCr>Fr?rDrAcSg|]}|jqSrEbinding_filter.0crErErG z2AsyncRealtimeChannel.subscribe..rBrCrDpostgres_changes access_tokenTrIr(cs|j}g}|rbtjD]U\}}|t|kr||nd}t|d||rJ|j|jkrJ|j|jkrJ|j |j krJ|j |j krJ|j |_ | |q t o^tjtddSdS|_ootjddSdS)Nz, z@mismatch between server and client bindings for postgres changes)r enumeraterelenrWdebugeventschema_schematablefilteridappendrPrQ unsubscriber CHANNEL_ERROR Exception SUBSCRIBED)rIserver_postgres_changesnew_postgres_bindingsipostgres_callbackserver_bindingr|r=rErGrUs>      z7AsyncRealtimeChannel.subscribe..on_join_push_okDict[str, Any]cs$otjtt|dSdSrJ)rrrjsondumps)rIr|rErGon_join_push_errors z:AsyncRealtimeChannel.subscribe..on_join_push_errorcso tjddSdSrJ)r TIMED_OUTargsrrErGr[ sz.on_join_push_timeoutr\)rIr)r4 is_connectedconnectr]rr7getrrC_has_callback_attachedrerraupdate_payloadrjrrkErrorrl_rejoin) r=r|r;rBrCrDpresence_enabledconfig_payloadrUrr[rErrG subscribesR       &  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)Nrnz leave)rWror5rqrrTrErG_close s z0AsyncRealtimeChannel.unsubscribe.._closeNr9r:)rryrLrMrNradestroyr0rleaverjrrkrrR)r=r leave_pushrErTrGrs  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_pushrRref start_timeoutrOr)r=rrIr_rSrErErGrS*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;rBrCrDFcSr~rErrrErErGrWrz-AsyncRealtimeChannel.join..r)r5rrIrN) r7rrer!r5rr`r4 _make_refrR)r=r;rBrCrDrmessagerErErGr`Gs,     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 )r|r)rcrr )r=rr|rErErG on_broadcastes 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 )r|rrrr)rrer)r=rr|rrrrErErGon_postgres_changests   z(AsyncRealtimeChannel.on_postgres_changes&Callable[[SuccessSystemPayload], None]cCs|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. )rdrr=r|rErErG on_systems 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=rrErErGrszAsyncRealtimeChannel.trackcs|diIdHdS)z> Stop tracking presence for the current user. untrackNrrTrErErGrszAsyncRealtimeChannel.untrackrcCs|jjS)z Get the current state of presence on this channel. :return: Dictionary mapping presence keys to lists of presence payloads )rCrLrTrErErGpresence_statesz#AsyncRealtimeChannel.presence_stateCallable[[], None]cC8|j||jrtd|jdt||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. rnB resubscribe due to change in presence callbacks on joined channel) rCon_syncr{rWror5rPrQ _resubscriberrErErGon_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. rnr) rCon_joinr{rWror5rPrQrrrErErGon_presence_joins  z%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. rnr) rCon_leaver{rWror5rPrQrrrErErGon_presence_leaves  z&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 rB)typerrIN)rSrrBr=rrrErErGsend_broadcasts  z#AsyncRealtimeChannel.send_broadcastcs"|IdH|IdHdSrJ)rrrTrErErGrsz!AsyncRealtimeChannel._resubscribecCst|jjdS)Nz/api/broadcast)r-r4 http_endpointrTrErErGrhsz,AsyncRealtimeChannel._broadcast_endpoint_urlcs:|jrdStd|jtj|_|jIdHdS)Nz&Rejoining channel after reconnection: ) rsrWrr5rrxrLraresendrTrErErGrs zAsyncRealtimeChannel._rejoincCs|jjo|jSrJ)r4rr]rTrErErGrszAsyncRealtimeChannel._can_pushcs |tj||dIdHdS)N)rrI)rSrrCrrErErGrsz"AsyncRealtimeChannel.send_presencerr)c Cst|jd|t|tr-t|jtr#|jD]}||jqdS|t |jdSt|t r_|j}|j r[|j |j d}r]|jdkrQ|tj|jdS|tj|jdSdSdSt|trs|j}|jD]}||qjdSt|tr|j|jdSt|tr|j|jdSt|tr|j}|jD]}||qdSt|tr||jdSt|tr|dSt|t rdSt!|dS)Nz : ok)"rWrr5 isinstancer,rIr+rdrudictr'rrbpopstatustriggerrrkresponserrrcr&rC_on_state_eventr%_on_diff_eventr"rerrrqr r ) r=rr| reply_payloadrSbroadcast_payloadbroadcast_callbackrIrrErErG_handle_messagesN                   z$AsyncRealtimeChannel._handle_messagecs*|j|jjr|IdHdSdSrJ)rMrZr4rrrTrErErGrg.s  z,AsyncRealtimeChannel._rejoin_until_connectedrJ)r4r2r5r6r7r8r9r:)rIrr)r|r}r9r3r)rr6rIrr_rr9r0)r9r3)rr6r|rr9r3)NNN) rrr|rrrrrrrr9r3)r|rr9r3)rrr9r:)r9r)r|rr9r3)r|rr9r3)r|rr9r3)rr6rrr9r:)rr))"__name__ __module__ __qualname____doc__rmrqrupropertyrtrVrsrzr{rrrSr`rrrrrrrrrrrrhrrrrrgrErErErGr3=sT B       l              (r3)@ __future__rrPrloggingtypingrrrrrrr typing_extensionsr realtime.typesr r r rrrrrrrrrrrrrrrrrrr r!r"r#r$r%r&r'r(r)r*r+r, transformersr-rCr/rSr0timerr1clientr2 getLoggerrrWr3rErErErGs $ LH