sends sent changes on reconnect

After auth and index and/or receive changes from server emits a 'ready' event for the channel.
After authing, listens for one ready event and resends any changes in the local queue

Author: beaucollins

src/simperium/channel.js

@@ -197,6 +197,7 @@ internal.indexingComplete = function() {
 	this.index_last_id = null;
 	this.index_cv = null;
 	this.bucket.removeListener( 'update', this.bucketUpdateListener );
+	this.emit( 'ready' )
 }
 
 export default function Channel( appid, access_token, bucket, store ) {
@@ -323,11 +324,11 @@ Channel.prototype.onAuth = function( data ) {
 		this.emit( 'unauthorized', auth );
 		return;
 	} catch ( error ) {
-		// Clear any unacknowledged changes on reconnect
-		this.localQueue.sent = {};
-
 		// request cv and then send method
-		init = function( cv ) {
+		this.once( 'ready', () => {
+			this.localQueue.resendSentChanges();
+		} )
+		init = ( cv ) => {
 			if ( cv ) {
 				this.localQueue.start();
 				this.sendChangeVersionRequest( cv );
@@ -338,7 +339,7 @@ Channel.prototype.onAuth = function( data ) {
 			}
 		};
 
-		this.store.getChangeVersion().then( init.bind( this ) );
+		this.store.getChangeVersion().then( init );
 
 		return;
 	}
@@ -400,6 +401,8 @@ Channel.prototype.onChanges = function( data ) {
 	changes.forEach( function( change ) {
 		onChange( change.id, change );
 	} );
+	// emit ready after all server changes have been applied
+	this.emit( 'ready' );
 }
 ;
 
@@ -590,6 +593,12 @@ LocalQueue.prototype.compressAndSend = function( id, ghost ) {
 	this.emit( 'send', change );
 }
 
+LocalQueue.prototype.resendSentChanges = function() {
+	for ( let ccid in this.sent ) {
+		this.emit( 'send', this.sent[ccid] )
+	}
+}
+
 function collectionRevisions( channel, id, callback ) {
 	var expectedVersions = -1;
 	var onGhostRetrieved = function( ghost ) {

test/simperium/channel_test.js

@@ -2,7 +2,7 @@
 import Channel from '../../src/simperium/channel'
 import util from 'util'
 import { parseMessage } from '../../src/simperium/util'
-import assert, { equal } from 'assert'
+import assert, { equal, ok } from 'assert'
 import * as fn from './fn'
 import jsondiff from '../../src/simperium/jsondiff'
 import defaultGhostStoreProvider from '../../src/simperium/ghost/default'
@@ -12,6 +12,11 @@ import mockBucketStore from './mock_bucket_store'
 
 const differ = jsondiff()
 const diff = differ.object_diff.bind( differ )
+const cycle = ( ... fns ) => ( ... args ) => {
+	const [ head, ... rest ] = fns
+	head( ... args )
+	fns = rest.concat( head )
+}
 
 describe( 'Channel', function() {
 	var channel, bucket, store;
@@ -158,6 +163,21 @@ describe( 'Channel', function() {
 			bucket.update( 'mock-id', data );
 		} );
 
+		it( 'should resend sent but unacknowledged changes on reconnect', () => new Promise( resolve => {
+			channel.localQueue.sent['fake-ccid'] = { fake: 'change' }
+
+			channel.on( 'send', cycle(
+				() => channel.handleMessage( 'i:{"index":[],"current":"cv"}'),
+				m => {
+					resolve()
+				}
+			) )
+
+			channel.handleMessage( 'auth:user@example.com' )
+			
+			channel.emit( 'ready' )
+		} ) )
+
 		it( 'should send remove operation', function( done ) {
 			channel.on( 'send', function( msg ) {
 				var message = parseMessage( msg ),
@@ -221,6 +241,11 @@ describe( 'Channel', function() {
 			channel.handleMessage( 'c:' + JSON.stringify( [change] ) );
 		} );
 
+		it( 'should emit ready after receiving changes', ( done ) => {
+			channel.on( 'ready', () => done() )
+			channel.handleMessage( 'c:[]' );
+		} )
+
 		it( 'should notify bucket after network deletion', function( done ) {
 			var key = 'deleteTest';
 
@@ -366,12 +391,10 @@ describe( 'Channel', function() {
 	} );
 
 	describe( 'after authorizing', function() {
-		beforeEach( function( next ) {
-			channel.once( 'send', function() {
-				next();
-			} );
+		beforeEach( () => new Promise( resolve => {
+			channel.once( 'send', () => resolve() );
 			channel.onConnect();
-		} );
+		} ) );
 
 		it( 'should request index', function( done ) {
 			channel.once( 'send', function( data ) {
@@ -402,17 +425,20 @@ describe( 'Channel', function() {
 			} );
 		} );
 
-		it( 'should emit index event when index complete', function( done ) {
+		it( 'should emit index and ready event when index complete', () => new Promise( resolve => {
 			var page = 'i:{"index":[{"id":"objectid","v":1,"d":{"title":"Hello World"}}],"current":"cv"}';
+			let indexed = false
 			channel.on( 'index', function( cv ) {
-				setImmediate( function() {
-					assert.equal( 'cv', cv );
-					assert( !bucket.isIndexing );
-					done();
-				} )
+				assert.equal( 'cv', cv );
+				assert( !bucket.isIndexing );
+				indexed = true
 			} );
+			channel.on( 'ready', () => {
+				ok( indexed )
+				resolve()
+			} )
 			channel.handleMessage( page );
-		} );
+		} ) );
 
 		it( 'should request next index page', function( done ) {
 			var page = 'i:{"index":[{"id":"objectid","v":1,"d":{"title":"Hello World"}}],"mark":"next-mark","current":"cv"}';