9677127838

Committed 26 Jun 2024 09:27AM UTC coverage: 87.744%. First build

Build # 9677127838

Build Type

Pull #6743

github

Committed by

web-flow

Commit Message

Merge 89fa94501 into b4734155c

Pull Request Pull Request #6743: RJS-2636: Add progress notifications and tests

Run Details

1247 of 1495 branches covered (83.41%)

Branch coverage included in aggregate %.

16 of 25 new or added lines in 2 files covered. (64.0%)

2934 of 3270 relevant lines covered (89.72%)

1133.71 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

82.14

/packages/realm/src/ProgressRealmPromise.ts

////////////////////////////////////////////////////////////////////////////
//
// Copyright 2022 Realm Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
////////////////////////////////////////////////////////////////////////////

import {
  Configuration,
  DynamicProgressNotificationCallback,
  OpenRealmBehaviorType,
  OpenRealmTimeOutBehavior,
  PartitionBasedSyncProgressNotificationCallback,
  ProgressNotificationCallback,
  PromiseHandle,
  Realm,
  SubscriptionSetState,
  TimeoutError,
  TimeoutPromise,
  assert,
  binding,
  flags,
  validateConfiguration,
} from "./internal";

type OpenBehavior = {
  openBehavior: OpenRealmBehaviorType;
  timeOut?: number;
  timeOutBehavior?: OpenRealmTimeOutBehavior;
};

function determineBehavior(config: Configuration, realmExists: boolean): OpenBehavior {
  const { sync, openSyncedRealmLocally } = config;
  if (!sync || openSyncedRealmLocally) {
    return { openBehavior: OpenRealmBehaviorType.OpenImmediately };
  } else {
    const configProperty = realmExists ? "existingRealmFileBehavior" : "newRealmFileBehavior";
    const configBehavior = sync[configProperty];
    if (configBehavior) {
      const { type, timeOut, timeOutBehavior } = configBehavior;
      if (typeof timeOut !== "undefined") {
        assert.number(timeOut, "timeOut");
      }
      return { openBehavior: type, timeOut, timeOutBehavior };
    } else {
      return {
        openBehavior: OpenRealmBehaviorType.DownloadBeforeOpen,
        timeOut: 30 * 1000,
        timeOutBehavior: OpenRealmTimeOutBehavior.ThrowException,
      };
    }
  }
}

export class ProgressRealmPromise implements Promise<Realm> {
  /** @internal */
  private static instances = new Set<binding.WeakRef<ProgressRealmPromise>>();
  /**
   * Cancels all unresolved `ProgressRealmPromise` instances.
   * @internal
   */
  public static cancelAll() {
    for (const promiseRef of ProgressRealmPromise.instances) {
      promiseRef.deref()?.cancel();
    }
    ProgressRealmPromise.instances.clear();
  }
  /** @internal */
  private task: binding.AsyncOpenTask | null = null;
  /** @internal */
  private listeners = new Set<ProgressNotificationCallback>();
  /** @internal */
  private handle = new PromiseHandle<Realm>();
  /** @internal */
  private timeoutPromise: TimeoutPromise<Realm> | null = null;
  /** @internal */
  private token = 0;

