Finished Live Query with Observable

CHANGELOG.md

@@ -1,3 +1,9 @@
+####3.0.0
+
+### New Feature
+
+* Live Query With Observable [#236](https://github.com/orientechnologies/orientjs/issues/236)
+
 ####2.2.4
 
 ###Bug Fixes

lib/session/session.js

@@ -135,11 +135,11 @@ class OSession extends DB {
             var iToken = response[0].token;
             var wrapperCallback = function (currentToken, operation, result) {
               if (currentToken == iToken) {
-                subscriber.next({token : currentToken , operation : operation , data : result});
+                subscriber.next({token: currentToken, operation: operation, data: result});
               }
             };
 
-            var unsubscribeCallbak = function(currentToken){
+            var unsubscribeCallbak = function (currentToken) {
               if (currentToken == iToken) {
                 subscriber.complete();
               }
@@ -150,7 +150,9 @@ class OSession extends DB {
             this.srv.on("live-query-end", unsubscribeCallbak);
 
           }
-        });
+        }).catch((error) => {
+        subscriber.error(error);
+      });
     });
 
     return result;

test/session/session-live-query-test.js

@@ -1,4 +1,5 @@
 var Promise = require('bluebird');
+var Errors = require('../../lib/errors');
 describe("Session API - Live Query ", function () {
 
   this.timeout(20000);
@@ -63,18 +64,18 @@ describe("Session API - Live Query ", function () {
     var session = this.session;
     this.session.liveQuery("LIVE SELECT FROM Test")
       .subscribe(function (live) {
-      count++;
-      live.data.content.name.should.eql('a');
-      if (count === TOTAL) {
-        session.query("live unsubscribe " + live.token).then(function (res) {
+        count++;
+        live.data.content.name.should.eql('a');
+        if (count === TOTAL) {
+          session.query("live unsubscribe " + live.token).then(function (res) {
 
-        });
-      }
-    }, function (err) {
+          });
+        }
+      }, function (err) {
 
-    }, function () {
+      }, function () {
         done();
-    });
+      });
     var self = this;
     setTimeout(function () {
       var promises = [];
@@ -93,4 +94,17 @@ describe("Session API - Live Query ", function () {
 
   });
 
+  it('should trigger live query error', function (done) {
+
+    this.session.liveQuery("LIVE SELECT FROM Test2")
+      .subscribe(function (live) {
+        throw new Error('Should never happen!');
+      }, function (err) {
+        err.should.be.an.instanceOf(Errors.RequestError);
+        done();
+      }, function () {
+        throw new Error('Should never happen!');
+      });
+  });
+
 });