lib/api.js
function(token,options){
      options = options||{};
      options.token = token;
      var obj = {
        type:"rest"
        ,args:{ 
          url: '/v1/sync',
          data: options,
          method: 'get' 
        }
      };

      var s = through();

      repipe(s,function(err,last,done){
        o.log('repipe got error? ',err,' should i repipe?');
        if(err && err.code != 'E_MDM') return done(err);

        getConnection(function(err,con){
          if(err) return done(err);
          var dbstream = con.mdm.createReadStream(obj);

          var noendstream = dbstream.pipe(through(function(data){
            this.queue(data);
          },function(){
            if(options.tail || options.tail == undefined) {
              // repipe resumes the stream on error events!

              var error = new Error("never end bro!");
              error.code = "E_MDM";

              this.emit("error",error);
            } else {
              this.queue(null);
            }
          }));

          dbstream.on('error',function(err){
            noendstream.emit("error",err)
          });

          done(false,noendstream);
        });
      });

      s.on('data',function(d){
        // on reconnect start sync from where i left off.
        obj.args.data.start = d.time;
      })
      return s;
    }
function(token,options){
      options = options||{};
      options.token = token;
      var obj = {
        type:"rest"
        ,args:{ 
          url: '/v1/sync',
          data: options,
          method: 'get' 
        }
      };

      var s = through();

      repipe(s,function(err,last,done){
        o.log('repipe got error? ',err,' should i repipe?');
        if(err && err.code != 'E_MDM') return done(err);

        getConnection(function(err,con){
          if(err) return done(err);
          var dbstream = con.mdm.createReadStream(obj);

          var noendstream = dbstream.pipe(through(function(data){
            this.queue(data);
          },function(){
            if(options.tail || options.tail == undefined) {
              // repipe resumes the stream on error events!

              var error = new Error("never end bro!");
              error.code = "E_MDM";

              this.emit("error",error);
            } else {
              this.queue(null);
            }
          }));

          dbstream.on('error',function(err){
            noendstream.emit("error",err)
          });

          done(false,noendstream);
        });
      });

      s.on('data',function(d){
        // on reconnect start sync from where i left off.
        obj.args.data.start = d.time;
      })
      return s;
    }
function ( token , options ) { options = options || { } ; options . token = token ; var obj = { type : "rest" , args : { url : '/v1/sync' , data : options , method : 'get' } } ; var s = through ( ) ; repipe ( s , function ( err , last , done ) { o . log ( 'repipe got error? ' , err , ' should i repipe?' ) ; if ( err && err . code != 'E_MDM' ) return done ( err ) ; getConnection ( function ( err , con ) { if ( err ) return done ( err ) ; var dbstream = con . mdm . createReadStream ( obj ) ; var noendstream = dbstream . pipe ( through ( function ( data ) { this . queue ( data ) ; } , function ( ) { if ( options . tail || options . tail == undefined ) { // repipe resumes the stream on error events! var error = new Error ( "never end bro!" ) ; error . code = "E_MDM" ; this . emit ( "error" , error ) ; } else { this . queue ( null ) ; } } ) ) ; dbstream . on ( 'error' , function ( err ) { noendstream . emit ( "error" , err ) } ) ; done ( false , noendstream ) ; } ) ; } ) ; s . on ( 'data' , function ( d ) { // on reconnect start sync from where i left off. obj . args . data . start = d . time ; } ) return s ; }
sync the account's data in realtime
sync the account s data in realtime