docs: function(topic, msg, cb0) {
              var cb1 = function(err, data) {
                  if (err) {
                      cb0(err);
                  } else {
                      /* We do not propagate the app error or exception
                       * in the callback because otherwise we would unsubscribe
                       * with, e.g., authorization failures, creating a denial of
                       * service.*/
                      if (data) {
                          if (json_rpc.isSystemError(data) ||
                              json_rpc.getAppReplyError(data)) {
                              var logMsg = 'Ignoring Error in pubsub handler ' +
                                      JSON.stringify(data);
                              $.ca.$.log && $.ca.$.log.debug(logMsg);
                          } else {
                              logMsg = 'Ignoring pubsub value' +
                                  JSON.stringify(data);
                              $.ca.$.log && $.ca.$.log.trace(logMsg);
                          }
                      }
                      cb0(null);
                  }
              };
  
              try {
                  var methodName = subscribed[topic];
                  if (that.__ca_isShutdown__) {
                      var err = new Error('pubsub: delivering to shutdown plug');
                      err['topic'] = topic;
                      err['msg'] = msg;
                      cb0(err);
                  } else if (!methodName) {
                      err = new Error('pubsub: Ignoring msg from unsubscribed ' +
                                      'channel');
                      err['topic'] = topic;
                      err['msg'] = msg;
                      // force unsubscribing to keep consistent
                      cb0(err);
                  } else {
                      var notif = JSON.parse(msg);
                      var all = [
                          json_rpc.getToken(notif), // token
                          $.ca.__ca_getName__(),  // to
                          json_rpc.getFrom(notif), // from
                          json_rpc.getSessionId(notif), //session
                          methodName // method
                      ].concat(json_rpc.getMethodArgs(notif)); // args
  
                      var req = json_rpc.request.apply(json_rpc.request, all);
                      $.ca.__ca_process__(req, cb1);
                  }
              } catch (ex) {
                  cb0(ex);
              }
          }