@@ -105,6 +105,8 @@ export const { use: useGlobalSDK, provider: GlobalSDKProvider } = createSimpleCo
105105 const aborted = ( error : unknown ) => abortError . safeParse ( error ) . success
106106
107107 let attempt : AbortController | undefined
108+ let run : Promise < void > | undefined
109+ let started = false
108110 const HEARTBEAT_TIMEOUT_MS = 15_000
109111 let lastEventAt = Date . now ( )
110112 let heartbeat : ReturnType < typeof setTimeout > | undefined
@@ -121,78 +123,93 @@ export const { use: useGlobalSDK, provider: GlobalSDKProvider } = createSimpleCo
121123 heartbeat = undefined
122124 }
123125
124- void ( async ( ) => {
125- while ( ! abort . signal . aborted ) {
126- attempt = new AbortController ( )
127- lastEventAt = Date . now ( )
128- const onAbort = ( ) => {
129- attempt ?. abort ( )
130- }
131- abort . signal . addEventListener ( "abort" , onAbort )
132- try {
133- const events = await eventSdk . global . event ( {
134- signal : attempt . signal ,
135- onSseError : ( error ) => {
136- if ( aborted ( error ) ) return
137- if ( streamErrorLogged ) return
126+ const start = ( ) => {
127+ if ( started ) return run
128+ started = true
129+ run = ( async ( ) => {
130+ while ( ! abort . signal . aborted && started ) {
131+ attempt = new AbortController ( )
132+ lastEventAt = Date . now ( )
133+ const onAbort = ( ) => {
134+ attempt ?. abort ( )
135+ }
136+ abort . signal . addEventListener ( "abort" , onAbort )
137+ try {
138+ const events = await eventSdk . global . event ( {
139+ signal : attempt . signal ,
140+ onSseError : ( error ) => {
141+ if ( aborted ( error ) ) return
142+ if ( streamErrorLogged ) return
143+ streamErrorLogged = true
144+ console . error ( "[global-sdk] event stream error" , {
145+ url : currentServer . http . url ,
146+ fetch : eventFetch ? "platform" : "webview" ,
147+ error,
148+ } )
149+ } ,
150+ } )
151+ let yielded = Date . now ( )
152+ resetHeartbeat ( )
153+ for await ( const event of events . stream ) {
154+ resetHeartbeat ( )
155+ streamErrorLogged = false
156+ const directory = event . directory ?? "global"
157+ const payload = event . payload
158+ const k = key ( directory , payload )
159+ if ( k ) {
160+ const i = coalesced . get ( k )
161+ if ( i !== undefined ) {
162+ queue [ i ] = { directory, payload }
163+ if ( payload . type === "message.part.updated" ) {
164+ const part = payload . properties . part
165+ staleDeltas . add ( deltaKey ( directory , part . messageID , part . id ) )
166+ }
167+ continue
168+ }
169+ coalesced . set ( k , queue . length )
170+ }
171+ queue . push ( { directory, payload } )
172+ schedule ( )
173+
174+ if ( Date . now ( ) - yielded < STREAM_YIELD_MS ) continue
175+ yielded = Date . now ( )
176+ await wait ( 0 )
177+ }
178+ } catch ( error ) {
179+ if ( ! aborted ( error ) && ! streamErrorLogged ) {
138180 streamErrorLogged = true
139- console . error ( "[global-sdk] event stream error " , {
181+ console . error ( "[global-sdk] event stream failed " , {
140182 url : currentServer . http . url ,
141183 fetch : eventFetch ? "platform" : "webview" ,
142184 error,
143185 } )
144- } ,
145- } )
146- let yielded = Date . now ( )
147- resetHeartbeat ( )
148- for await ( const event of events . stream ) {
149- resetHeartbeat ( )
150- streamErrorLogged = false
151- const directory = event . directory ?? "global"
152- const payload = event . payload
153- const k = key ( directory , payload )
154- if ( k ) {
155- const i = coalesced . get ( k )
156- if ( i !== undefined ) {
157- queue [ i ] = { directory, payload }
158- if ( payload . type === "message.part.updated" ) {
159- const part = payload . properties . part
160- staleDeltas . add ( deltaKey ( directory , part . messageID , part . id ) )
161- }
162- continue
163- }
164- coalesced . set ( k , queue . length )
165186 }
166- queue . push ( { directory, payload } )
167- schedule ( )
168-
169- if ( Date . now ( ) - yielded < STREAM_YIELD_MS ) continue
170- yielded = Date . now ( )
171- await wait ( 0 )
172- }
173- } catch ( error ) {
174- if ( ! aborted ( error ) && ! streamErrorLogged ) {
175- streamErrorLogged = true
176- console . error ( "[global-sdk] event stream failed" , {
177- url : currentServer . http . url ,
178- fetch : eventFetch ? "platform" : "webview" ,
179- error,
180- } )
187+ } finally {
188+ abort . signal . removeEventListener ( "abort" , onAbort )
189+ attempt = undefined
190+ clearHeartbeat ( )
181191 }
182- } finally {
183- abort . signal . removeEventListener ( "abort" , onAbort )
184- attempt = undefined
185- clearHeartbeat ( )
192+
193+ if ( abort . signal . aborted || ! started ) return
194+ await wait ( RECONNECT_DELAY_MS )
186195 }
196+ } ) ( ) . finally ( ( ) => {
197+ run = undefined
198+ flush ( )
199+ } )
200+ return run
201+ }
187202
188- if ( abort . signal . aborted ) return
189- await wait ( RECONNECT_DELAY_MS )
190- }
191- } ) ( ) . finally ( flush )
203+ const stop = ( ) => {
204+ started = false
205+ attempt ?. abort ( )
206+ clearHeartbeat ( )
207+ }
192208
193209 const onVisibility = ( ) => {
194210 if ( typeof document === "undefined" ) return
195211 if ( document . visibilityState !== "visible" ) return
212+ if ( ! started ) return
196213 if ( Date . now ( ) - lastEventAt < HEARTBEAT_TIMEOUT_MS ) return
197214 attempt ?. abort ( )
198215 }
@@ -204,6 +221,7 @@ export const { use: useGlobalSDK, provider: GlobalSDKProvider } = createSimpleCo
204221 if ( typeof document !== "undefined" ) {
205222 document . removeEventListener ( "visibilitychange" , onVisibility )
206223 }
224+ stop ( )
207225 abort . abort ( )
208226 flush ( )
209227 } )
@@ -217,7 +235,11 @@ export const { use: useGlobalSDK, provider: GlobalSDKProvider } = createSimpleCo
217235 return {
218236 url : currentServer . http . url ,
219237 client : sdk ,
220- event : emitter ,
238+ event : {
239+ on : emitter . on . bind ( emitter ) ,
240+ listen : emitter . listen . bind ( emitter ) ,
241+ start,
242+ } ,
221243 createClient ( opts : Omit < Parameters < typeof createSdkForServer > [ 0 ] , "server" | "fetch" > ) {
222244 const s = server . current
223245 if ( ! s ) throw new Error ( language . t ( "error.globalSDK.serverNotAvailable" ) )
0 commit comments