| 58 | // invariant before the stream is read) and an empty hasObjects (forces |
| 59 | // applyChanges to throw on a missing object mid-stream). |
| 60 | function trackFetchDisposal( |
| 61 | rpc: SyncRPC, |
| 62 | opts: { appliedPushCursor?: ChangeCursor; failHasObjects?: boolean } = {}, |
| 63 | ): { rpc: SyncRPC; disposeCount: () => number } { |
| 64 | let disposeCount = 0; |
| 65 | const wrapped = new Proxy(rpc as object, { |
| 66 | get(target, prop, receiver) { |
| 67 | if (prop === "fetchChanges") { |
| 68 | return async (...args: Parameters<SyncRPC["fetchChanges"]>) => { |
| 69 | const real = await Reflect.get(target, prop, receiver).call(target, ...args); |
| 70 | return { |
| 71 | ...real, |
| 72 | ...(opts.appliedPushCursor !== undefined |
| 73 | ? { appliedPushCursor: opts.appliedPushCursor } |
| 74 | : {}), |
| 75 | [Symbol.dispose]() { |
| 76 | disposeCount += 1; |
| 77 | }, |
| 78 | }; |
| 79 | }; |
| 80 | } |
| 81 | if (prop === "hasObjects" && opts.failHasObjects) { |
| 82 | return async () => []; |
| 83 | } |
| 84 | return Reflect.get(target, prop, receiver); |
| 85 | }, |
| 86 | }) as SyncRPC; |
| 87 | return { rpc: wrapped, disposeCount: () => disposeCount }; |
| 88 | } |
| 89 | |
| 90 | describe("sync driver — pullOnce", () => { |
| 91 | it("pulls a single entry from upstream", async () => { |