Skip to content

Commit

Permalink
fix(sanity): fix race condition introduced by #8120 (#8211)
Browse files Browse the repository at this point in the history
  • Loading branch information
bjoerge authored Jan 8, 2025
1 parent 1e12bc9 commit 27feda0
Show file tree
Hide file tree
Showing 2 changed files with 12 additions and 11 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -65,7 +65,6 @@ export interface Pair {
transactionsPendingEvents$: Observable<PendingMutationsEvent>
published: DocumentVersion
draft: DocumentVersion
_keepalive: Observable<never>
}

function setVersion<T>(version: 'draft' | 'published') {
Expand Down Expand Up @@ -252,8 +251,5 @@ export function checkoutPair(
consistency$: published.consistency$,
remoteSnapshot$: published.remoteSnapshot$.pipe(map(setVersion('published'))),
},
// Use this to keep the mutation pipeline active.
// It won't ever emit any events, but it will prevent the eventsource connection from completing for as long as it is subscribed to
_keepalive: merge(listenerEvents$, commits$).pipe(mergeMap(() => EMPTY)),
}
}
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import {type SanityClient} from '@sanity/client'
import {merge, type Observable, of, ReplaySubject, share, timer} from 'rxjs'
import {EMPTY, merge, Observable, of, ReplaySubject, share, timer} from 'rxjs'
import {mergeMap} from 'rxjs/operators'

import {type PairListenerOptions} from '../getPairListener'
import {type IdPair} from '../types'
Expand All @@ -24,12 +25,16 @@ export const memoizedPair: (
serverActionsEnabled: Observable<boolean>,
pairListenerOptions?: PairListenerOptions,
): Observable<Pair> => {
const pair = checkoutPair(client, idPair, serverActionsEnabled, pairListenerOptions)
return merge(
of(pair),
// makes sure the pair listener is kept alive for as long as there are subscribers
pair._keepalive,
).pipe(
return new Observable<Pair>((subscriber) => {
const pair = checkoutPair(client, idPair, serverActionsEnabled, pairListenerOptions)
return merge(
of(pair),
// merge in draft events and published events to makes sure they receive
// the events they need for as long as the pair is subscribed to
pair.draft.events.pipe(mergeMap(() => EMPTY)),
pair.published.events.pipe(mergeMap(() => EMPTY)),
).subscribe(subscriber)
}).pipe(
share({
connector: () => new ReplaySubject(1),
resetOnComplete: true,
Expand Down

0 comments on commit 27feda0

Please sign in to comment.