Process nosub messages on the client.

The visible effect of this is that attempts to subscribe to non-existent
subscriptions will be noticed by the client and they will not attempt to re-sub
on reconnect.

packages/livedata/livedata_connection.js

@@ -467,7 +467,6 @@ _.extend(Meteor._LivedataConnection.prototype, {
       if (!_.has(self._subscriptions, id))
         return;
       self._send({msg: 'unsub', id: id});
-      // We assume that all unsubs are successful.
       delete self._subscriptions[id];
     }};
 
@@ -1129,7 +1128,11 @@ _.extend(Meteor._LivedataConnection.prototype, {
 
   _livedata_nosub: function (msg) {
     var self = this;
-    // Meteor._debug("NOSUB", msg);
+    // we weren't subbed anyway, or we initiated the unsub.
+    if (!_.has(self._subscriptions, msg.id))
+      return;
+    delete self._subscriptions[msg.id];
+    // XXX call on-error callback
   },
 
   _livedata_result: function (msg) {

packages/livedata/livedata_connection_tests.js

@@ -1389,6 +1389,38 @@ Tinytest.add("livedata stub - reconnect double wait method", function (test) {
                  {msg: 'method', method: 'lastMethod', params: [], id: '*'});
 });
 
+Tinytest.add("livedata stub - subscribe failure", function (test) {
+  var stream = new Meteor._StubStream();
+  var conn = newConnection(stream);
+
+  startAndConnect(test, stream);
+
+  // subscribe
+  var onReadyFired = false;
+  var sub = conn.subscribe('unknownSub', function () {
+    onReadyFired = true;
+  });
+  test.isFalse(onReadyFired);
+
+  var subMessage = JSON.parse(stream.sent.shift());
+  test.equal(subMessage, {msg: 'sub', name: 'unknownSub', params: [],
+                          id: subMessage.id});
+
+  // Reject the sub.
+  stream.receive({msg: 'nosub', id: subMessage.id,
+                  error: {error: 404, reason: "Subscription not found"}});
+  test.isFalse(onReadyFired);
+
+  // stream reset: reconnect!
+  stream.reset();
+  // We send a connect.
+  testGotMessage(test, stream, makeConnectMessage(SESSION_ID));
+  // We should NOT re-sub to the sub, because we processed the error.
+  test.length(stream.sent, 0);
+  test.isFalse(onReadyFired);
+});
+
+
 // XXX also test:
 // - reconnect, with session resume.
 // - restart on update flag