  /** @internal */
  constructor(config: Configuration) {
    if (flags.ALLOW_CLEAR_TEST_STATE) {
      ProgressRealmPromise.instances.add(new binding.WeakRef(this));
    }
    try {
      validateConfiguration(config);
      // Calling `Realm.exists()` before `binding.Realm.getSynchronizedRealm()` is necessary to capture
      // the correct value when this constructor was called since `binding.Realm.getSynchronizedRealm()`
      // will open the realm. This is needed when calling the Realm constructor.
      const realmExists = Realm.exists(config);
      const { openBehavior, timeOut, timeOutBehavior } = determineBehavior(config, realmExists);
      if (openBehavior === OpenRealmBehaviorType.OpenImmediately) {
        const realm = new Realm(config);
        this.handle.resolve(realm);
      } else if (openBehavior === OpenRealmBehaviorType.DownloadBeforeOpen) {
        const { bindingConfig } = Realm.transformConfig(config);

        // Construct an async open task
        this.task = binding.Realm.getSynchronizedRealm(bindingConfig);
        // If the promise handle gets rejected, we should cancel the open task
        // to avoid consuming a thread safe reference which is no longer registered
        this.handle.promise.catch(() => this.task?.cancel());

        this.createTimeoutPromise(config, { openBehavior, timeOut, timeOutBehavior });

        this.task
          .start()
          .then(async (tsr) => {
            const realm = new Realm(config, {
              internal: binding.Helpers.consumeThreadSafeReferenceToSharedRealm(tsr),
              // Do not call `Realm.exists()` here in case the realm has been opened by this point in time.
              realmExists,
            });
            if (config.sync?.flexible && !config.openSyncedRealmLocally) {
              const { subscriptions } = realm;
              if (subscriptions.state === SubscriptionSetState.Pending) {
                await subscriptions.waitForSynchronization();
              }
            }
            return realm;
          })
          .then(this.handle.resolve, (err) => {
            assert.undefined(err.code, "Update this to use the error code instead of matching on message");
            if (err instanceof Error && err.message === "Sync session became inactive") {
              // This can happen when two async tasks are opened for the same Realm and one gets canceled
              this.rejectAsCanceled();
            } else {
              this.handle.reject(err);
            }
          });
        if (this.listeners.size > 0) {
          this.token = this.task.registerDownloadProgressNotifier(this.emitProgress);
        }
      } else {
        throw new Error(`Unexpected open behavior '${openBehavior}'`);
      }
    } catch (err) {
      if (this.token !== 0) {
        this.task?.unregisterDownloadProgressNotifier(this.token);
      }
      this.handle.reject(err);
    }
  }

  /**
   * Cancels the download of the Realm
   * If multiple `ProgressRealmPromise` instances are in progress for the same Realm, then canceling one of them
   * will cancel all of them.
   */
  cancel(): void {
    this.cancelAndResetTask();
    this.timeoutPromise?.cancel();
    this.task?.unregisterDownloadProgressNotifier(this.token);
    // Clearing all listeners to avoid accidental progress notifications
    this.listeners.clear();
    // Tell anything awaiting the promise
    this.rejectAsCanceled();
  }

  /**
   * Register to receive progress notifications while the download is in progress.
   * @param callback Called multiple times as the client receives data, with two arguments:
   * 1. `transferred` The current number of bytes already transferred
   * 2. `transferable` The total number of transferable bytes (i.e. the number of bytes already transferred plus the number of bytes pending transfer)
   */
  progress(callback: ProgressNotificationCallback): this {
    this.listeners.add(callback);
    if (callback.length === 1) {
      const estimateCallback = callback as DynamicProgressNotificationCallback;
      estimateCallback(0.0);
    } else {
      const pbsCallback = callback as PartitionBasedSyncProgressNotificationCallback;
      pbsCallback(0.0, 0.0);
    }
    return this;
  }

  then = this.handle.promise.then.bind(this.handle.promise);
  catch = this.handle.promise.catch.bind(this.handle.promise);
  finally = this.handle.promise.finally.bind(this.handle.promise);

  private emitProgress = (transferredArg: binding.Int64, transferableArg: binding.Int64, progressEstimate: number) => {
    const transferred = binding.Int64.intToNum(transferredArg);
    const transferable = binding.Int64.intToNum(transferableArg);
    for (const listener of this.listeners) {
      if (listener.length === 1) {
        const estimateListener = listener as DynamicProgressNotificationCallback;
        estimateListener(progressEstimate);
      } else {
        const pbsListener = listener as PartitionBasedSyncProgressNotificationCallback;
        pbsListener(transferred, transferable);
      }
    }
  };

