Skip to content

Commit

Permalink
fixup! fix(sanity): fix race condition regression introduced by #8120
Browse files Browse the repository at this point in the history
  • Loading branch information
bjoerge committed Jan 8, 2025
1 parent ae855bd commit ce0badc
Show file tree
Hide file tree
Showing 2 changed files with 11 additions and 11 deletions.
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@ import {type Action, type SanityClient} from '@sanity/client'
import {type Mutation} from '@sanity/mutator'
import {type SanityDocument} from '@sanity/types'
import {omit} from 'lodash'
import {EMPTY, from, merge, type Observable, Subject} from 'rxjs'
import {EMPTY, from, merge, type Observable} from 'rxjs'
import {filter, map, mergeMap, share, take, tap} from 'rxjs/operators'

import {
Expand Down Expand Up @@ -65,7 +65,6 @@ export interface Pair {
transactionsPendingEvents$: Observable<PendingMutationsEvent>
published: DocumentVersion
draft: DocumentVersion
complete: () => void
}

function setVersion<T>(version: 'draft' | 'published') {
Expand Down Expand Up @@ -204,10 +203,7 @@ export function checkoutPair(
): Pair {
const {publishedId, draftId} = idPair

const listenerEventsConnector = new Subject<ListenerEvent>()
const listenerEvents$ = getPairListener(client, idPair, pairListenerOptions).pipe(
share({connector: () => listenerEventsConnector}),
)
const listenerEvents$ = getPairListener(client, idPair, pairListenerOptions).pipe(share())

const reconnect$ = listenerEvents$.pipe(
filter((ev) => ev.type === 'reconnect'),
Expand Down Expand Up @@ -255,6 +251,5 @@ export function checkoutPair(
consistency$: published.consistency$,
remoteSnapshot$: published.remoteSnapshot$.pipe(map(setVersion('published'))),
},
complete: () => listenerEventsConnector.complete(),
}
}
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
import {type SanityClient} from '@sanity/client'
import {Observable, 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 @@ -26,9 +27,13 @@ export const memoizedPair: (
): Observable<Pair> => {
return new Observable<Pair>((subscriber) => {
const pair = checkoutPair(client, idPair, serverActionsEnabled, pairListenerOptions)
subscriber.next(pair)

return pair.complete
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),
Expand Down

0 comments on commit ce0badc

Please sign in to comment.