dojo.provide("dojox.cometd._base"); dojo.require("dojo.AdapterRegistry"); /* * this file defines Comet protocol client. Actual message transport is * deferred to one of several connection type implementations. The default is a * long-polling implementation. A single global object named "dojox.cometd" is * used to mediate for these connection types in order to provide a stable * interface. * * extensions modules may be loaded (eg "dojox.cometd.timestamp", that use * the cometd._extendInList and cometd._extendOutList fields to provide functions * that extend and handling incoming and outgoing messages. * * By default the long-polling and callback-polling transports will be required. * If specific or alternative transports are required, then they can be directly * loaded. For example dojo.require('dojox.cometd.longPollTransportJsonEncoded') * will load cometd with only the json encoded variant of the long polling transport. */ dojox.cometd = { Connection: function(prefix){ // This constructor is stored as dojox.cometd.Connection // summary // This constructor is used to create new cometd connections. Generally, you should use // one cometd connection for each server you connect to. A default connection instance is // created at dojox.cometd. // To connect to a new server you can create an instance like: // var cometd = new dojox.cometd.Connection("/otherServer"); // cometd.init("http://otherServer.com/cometd"); // // prefix is the prefix for all the events that are published in the Dojo pub/sub system. // You must include this prefix, and it should start with a slash like "/myprefix". // cometd states: // unconnected, handshaking, connecting, connected, disconnected dojo.mixin(this, { prefix: prefix, _status: "unconnected", _handshook: false, _initialized: false, _polling: false, expectedNetworkDelay: 10000, // expected max network delay connectTimeout: 0, // If set, used as ms to wait for a connect response and sent as the advised timeout version: "1.0", minimumVersion: "0.9", clientId: null, messageId: 0, batch: 0, _isXD: false, handshakeReturn: null, currentTransport: null, url: null, lastMessage: null, _messageQ: [], handleAs: "json", _advice: {}, _backoffInterval: 0, _backoffIncrement: 1000, _backoffMax: 60000, _deferredSubscribes: {}, _deferredUnsubscribes: {}, _subscriptions: [], _extendInList: [], // List of functions invoked before delivering messages _extendOutList: [] // List of functions invoked before sending messages }); this.state = function() { return this._status; } this.init = function( /*String*/ root, /*Object?*/ props, /*Object?*/ bargs){ // return: dojo.Deferred // summary: // Initialize the cometd implementation of the Bayeux protocol // description: // Initialize the cometd implementation of the Bayeux protocol by // sending a handshake message. The cometd state will be changed to CONNECTING // until a handshake response is received and the first successful connect message // has returned. // The protocol state changes may be monitored // by subscribing to the dojo topic "/prefix/meta" (typically "/cometd/meta") where // events are published in the form // {cometd:this,action:"handshake",successful:true,state:this.state()} // root: // The URL of the cometd server. If the root is absolute, the host // is examined to determine if xd transport is needed. Otherwise the // same domain is assumed. // props: // An optional object that is used as the basis of the handshake message // bargs: // An optional object of bind args mixed in with the send of the handshake // example: // | dojox.cometd.init("/cometd"); // | dojox.cometd.init("http://xdHost/cometd",{ext:{user:"fred",pwd:"secret"}}); // FIXME: if the root isn't from the same host, we should automatically // try to select an XD-capable transport props = props || {}; // go ask the short bus server what we can support props.version = this.version; props.minimumVersion = this.minimumVersion; props.channel = "/meta/handshake"; props.id = "" + this.messageId++; this.url = root || dojo.config["cometdRoot"]; if(!this.url){ throw "no cometd root"; return null; } // Are we x-domain? borrowed from dojo.uri.Uri in lieu of fixed host and port properties var regexp = "^(([^:/?#]+):)?(//([^/?#]*))?([^?#]*)(\\?([^#]*))?(#(.*))?$"; var parts = ("" + window.location).match(new RegExp(regexp)); if(parts[4]){ var tmp = parts[4].split(":"); var thisHost = tmp[0]; var thisPort = tmp[1]||"80"; // FIXME: match 443 parts = this.url.match(new RegExp(regexp)); if(parts[4]){ tmp = parts[4].split(":"); var urlHost = tmp[0]; var urlPort = tmp[1]||"80"; this._isXD = ((urlHost != thisHost)||(urlPort != thisPort)); } } if(!this._isXD){ props.supportedConnectionTypes = dojo.map(dojox.cometd.connectionTypes.pairs, "return item[0]"); } props = this._extendOut(props); var bindArgs = { url: this.url, handleAs: this.handleAs, content: { "message": dojo.toJson([props]) }, load: dojo.hitch(this,function(msg){ this._backon(); this._finishInit(msg); }), error: dojo.hitch(this,function(e){ this._backoff(); this._finishInit(e); }), timeout: this.expectedNetworkDelay }; if(bargs){ dojo.mixin(bindArgs, bargs); } this._props = props; for(var tname in this._subscriptions){ for(var sub in this._subscriptions[tname]){ if(this._subscriptions[tname][sub].topic){ dojo.unsubscribe(this._subscriptions[tname][sub].topic); } } } this._messageQ = []; this._subscriptions = []; this._initialized = true; this._status = "handshaking"; this.batch = 0; this.startBatch(); var r; // if xdomain, then we assume jsonp for handshake if(this._isXD){ bindArgs.callbackParamName = "jsonp"; r = dojo.io.script.get(bindArgs); }else{ r = dojo.xhrPost(bindArgs); } return r; } this.publish = function(/*String*/ channel, /*Object*/ data, /*Object?*/ props){ // summary: // publishes the passed message to the cometd server for delivery // on the specified topic // channel: // the destination channel for the message // data: // a JSON object containing the message "payload" // properties: // Optional. Other meta-data to be mixed into the top-level of the // message var message = { data: data, channel: channel }; if(props){ dojo.mixin(message, props); } this._sendMessage(message); } this.subscribe = function( /*String */ channel, /*Object */ objOrFunc, /*String */ funcName, /*Object?*/ props){ // return: dojo.Deferred // summary: // inform the server of this client's interest in channel // description: // `dojox.cometd.subscribe()` handles all the hard work of telling // the server that we want to be notified when events are // published on a particular topic. `subscribe` accepts a function // to handle messages and returns a `dojo.Deferred` object which // has an extra property added to it which makes it suitable for // passing to `dojox.cometd.unsubscribe()` as a "subscription // handle" (much like the handle object that `dojo.connect()` // produces and which `dojo.disconnect()` expects). // // Note that of a subscription is registered before a connection // with the server is established, events sent before the // connection is established will not be delivered to this client. // The deferred object which `subscribe` returns will callback // when the server successfuly acknolwedges receipt of our // "subscribe" request. // channel: // name of the cometd channel to subscribe to // objOrFunc: // an object scope for funcName or the name or reference to a // function to be called when messages are delivered to the // channel // funcName: // the second half of the objOrFunc/funcName pair for identifying // a callback function to notifiy upon channel message delivery // example: // Simple subscribe use-case // | dojox.cometd.init("http://myserver.com:8080/cometd"); // | // log out all incoming messages on /foo/bar // | dojox.cometd.subscribe("/foo/bar", console, "debug"); // example: // Subscribe before connection is initialized // | dojox.cometd.subscribe("/foo/bar", console, "debug"); // | dojox.cometd.init("http://myserver.com:8080/cometd"); // example: // Subscribe an unsubscribe // | dojox.cometd.init("http://myserver.com:8080/cometd"); // | var h = dojox.cometd.subscribe("/foo/bar", console, "debug"); // | dojox.cometd.unsubscribe(h); // example: // Listen for successful subscription: // | dojox.cometd.init("http://myserver.com:8080/cometd"); // | var h = dojox.cometd.subscribe("/foo/bar", console, "debug"); // | h.addCallback(function(){ // | console.debug("subscription to /foo/bar established"); // | }); props = props||{}; if(objOrFunc){ var tname = prefix + channel; var subs = this._subscriptions[tname]; if(!subs || subs.length == 0){ subs = []; props.channel = "/meta/subscribe"; props.subscription = channel; this._sendMessage(props); var _ds = this._deferredSubscribes; if(_ds[channel]){ _ds[channel].cancel(); delete _ds[channel]; } _ds[channel] = new dojo.Deferred(); } for(var i in subs){ if(subs[i].objOrFunc === objOrFunc && (!subs[i].funcName&&!funcName||subs[i].funcName==funcName) ){ return null; } } var topic = dojo.subscribe(tname, objOrFunc, funcName); subs.push({ topic: topic, objOrFunc: objOrFunc, funcName: funcName }); this._subscriptions[tname] = subs; } var ret = this._deferredSubscribes[channel] || {}; ret.args = dojo._toArray(arguments); return ret; // dojo.Deferred } this.unsubscribe = function( /*String*/ channel, /*Object?*/ objOrFunc, /*String?*/ funcName, /*Object?*/ props){ // summary: // inform the server of this client's disinterest in channel // channel: // name of the cometd channel to unsubscribe from // objOrFunc: // an object scope for funcName or the name or reference to a // function to be called when messages are delivered to the // channel. If null then all subscribers to the channel are unsubscribed. // funcName: // the second half of the objOrFunc/funcName pair for identifying // a callback function to notifiy upon channel message delivery if( (arguments.length == 1) && (!dojo.isString(channel)) && (channel.args) ){ // it's a subscription handle, unroll return this.unsubscribe.apply(this, channel.args); } var tname = prefix + channel; var subs = this._subscriptions[tname]; if(!subs || subs.length==0){ return null; } var s=0; for(var i in subs){ var sb = subs[i]; if((!objOrFunc) || ( sb.objOrFunc===objOrFunc && (!sb.funcName && !funcName || sb.funcName==funcName) ) ){ dojo.unsubscribe(subs[i].topic); delete subs[i]; }else{ s++; } } if(s == 0){ props = props || {}; props.channel = "/meta/unsubscribe"; props.subscription = channel; delete this._subscriptions[tname]; this._sendMessage(props); this._deferredUnsubscribes[channel] = new dojo.Deferred(); if(this._deferredSubscribes[channel]){ this._deferredSubscribes[channel].cancel(); delete this._deferredSubscribes[channel]; } } return this._deferredUnsubscribes[channel]; // dojo.Deferred } this.disconnect = function(){ // summary: // Disconnect from the server. // description: // Disconnect from the server by sending a disconnect message // example: // | dojox.cometd.disconnect(); for(var tname in this._subscriptions){ for(var sub in this._subscriptions[tname]){ if(this._subscriptions[tname][sub].topic){ dojo.unsubscribe(this._subscriptions[tname][sub].topic); } } } this._subscriptions = []; this._messageQ = []; if(this._initialized && this.currentTransport){ this._initialized=false; this.currentTransport.disconnect(); } if(!this._polling) { this._publishMeta("connect",false); } this._initialized=false; this._handshook=false; this._status = "disconnected"; //should be disconnecting, but we ignore the reply to this message this._publishMeta("disconnect",true); } // public extension points this.subscribed = function( /*String*/channel, /*Object*/message){ } this.unsubscribed = function(/*String*/channel, /*Object*/message){ } // private methods (TODO name all with leading _) this.tunnelInit = function(childLocation, childDomain){ // placeholder - replaced by _finishInit } this.tunnelCollapse = function(){ // placeholder - replaced by _finishInit } this._backoff = function(){ if(!this._advice){ this._advice={reconnect:"retry",interval:0}; }else if(!this._advice.interval){ this._advice.interval = 0; } if(this._backoffInterval < this._backoffMax){ this._backoffInterval += this._backoffIncrement; } } this._backon = function(){ this._backoffInterval=0; } this._interval = function(){ var i = this._backoffInterval + (this._advice ? (this._advice.interval ? this._advice.interval : 0) : 0); if (i>0){ console.log("Retry in interval+backoff=" + this._advice.interval + "+" + this._backoffInterval+"="+i+"ms"); } return i; } this._publishMeta = function(action,successful,props){ try { var meta = {cometd:this,action:action,successful:successful,state:this.state()}; if (props){ dojo.mixin(meta, props); } dojo.publish(this.prefix + "/meta", [meta]); } catch(e) { console.log(e); } } this._finishInit = function(data){ // summary: // Handle the handshake return from the server and initialize // connection if all is OK if(this._status!="handshaking") {return;} var wasHandshook = this._handshook; var successful = false; var metaMsg = {}; if (data instanceof Error) { dojo.mixin(metaMsg,{ reestablish:false, failure: true, error: data, advice: this._advice }); } else { data = data[0]; data = this._extendIn(data); this.handshakeReturn = data; // remember any advice if(data["advice"]){ this._advice = data.advice; } successful = data.successful ? data.successful : false; // check version if(data.version < this.minimumVersion){ if (console.log) console.log("cometd protocol version mismatch. We wanted", this.minimumVersion, "but got", data.version); successful=false; this._advice.reconnect="none"; } dojo.mixin(metaMsg,{reestablish: successful && wasHandshook, response:data}); } this._publishMeta("handshake",successful,metaMsg); //in the meta listeners, disconnect() may have been called, so recheck it now to //prevent resends or continuing with initializing the protocol if(this._status!="handshaking") {return;} // If all OK if(successful){ this._status = "connecting"; this._handshook = true; // pick a transport this.currentTransport = dojox.cometd.connectionTypes.match( data.supportedConnectionTypes, data.version, this._isXD ); var transport = this.currentTransport; // initialize the transport transport._cometd = this; transport.version = data.version; this.clientId = data.clientId; this.tunnelInit = transport.tunnelInit && dojo.hitch(transport, "tunnelInit"); this.tunnelCollapse = transport.tunnelCollapse && dojo.hitch(transport, "tunnelCollapse"); transport.startup(data); }else{ // If there is a problem follow advice if(!this._advice || this._advice["reconnect"] != "none"){ setTimeout(dojo.hitch(this, "init", this.url, this._props), this._interval()); } } } // FIXME: lots of repeated code...why? this._extendIn = function(message){ // summary: Handle extensions for inbound messages dojo.forEach(dojox.cometd._extendInList, function(f){ message = f(message) || message; }); return message; } this._extendOut = function(message){ // summary: Handle extensions for inbound messages dojo.forEach(dojox.cometd._extendOutList, function(f){ message = f(message) || message; }); return message; } this.deliver = function(messages){ dojo.forEach(messages, this._deliver, this); return messages; } this._deliver = function(message){ // dipatch events along the specified path message = this._extendIn(message); if(!message["channel"]){ if(message["success"] !== true){ return; } } this.lastMessage = message; if(message.advice){ this._advice = message.advice; // TODO maybe merge? } // check to see if we got a /meta channel message that we care about var deferred=null; if( (message["channel"]) && (message.channel.length > 5) && (message.channel.substr(0, 5) == "/meta")){ // check for various meta topic actions that we need to respond to switch(message.channel){ case "/meta/connect": var metaMsg = {response: message}; if(message.successful) { if (this._status != "connected"){ this._status = "connected"; this.endBatch(); } } if(this._initialized){ this._publishMeta("connect",message.successful, metaMsg); } break; case "/meta/subscribe": deferred = this._deferredSubscribes[message.subscription]; try { if(!message.successful){ if(deferred){ deferred.errback(new Error(message.error)); } this.currentTransport.cancelConnect(); return; } if(deferred){ deferred.callback(true); } this.subscribed(message.subscription, message); } catch(e) { log.warn(e); } break; case "/meta/unsubscribe": deferred = this._deferredUnsubscribes[message.subscription]; try { if(!message.successful){ if(deferred){ deferred.errback(new Error(message.error)); } this.currentTransport.cancelConnect(); return; } if(deferred){ deferred.callback(true); } this.unsubscribed(message.subscription, message); } catch(e) { log.warn(e); } break; default: if(message.successful && !message.successful){ this.currentTransport.cancelConnect(); return; } } } // send the message down for processing by the transport this.currentTransport.deliver(message); if(message.data){ // dispatch the message to any locally subscribed listeners try{ var messages = [message]; // Determine target topic var tname = prefix + message.channel; // Deliver to globs that apply to target topic var tnameParts = message.channel.split("/"); var tnameGlob = prefix; for (var i = 1; i < tnameParts.length - 1; i++){ dojo.publish(tnameGlob + "/**", messages); tnameGlob += "/" + tnameParts[i]; } dojo.publish(tnameGlob + "/**", messages); dojo.publish(tnameGlob + "/*", messages); // deliver to target topic dojo.publish(tname,messages); }catch(e){ console.log(e); } } } this._sendMessage = function(/* object */ message){ if(this.currentTransport && !this.batch){ return this.currentTransport.sendMessages([message]); }else{ this._messageQ.push(message); return null; } } this.startBatch = function(){ this.batch++; } this.endBatch = function(){ if(--this.batch <= 0 && this.currentTransport && this._status == "connected"){ this.batch = 0; var messages = this._messageQ; this._messageQ = []; if(messages.length > 0){ this.currentTransport.sendMessages(messages); } } } this._onUnload = function(){ // make this the last of the onUnload method dojo.addOnUnload(dojox.cometd, "disconnect"); } this._connectTimeout = function(){ // summary: Return the connect timeout in ms, calculated as the minimum of the advised timeout // and the configured timeout. Else 0 to indicate no client side timeout var advised=0; if(this._advice && this._advice.timeout && this.expectedNetworkDelay > 0){ advised = this._advice.timeout + this.expectedNetworkDelay; } if(this.connectTimeout > 0 && this.connectTimeout < advised){ return this.connectTimeout; } return advised; } }, // connectionTypes are shared by all cometd Connection. connectionTypes : new dojo.AdapterRegistry(true) } // create the default instance dojox.cometd.Connection.call(dojox.cometd,"/cometd"); /* FIXME: TODOC: this info should be part of the relevant functions and/or overview so the parser can find it. transport objects MUST expose the following methods: - check - startup - sendMessages - deliver - disconnect optional, standard but transport dependent methods are: - tunnelCollapse - tunnelInit Transports SHOULD be namespaced under the cometd object and transports MUST register themselves with cometd.connectionTypes here's a stub transport defintion: cometd.blahTransport = new function(){ this._connectionType="my-polling"; this._cometd=null; this.lastTimestamp = null; this.check = function(types, version, xdomain){ // summary: // determines whether or not this transport is suitable given a // list of transport types that the server supports return dojo.inArray(types, "blah"); } this.startup = function(){ if(dojox.cometd._polling){ return; } // FIXME: fill in startup routine here dojox.cometd._polling = true; } this.sendMessages = function(message){ // FIXME: fill in message array sending logic } this.deliver = function(message){ } this.disconnect = function(){ // send orderly disconnect message } this.cancelConnect = function(){ // cancel the current connection } } cometd.connectionTypes.register("blah", cometd.blahTransport.check, cometd.blahTransport); */ dojo.addOnUnload(dojox.cometd, "_onUnload");