  private createTimeoutPromise(config: Configuration, { timeOut, timeOutBehavior }: OpenBehavior) {
    if (typeof timeOut === "number") {
      this.timeoutPromise = new TimeoutPromise(
        this.handle.promise, // Ensures the timeout gets cancelled when the realm opens
        {
          ms: timeOut,
          message: `Realm could not be downloaded in the allocated time: ${timeOut} ms.`,
        },
      );
      if (timeOutBehavior === OpenRealmTimeOutBehavior.ThrowException) {
        // Make failing the timeout, reject the promise
        this.timeoutPromise.catch(this.handle.reject);
      } else if (timeOutBehavior === OpenRealmTimeOutBehavior.OpenLocalRealm) {
        // Make failing the timeout, resolve the promise
        this.timeoutPromise.catch((err) => {
          if (err instanceof TimeoutError) {
            this.cancelAndResetTask();
            const realm = new Realm(config);
            this.handle.resolve(realm);
          } else {
            this.handle.reject(err);
          }
        });
      } else {
        throw new Error(
          `Invalid 'timeOutBehavior': '${timeOutBehavior}'. Only 'throwException' and 'openLocalRealm' is allowed.`,
        );
      }
    }
  }

  private cancelAndResetTask() {
    if (this.task) {
      this.task.cancel();
      this.task.$resetSharedPtr();
      this.task = null;
    }
  }

  private rejectAsCanceled() {
    const err = new Error("Async open canceled");
    this.handle.reject(err);
  }

  get [Symbol.toStringTag]() {
    return ProgressRealmPromise.name;
  }
}

