RxFire Realtime Database

packages/rxfire/database/fromRef.ts

@@ -0,0 +1,69 @@
+/**
+ * Copyright 2018 Google 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 { database } from 'firebase';
+import { Observable } from 'rxjs';
+import { map, delay, share } from 'rxjs/operators';
+import { ListenEvent, SnapshotPrevKey } from './interfaces';
+
+/**
+ * Create an observable from a Database Reference or Database Query.
+ * @param ref Database Reference
+ * @param event Listen event type ('value', 'added', 'changed', 'removed', 'moved')
+ */
+export function fromRef(
+  ref: database.Query,
+  event: ListenEvent,
+  listenType = 'on'
+): Observable<SnapshotPrevKey> {
+  return new Observable<SnapshotPrevKey>(subscriber => {
+    const fn = ref[listenType](
+      event,
+      (snapshot, prevKey) => {
+        subscriber.next({ snapshot, prevKey, event });
+        if (listenType == 'once') {
+          subscriber.complete();
+        }
+      },
+      subscriber.error.bind(subscriber)
+    );
+    if (listenType == 'on') {
+      return {
+        unsubscribe() {
+          ref.off(event, fn);
+        }
+      };
+    } else {
+      return { unsubscribe() {} };
+    }
+  }).pipe(
+    // Ensures subscribe on observable is async. This handles
+    // a quirk in the SDK where on/once callbacks can happen
+    // synchronously.
+    delay(0),
+    share()
+  );
+}
+
+export const unwrap = () =>
+  map((payload: SnapshotPrevKey) => {
+    const { snapshot, prevKey } = payload;
+    let key: string | null = null;
+    if (snapshot.exists()) {
+      key = snapshot.key;
+    }
+    return { type: event, payload: snapshot, prevKey, key };
+  });

packages/rxfire/database/index.ts

@@ -0,0 +1,20 @@
+/**
+ * Copyright 2018 Google 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.
+ */
+
+export * from './fromRef';
+export * from './interfaces';
+export * from './list';
+export * from './object';

packages/rxfire/database/interfaces.ts

@@ -0,0 +1,31 @@
+/**
+ * Copyright 2018 Google 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 { database } from 'firebase';
+
+export type QueryFn = (ref: database.Reference) => database.Query;
+export type ChildEvent =
+  | 'child_added'
+  | 'child_removed'
+  | 'child_changed'
+  | 'child_moved';
+export type ListenEvent = 'value' | ChildEvent;
+
+export interface SnapshotPrevKey {
+  snapshot: database.DataSnapshot;
+  prevKey: string | null | undefined;
+  event: ListenEvent;
+}

packages/rxfire/database/list/audit-trail.ts

@@ -0,0 +1,82 @@
+/**
+ * Copyright 2018 Google 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 { database } from 'firebase';
+import { Observable } from 'rxjs';
+import { SnapshotPrevKey, ChildEvent } from '../interfaces';
+import { fromRef } from '../fromRef';
+import { map, withLatestFrom, scan, skipWhile } from 'rxjs/operators';
+import { stateChanges } from './';
+
+interface LoadedMetadata {
+  data: SnapshotPrevKey;
+  lastKeyToLoad: any;
+}
+
+export function auditTrail(
+  query: database.Query,
+  events?: ChildEvent[]
+): Observable<SnapshotPrevKey[]> {
+  const auditTrail$ = stateChanges(query, events).pipe(
+    scan<SnapshotPrevKey>((current, changes) => [...current, changes], [])
+  );
+  return waitForLoaded(query, auditTrail$);
+}
+
+function loadedData(query: database.Query): Observable<LoadedMetadata> {
+  // Create an observable of loaded values to retrieve the
+  // known dataset. This will allow us to know what key to
+  // emit the "whole" array at when listening for child events.
+  return fromRef(query, 'value').pipe(
+    map(data => {
+      // Store the last key in the data set
+      let lastKeyToLoad;
+      // Loop through loaded dataset to find the last key
+      data.snapshot.forEach(child => {
+        lastKeyToLoad = child.key;
+        return false;
+      });
+      // return data set and the current last key loaded
+      return { data, lastKeyToLoad };
+    })
+  );
+}
+
+function waitForLoaded(
+  query: database.Query,
+  snap$: Observable<SnapshotPrevKey[]>
+) {
+  const loaded$ = loadedData(query);
+  return loaded$.pipe(
+    withLatestFrom(snap$),
+    // Get the latest values from the "loaded" and "child" datasets
+    // We can use both datasets to form an array of the latest values.
+    map(([loaded, changes]) => {
+      // Store the last key in the data set
+      let lastKeyToLoad = loaded.lastKeyToLoad;
+      // Store all child keys loaded at this point
+      const loadedKeys = changes.map(change => change.snapshot.key);
+      return { changes, lastKeyToLoad, loadedKeys };
+    }),
+    // This is the magical part, only emit when the last load key
+    // in the dataset has been loaded by a child event. At this point
+    // we can assume the dataset is "whole".
+    skipWhile(meta => meta.loadedKeys.indexOf(meta.lastKeyToLoad) === -1),
+    // Pluck off the meta data because the user only cares
+    // to iterate through the snapshots
+    map(meta => meta.changes)
+  );
+}

packages/rxfire/database/list/index.ts

@@ -0,0 +1,116 @@
+/**
+ * Copyright 2018 Google 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 { database } from 'firebase';
+import { ChildEvent, SnapshotPrevKey } from '../interfaces';
+import { Observable, of, merge } from 'rxjs';
+import { validateEventsArray, isNil } from '../utils';
+import { fromRef } from '../fromRef';
+import { switchMap, scan, distinctUntilChanged } from 'rxjs/operators';
+
+export function stateChanges(query: database.Query, events?: ChildEvent[]) {
+  events = validateEventsArray(events);
+  const childEvent$ = events.map(event => fromRef(query, event));
+  return merge(...childEvent$);
+}
+
+export function list(
+  query: database.Query,
+  events?: ChildEvent[]
+): Observable<SnapshotPrevKey[]> {
+  events = validateEventsArray(events);
+  return fromRef(query, 'value', 'once').pipe(
+    switchMap(change => {
+      const childEvent$ = [of(change)];
+      events.forEach(event => childEvent$.push(fromRef(query, event)));
+      return merge(...childEvent$).pipe(scan(buildView, []));
+    }),
+    distinctUntilChanged()
+  );
+}
+
+function positionFor<T>(changes: SnapshotPrevKey[], key) {
+  const len = changes.length;
+  for (let i = 0; i < len; i++) {
+    if (changes[i].snapshot.key === key) {
+      return i;
+    }
+  }
+  return -1;
+}
+
+function positionAfter<T>(changes: SnapshotPrevKey[], prevKey?: string) {
+  if (isNil(prevKey)) {
+    return 0;
+  } else {
+    const i = positionFor(changes, prevKey);
+    if (i === -1) {
+      return changes.length;
+    } else {
+      return i + 1;
+    }
+  }
+}
+
+function buildView(current: SnapshotPrevKey[], change: SnapshotPrevKey) {
+  const { snapshot, prevKey, event } = change;
+  const { key } = snapshot;
+  const currentKeyPosition = positionFor(current, key);
+  const afterPreviousKeyPosition = positionAfter(current, prevKey);
+  switch (event) {
+    case 'value':
+      if (change.snapshot && change.snapshot.exists()) {
+        let prevKey = null;
+        change.snapshot.forEach(snapshot => {
+          const action: SnapshotPrevKey = { snapshot, event: 'value', prevKey };
+          prevKey = snapshot.key;
+          current = [...current, action];
+          return false;
+        });
+      }
+      return current;
+    case 'child_added':
+      if (currentKeyPosition > -1) {
+        // check that the previouskey is what we expect, else reorder
+        const previous = current[currentKeyPosition - 1];
+        if (((previous && previous.snapshot.key) || null) != prevKey) {
+          current = current.filter(x => x.snapshot.key !== snapshot.key);
+          current.splice(afterPreviousKeyPosition, 0, change);
+        }
+      } else if (prevKey == null) {
+        return [change, ...current];
+      } else {
+        current = current.slice();
+        current.splice(afterPreviousKeyPosition, 0, change);
+      }
+      return current;
+    case 'child_removed':
+      return current.filter(x => x.snapshot.key !== snapshot.key);
+    case 'child_changed':
+      return current.map(x => (x.snapshot.key === key ? change : x));
+    case 'child_moved':
+      if (currentKeyPosition > -1) {
+        const data = current.splice(currentKeyPosition, 1)[0];
+        current = current.slice();
+        current.splice(afterPreviousKeyPosition, 0, data);
+        return current;
+      }
+      return current;
+    // default will also remove null results
+    default:
+      return current;
+  }
+}

packages/rxfire/database/object/index.ts

@@ -0,0 +1,28 @@
+/**
+ * Copyright 2018 Google 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 { database } from 'firebase';
+import { SnapshotPrevKey } from '../interfaces';
+import { fromRef } from '../fromRef';
+import { Observable } from 'rxjs';
+
+/**
+ * Get the snapshot changes of an object
+ * @param query
+ */
+export function object(query: database.Query): Observable<SnapshotPrevKey> {
+  return fromRef(query, 'value');
+}

packages/rxfire/database/package.json

@@ -0,0 +1,5 @@
+{
+  "name": "rxfire/database",
+  "main": "dist/index.cjs.js",
+  "module": "dist/index.esm.js"
+}

packages/rxfire/database/utils.ts

@@ -0,0 +1,31 @@
+/**
+ * Copyright 2018 Google 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.
+ */
+
+export function isNil(obj: any): boolean {
+  return obj === undefined || obj === null;
+}
+
+/**
+ * Check the length of the provided array. If it is empty return an array
+ * that is populated with all the Realtime Database child events.
+ * @param events
+ */
+export function validateEventsArray(events?: any[]) {
+  if (isNil(events) || events!.length === 0) {
+    events = ['child_added', 'child_removed', 'child_changed', 'child_moved'];
+  }
+  return events;
+}