Merge pull request #21 from mikemintz/mikemintz/atomic-changefeeds

Use atomic changefeeds

README.md

@@ -64,8 +64,8 @@ Most of the query validation logic can be found in
 ## How do I use this?
 
 Check out the [`examples/`] folder in this repository for fully-working React
-applications. You will need to [install RethinkDB] first if you haven't
-already.
+applications. You will need to [install RethinkDB] version 2.2 or newer first
+if you haven't already.
 
 The [`examples/tutorial/`] project has in-depth instructions explaining how to
 create a simple app from scratch.
@@ -164,6 +164,37 @@ IE9 and old android browsers are not supported because [they don't have
 WebSockets](http://caniuse.com/#feat=websockets), and
 [rethinkdb-websocket-client] currently requires WebSocket support.
 
+## Upgrade guide
+
+Most new versions of react-rethinkdb are backwards compatible with previous
+versions. Below are exceptions with breaking changes:
+
+### Upgrading to 0.5 (from 0.4)
+
+Version 0.5 of react-rethinkdb [saw the introduction of atomic
+changefeeds](https://github.com/mikemintz/react-rethinkdb/issues/20), which is
+a new feature in [RethinkDB
+2.2](https://github.com/rethinkdb/rethinkdb/blob/5b7b03f017d7e4f560aa3cc3f2c286fefeae3dae/NOTES.md).
+This simplifies the logic by sending one "atomic" changefeed query, rather than
+a static query for initial results followed by a changefeed query for realtime
+updates. This saves bandwidth and prevents the race condition where data
+changes in between the two queries.
+
+Regular static queries will continue to work the same. But in order to use
+react-rethinkdb 0.5 with changefeed queries, you must both:
+* Upgrade to RethinkDB 2.2 in your backend.
+* Add the `include_initial=true` option to all changefeed queries in your
+  rethinkdb-websocket-server query whitelist. Below is an example:
+
+```js
+RQ(
+  RQ.CHANGES(
+    RQ.TABLE("tortoises")
+  ).opt("include_states", true)
+   .opt("include_initial", true) // this line is now required
+).opt("db", RQ.DB("test")),
+```
+
 ## Roadmap
 
 * Investigate performance. I haven't tested with large queries, large numbers

examples/chat/server/queries.js

@@ -39,7 +39,7 @@ export const queryWhitelist = [
           )
         )
       )
-    ).opt("include_states", true)
+    ).opt("include_states", true).opt("include_initial", true)
   ).opt("db", RQ.DB(cfg.dbName)),
 
   // Insert new message

examples/chess/server/queries.js

@@ -89,7 +89,7 @@ export const queryWhitelist = [
           }
         )
       )
-    ).opt("include_states", true)
+    ).opt("include_states", true).opt("include_initial", true)
   ).opt("db", RQ.DB(cfg.dbName)),
 
   // List moves for a game
@@ -107,7 +107,7 @@ export const queryWhitelist = [
         RQ.TABLE("moves"),
         {"gameId": x => typeof x === 'string'}
       )
-    ).opt("include_states", true)
+    ).opt("include_states", true).opt("include_initial", true)
   ).opt("db", RQ.DB(cfg.dbName)),
 
   // Create new game

examples/tutorial/README.md

@@ -67,6 +67,8 @@ To install RethinkDB, run the following command in your terminal:
 | Arch Linux | `sudo pacman -S rethinkdb`         |
 | Other      | http://rethinkdb.com/docs/install/ |
 
+Make sure you are using version 2.2 or newer.
+
 
 ## Set up npm package
 

package.json

@@ -17,7 +17,7 @@
     "dist"
   ],
   "dependencies": {
-    "rethinkdb-websocket-client": "^0.4.3"
+    "rethinkdb-websocket-client": "^0.4.5"
   },
   "devDependencies": {
     "babel": "^5.5.6",

src/QueryState.js

@@ -30,22 +30,13 @@ export class QueryState {
   }
 
   handleConnect() {
-    // TODO It would be great not to have to run two separate queries, one with
-    // changefeed and one without. Not only is it inefficient, but it could be
-    // inaccurate if changes happen in the time between when the two queries
-    // are sent. Unfortunately, many changefeed queries do not provide initial
-    // values, so the static query is necessary to support those.
-    //
-    // RethinkDB 2.2 should support initial values for all change feeds. Let's
-    // remove the redundant static query when queryRequest.changes=true once
-    // this RethinkDB feature is implemented.
-    // See https://github.com/rethinkdb/rethinkdb/issues/3579
     if (this.loading || this.queryRequest.changes) {
       this.loading = true;
       this.errors = [];
-      this._runStaticQuery(this.queryRequest.query, this.runQuery);
       if (this.queryRequest.changes) {
         this._runChangeQuery(this.queryRequest.query, this.runQuery);
+      } else {
+        this._runStaticQuery(this.queryRequest.query, this.runQuery);
       }
     }
   }
@@ -80,24 +71,30 @@ export class QueryState {
   }
 
   _runChangeQuery(query, runQuery) {
-    const changeQuery = query.changes({includeStates: true});
+    const changeQuery = query.changes({includeStates: true, includeInitial: true});
     const promise = runQuery(changeQuery);
     this.closeHandlers.push(() => promise.then(x => isCursor(x) && x.close()));
     promise.then(cursor => {
-      let feedStateReady = false;
+      const isPointFeed = cursor.constructor.name === 'AtomFeed';
+      this.value = isPointFeed ? undefined : [];
       cursor.each((error, row) => {
         if (error) {
           this._addError(error);
         } else {
           if (row.state) {
-            feedStateReady = row.state === 'ready';
-          } else if (feedStateReady) {
-            // We ignore initial values, since we use _runStaticQuery for that
+            if (row.state === 'ready') {
+              this.loading = false;
+              this._updateSubscriptions();
+            }
+          } else {
             this._applyChangeDelta(row.old_val, row.new_val);
           }
         }
       });
     }, error => {
+      if (error.msg === 'Unrecognized optional argument `include_initial`.') {
+        console.error('react-rethinkdb requires rethinkdb >= 2.2 on backend');
+      }
       this._addError(error);
     });
   }
@@ -134,10 +131,6 @@ export class QueryState {
   }
 
   _applyChangeDelta(oldVal, newVal) {
-    if (this.loading) {
-      // TODO Remove this after RethinkDB #3579 is implemented.
-      throw new Error('Received changefeed update before static query result');
-    }
     if (Array.isArray(this.value)) {
       // TODO Make more efficient, with O(1) hashtables with cached
       // JSON.stringify keys. But this may not be necessary after RethinkDB
@@ -163,6 +156,8 @@ export class QueryState {
     } else {
       this.value = newVal;
     }
-    this._updateSubscriptions();
+    if (!this.loading) {
+      this._updateSubscriptions();
+    }
   }
 }