1	////////////////////////////////////////////////////////////////////////////
2	//
3	// Copyright 2022 Realm Inc.
4	//
5	// Licensed under the Apache License, Version 2.0 (the "License");
6	// you may not use this file except in compliance with the License.
7	// You may obtain a copy of the License at
8	//
9	// http://www.apache.org/licenses/LICENSE-2.0
10	//
11	// Unless required by applicable law or agreed to in writing, software
12	// distributed under the License is distributed on an "AS IS" BASIS,
13	// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14	// See the License for the specific language governing permissions and
15	// limitations under the License.
16	//
17	////////////////////////////////////////////////////////////////////////////
18
19	import {	1✔
20	Configuration,
21	DynamicProgressNotificationCallback,
22	OpenRealmBehaviorType,
23	OpenRealmTimeOutBehavior,
24	PartitionBasedSyncProgressNotificationCallback,
25	ProgressNotificationCallback,
26	PromiseHandle,
27	Realm,
28	SubscriptionSetState,
29	TimeoutError,
30	TimeoutPromise,
31	assert,
32	binding,
33	flags,
34	validateConfiguration,
35	} from "./internal";
36
37	type OpenBehavior = {
38	openBehavior: OpenRealmBehaviorType;
39	timeOut?: number;
40	timeOutBehavior?: OpenRealmTimeOutBehavior;
41	};
42
43	function determineBehavior(config: Configuration, realmExists: boolean): OpenBehavior {
44	const { sync, openSyncedRealmLocally } = config;	741✔
45	if (!sync \|\| openSyncedRealmLocally) {	741✔
46	return { openBehavior: OpenRealmBehaviorType.OpenImmediately };	425✔
47	} else {
48	const configProperty = realmExists ? "existingRealmFileBehavior" : "newRealmFileBehavior";	316✔
49	const configBehavior = sync[configProperty];	316✔
50	if (configBehavior) {	316✔
51	const { type, timeOut, timeOutBehavior } = configBehavior;	15✔
52	if (typeof timeOut !== "undefined") {	15✔
53	assert.number(timeOut, "timeOut");	5✔
54	}
55	return { openBehavior: type, timeOut, timeOutBehavior };	15✔
56	} else {
57	return {	301✔
58	openBehavior: OpenRealmBehaviorType.DownloadBeforeOpen,
59	timeOut: 30 * 1000,
60	timeOutBehavior: OpenRealmTimeOutBehavior.ThrowException,
61	};
62	}
63	}
64	}
65
66	export class ProgressRealmPromise implements Promise<Realm> {	1✔
67	/** @internal */
68	private static instances = new Set<binding.WeakRef<ProgressRealmPromise>>();	1✔
69	/**
70	* Cancels all unresolved `ProgressRealmPromise` instances.
71	* @internal
72	*/
73	public static cancelAll() {
74	for (const promiseRef of ProgressRealmPromise.instances) {	975✔
75	promiseRef.deref()?.cancel();	748✔
76	}
77	ProgressRealmPromise.instances.clear();	975✔
78	}
79	/** @internal */
80	private task: binding.AsyncOpenTask \| null = null;	748✔
81	/** @internal */
82	private listeners = new Set<ProgressNotificationCallback>();	748✔
83	/** @internal */
84	private handle = new PromiseHandle<Realm>();	748✔
85	/** @internal */
86	private timeoutPromise: TimeoutPromise<Realm> \| null = null;	748✔
87	/** @internal */
88	private token = 0;	748✔
89
90	/** @internal */
91	constructor(config: Configuration) {
92	if (flags.ALLOW_CLEAR_TEST_STATE) {	748!
93	ProgressRealmPromise.instances.add(new binding.WeakRef(this));	748✔
94	}
95	try {	748✔
96	validateConfiguration(config);	748✔
97	// Calling `Realm.exists()` before `binding.Realm.getSynchronizedRealm()` is necessary to capture
98	// the correct value when this constructor was called since `binding.Realm.getSynchronizedRealm()`
99	// will open the realm. This is needed when calling the Realm constructor.
100	const realmExists = Realm.exists(config);	741✔
101	const { openBehavior, timeOut, timeOutBehavior } = determineBehavior(config, realmExists);	741✔
102	if (openBehavior === OpenRealmBehaviorType.OpenImmediately) {	741✔
103	const realm = new Realm(config);	430✔
104	this.handle.resolve(realm);	430✔
105	} else if (openBehavior === OpenRealmBehaviorType.DownloadBeforeOpen) {	311!
106	const { bindingConfig } = Realm.transformConfig(config);	311✔
107
108	// Construct an async open task
109	this.task = binding.Realm.getSynchronizedRealm(bindingConfig);	311✔
110	// If the promise handle gets rejected, we should cancel the open task
111	// to avoid consuming a thread safe reference which is no longer registered
112	this.handle.promise.catch(() => this.task?.cancel());	311✔
113
114	this.createTimeoutPromise(config, { openBehavior, timeOut, timeOutBehavior });	311✔
115
116	this.task	311✔
117	.start()
118	.then(async (tsr) => {
119	const realm = new Realm(config, {	299✔
120	internal: binding.Helpers.consumeThreadSafeReferenceToSharedRealm(tsr),
121	// Do not call `Realm.exists()` here in case the realm has been opened by this point in time.
122	realmExists,
123	});
124	if (config.sync?.flexible && !config.openSyncedRealmLocally) {	299✔
125	const { subscriptions } = realm;	226✔
126	if (subscriptions.state === SubscriptionSetState.Pending) {	226✔
127	await subscriptions.waitForSynchronization();	7✔
128	}
129	}
130	return realm;	299✔
131	})
132	.then(this.handle.resolve, (err) => {
133	assert.undefined(err.code, "Update this to use the error code instead of matching on message");	3✔
134	if (err instanceof Error && err.message === "Sync session became inactive") {	3✔
135	// This can happen when two async tasks are opened for the same Realm and one gets canceled
136	this.rejectAsCanceled();	1✔
137	} else {
138	this.handle.reject(err);	2✔
139	}
140	});
141	if (this.listeners.size > 0) {	311!
NEW 142	this.token = this.task.registerDownloadProgressNotifier(this.emitProgress);	×
143	}
144	} else {
145	throw new Error(`Unexpected open behavior '${openBehavior}'`);	×
146	}
147	} catch (err) {
148	if (this.token !== 0) {	7!
NEW 149	this.task?.unregisterDownloadProgressNotifier(this.token);	×
150	}
151	this.handle.reject(err);	7✔
152	}
153	}
154
155	/**
156	* Cancels the download of the Realm
157	* If multiple `ProgressRealmPromise` instances are in progress for the same Realm, then canceling one of them
158	* will cancel all of them.
159	*/
160	cancel(): void {
161	this.cancelAndResetTask();	652✔
162	this.timeoutPromise?.cancel();	652✔
163	this.task?.unregisterDownloadProgressNotifier(this.token);	652✔
164	// Clearing all listeners to avoid accidental progress notifications
165	this.listeners.clear();	652✔
166	// Tell anything awaiting the promise
167	this.rejectAsCanceled();	652✔
168	}
169
170	/**
171	* Register to receive progress notifications while the download is in progress.
172	* @param callback Called multiple times as the client receives data, with two arguments:
173	* 1. `transferred` The current number of bytes already transferred
174	* 2. `transferable` The total number of transferable bytes (i.e. the number of bytes already transferred plus the number of bytes pending transfer)
175	*/
176	progress(callback: ProgressNotificationCallback): this {
177	this.listeners.add(callback);	2✔
178	if (callback.length === 1) {	2!
NEW 179	const estimateCallback = callback as DynamicProgressNotificationCallback;	×
NEW 180	estimateCallback(0.0);	×
181	} else {
182	const pbsCallback = callback as PartitionBasedSyncProgressNotificationCallback;	2✔
183	pbsCallback(0.0, 0.0);	2✔
184	}
185	return this;	2✔
186	}
187
188	then = this.handle.promise.then.bind(this.handle.promise);	748✔
189	catch = this.handle.promise.catch.bind(this.handle.promise);	748✔
190	finally = this.handle.promise.finally.bind(this.handle.promise);	748✔
191
192	private emitProgress = (transferredArg: binding.Int64, transferableArg: binding.Int64, progressEstimate: number) => {	748✔
193	const transferred = binding.Int64.intToNum(transferredArg);	×
194	const transferable = binding.Int64.intToNum(transferableArg);	×
195	for (const listener of this.listeners) {	×
NEW 196	if (listener.length === 1) {	×
NEW 197	const estimateListener = listener as DynamicProgressNotificationCallback;	×
NEW 198	estimateListener(progressEstimate);	×
199	} else {
NEW 200	const pbsListener = listener as PartitionBasedSyncProgressNotificationCallback;	×
NEW 201	pbsListener(transferred, transferable);	×
202	}
203	}
204	};
205
206	private createTimeoutPromise(config: Configuration, { timeOut, timeOutBehavior }: OpenBehavior) {
207	if (typeof timeOut === "number") {	311✔
208	this.timeoutPromise = new TimeoutPromise(	306✔
209	this.handle.promise, // Ensures the timeout gets cancelled when the realm opens
210	{
211	ms: timeOut,
212	message: `Realm could not be downloaded in the allocated time: ${timeOut} ms.`,
213	},
214	);
215	if (timeOutBehavior === OpenRealmTimeOutBehavior.ThrowException) {	306✔
216	// Make failing the timeout, reject the promise
217	this.timeoutPromise.catch(this.handle.reject);	303✔
218	} else if (timeOutBehavior === OpenRealmTimeOutBehavior.OpenLocalRealm) {	3!
219	// Make failing the timeout, resolve the promise
220	this.timeoutPromise.catch((err) => {	3✔
221	if (err instanceof TimeoutError) {	2!
222	this.cancelAndResetTask();	2✔
223	const realm = new Realm(config);	2✔
224	this.handle.resolve(realm);	2✔
225	} else {
226	this.handle.reject(err);	×
227	}
228	});
229	} else {
230	throw new Error(	×
231	`Invalid 'timeOutBehavior': '${timeOutBehavior}'. Only 'throwException' and 'openLocalRealm' is allowed.`,
232	);
233	}
234	}
235	}
236
237	private cancelAndResetTask() {
238	if (this.task) {	654✔
239	this.task.cancel();	231✔
240	this.task.$resetSharedPtr();	231✔
241	this.task = null;	231✔
242	}
243	}
244
245	private rejectAsCanceled() {
246	const err = new Error("Async open canceled");	653✔
247	this.handle.reject(err);	653✔
248	}
249
250	get [Symbol.toStringTag]() {
251	return ProgressRealmPromise.name;	×
252	}
253	}

realm / realm-js / 9677127838

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous