Wrapping IndexDB with RxJs

Question

I find IndexedDB cumbersome and want to wrap the functionality with rxjs observables. I've come up with an angular2 service that seems to fit the bill, but I'm wondering if I'm using the observables correctly or if there might be any subscriptions I'm missing (plnkr):

Usage:

this.localDbService.set('storeName', key, value)
  .subscribe(
    x => console.log('stored key ', x),
    err => console.error('error storing value', err)
  );

this.localDbService.get('storeName', key)
  .subscribe(
    x => console.log('got value', x),
    err => console.error('error getting value', err)
  );

The service opens the database in the constructor and provides it using a ReplaySubject. This is the first thing I think might have a better solution. I want some sort of observable that will provide me with a single value but only after initialization has finished, and provide the same value any time it is called. I use a ReplaySubject(1) and use .take(1) on it, so any subscribers will not get a value until one is provided, and will only get a single value and won't have to unsubscribe. Should I perhaps use an AsyncSubject?

Also I'm catching errors in storing and querying and passing any errors from db creation up to the caller, am I missing anything here?:

store(storeName: string, key: any, value: any): Observable<any> {
  return Observable.create((observer: Observer<any>) => {
    try {
      this.db.take(1).subscribe(db => {
        if (!db) {
          observer.error("IndexDB not supported!");
          return;
        }

        var txn = db.transaction([storeName], "readwrite");
        var store = txn.objectStore(storeName);
        var req = store.add(value, key);
        req.onerror = function(err) {
          observer.error(err);
          return;
        };
        req.onsuccess = function(e: any) {
          observer.next(e.target.result);
          observer.complete();
        };
      });
    } catch(err) {
      observer.error(err);
    }
  });
}

Answer 1

updated version of the code, based on @MarcoTerzolo 's answer

import { Injectable } from '@angular/core';

import { Observable, Observer, ReplaySubject, Subject } from 'rxjs';
import { take } from 'rxjs/operators';

const VERSION = 2;

// good tutorial: https://code.tutsplus.com/tutorials/working-with-indexeddb--net-34673
// reference: https://developer.mozilla.org/en-US/docs/Web/API/IndexedDB_API/Using_IndexedDB

@Injectable()
export class LocalDbService {
  db: Subject<IDBDatabase> = new ReplaySubject<IDBDatabase>(1);

  constructor() {
    window['SLOCALDB'] = this;

    if (!window.indexedDB) {
      this.db.next(undefined);
      this.db.complete();
    } else {
      console.log(`localdb - requesting open of 'swbuddy' version ${VERSION}`);

      const openRequest = indexedDB.open('swbuddy', VERSION);
      openRequest.onerror = err => {
        console.error('localdb - open has error:', err);

        this.db.error(err);
        this.db.complete();
      };

      openRequest.onupgradeneeded = function(e: any) {
        console.log('localdb - upgrade needed!');

        // create object stores for 'rawLogin' and 'rawVisit'
        const db: IDBDatabase = e.target.result;
        if (!db.objectStoreNames.contains('rawLogin')) {
          db.createObjectStore('rawLogin');
        }

        if (!db.objectStoreNames.contains('rawVisit')) {
          db.createObjectStore('rawVisit');
        }

        // for testing any type of object setup
        if (!db.objectStoreNames.contains('test')) {
          db.createObjectStore('test');
        }
      };

      openRequest.onsuccess = (e: any) => {
        console.log('localdb - open success!', e.target.result);

        const db: IDBDatabase = e.target.result;
        this.db.next(db);
      };
    }
  }

  get(storeName: string, key: any): Observable<any> {
    console.log('localdb.query()');
    return Observable.create((observer: Observer<any>) => {
      try {
        console.log('localdb.query() - subscribed!');
        this.db.pipe(
          take(1)
        ).subscribe(db => {
          console.log('localdb.query() got db:', db);
          if (!db) {
            observer.error('IndexDB not supported!');
            return;
          }

          const txn = db.transaction([storeName], 'readonly');
          const store = txn.objectStore(storeName);
          const req = store.get(key);
          req.onerror = function(e: any) {
            observer.error(e.target.error);
            return;
          };
          req.onsuccess = function(e: any) {
            observer.next(e.target.result);
            observer.complete();
          };
        });
      } catch (err) {
        observer.error(err);
      }
    });
  }

  put(storeName: string, key: any, value: any): Observable<any> {
    return Observable.create((observer: Observer<any>) => {
      try {
        this.db.pipe(
          take(1)
        ).subscribe(db => {
          if (!db) {
            observer.error('IndexDB not supported!');
            return;
          }

          const txn = db.transaction([storeName], 'readwrite');
          const store = txn.objectStore(storeName);
          const req = store.put(value, key);
          req.onerror = function(e: any) {
            console.log('store error event:', e);
            observer.error(e.target.error);
            return;
          };
          req.onsuccess = function(e: any) {
            console.log('store success:', e);
            observer.next(e.target.result);
            observer.complete();
          };
        });
      } catch (err) {
        observer.error(err);
      }
    });
  }

  delete(storeName: string, key: any): Observable<any> {
    console.log('localdb.delete()');
    return Observable.create((observer: Observer<any>) => {
      try {
        console.log('localdb.delete() - subscribed!');
        this.db.pipe(
          take(1)
        ).subscribe(db => {
          console.log('localdb.delete() got db:', db);
          if (!db) {
            observer.error('IndexDB not supported!');
            return;
          }

          const txn = db.transaction([storeName], 'readwrite');
          const store = txn.objectStore(storeName);
          const req = store.delete(key);
          req.onerror = function(e: any) {
            observer.error(e.target.error);
            return;
          };
          req.onsuccess = function(e: any) {
            observer.next(e.target.result);
            observer.complete();
          };
        });
      } catch (err) {
        observer.error(err);
      }
    });
  }


  query(storeName: string): Observable<any> {
    console.log('localdb.query()');
    return Observable.create((observer: Observer<any>) => {
      try {
        console.log('localdb.query() - subscribed!');
        this.db.pipe(
          take(1)
        ).subscribe(db => {
          console.log('localdb.query() got db:', db);
          if (!db) {
            observer.error('IndexDB not supported!');
            return;
          }

          const txn = db.transaction([storeName], 'readonly');
          const store = txn.objectStore(storeName);
          const req = store.openCursor();
          req.onerror = function(e: any) {
            observer.error(e.target.error);
            return;
          };
          req.onsuccess = function(e: any) {
            const cursor = e.target.result;
            if (cursor) {
              observer.next({key: cursor.key, value: cursor.value});
              cursor.continue();
            } else {
              observer.complete();
            }
          };
        });
      } catch (err) {
        observer.error(err);
      }
    });
  }
}

Answer 2

Two years have passed but your wrap is really a good start for wrapping indexedDb with rxjs. Take(1) now lives in rxjs/operators import and must be used with a pipe but besides this needed an upgrade.

Your approach with rxjs seems like the right choice when handling indexedDb without external libraries. Thumbs up!

Answer 3

Very nice.

I've used your code as a foundation for IndexDB (RxJs) - Angular Service.

Some things I noticed:

1. The try/catch won't intercept errors inside the subscription block (which is async). By adding try/catch block inside the subscription I was able to catch errors due to schema changes for get/put operations.
2. No need to accept e: any in the callback functions. Inside req.onerror you can read the req.error value. Inside req.onsuccess you can read the req.result value. These values are also typed.