深入了解 cometd的服务器推送技术

				 <form action="" method="get" οnsubmit="comet.doRequest($('word').value);$('word').value='';return false;"> <input type="text" name="word" id="word" value="" /> <input type="submit" name="submit" value="Send" /> </form> 

				 var Comet = Class.create();
Comet.prototype = {timestamp: 0,url: './backend.php',noerror: true,initialize: function(){},connect: function(){this.ajax = new Ajax.Request(this.url, {method: 'get',parameters: {'timestamp': this.timestamp},onSuccess: function(transport){var response = transport.responseText.evalJSON();this.comet.timestamp = response['timestamp'];this.comet.handleResponse(response);this.comet.noerror = true;},onComplete: function(transport){if (!this.comet.noerror) setTimeout(function(){comet.connect()}, 5000);else this.comet.connect();this.comet.noerror = false;}});this.ajax.comet = this;},handleResponse: function(response){$('content').innerHTML += '<div>' + response['msg'] + '</div>';},doRequest: function(request){new Ajax.Request(this.url, {method: 'get',parameters: {'msg': request}});}
}var comet = new Comet();
comet.connect();

				 $filename  = dirname(__FILE__).'/data.txt'; // 将新消息存入文件中$msg = isset($_GET['msg']) ? $_GET['msg'] : ''; if ($msg != '') { file_put_contents($filename,$msg); die(); } // 这是一个无限循环，一旦发现文件被修改，便会跳出循环并返回文件修改数据。如果文件一直没有修改，则会一 // 直处于循环检测状态，此时的 Ajax 连接也会一直保留，直到文件被修改为止，这就是所谓的“长轮询”。$lastmodif    = isset($_GET['timestamp']) ? $_GET['timestamp'] : 0; $currentmodif = filemtime($filename); while ($currentmodif <= $lastmodif) // 检测文件是否被修改{ usleep(10000); // sleep 10ms to unload the CPU clearstatcache(); $currentmodif = filemtime($filename); } // 返回 JSON 数组$response = array(); $response['msg']       = file_get_contents($filename); $response['timestamp'] = $currentmodif; echo json_encode($response); flush(); 

				 dojox.cometd.init("http://www.xxx.com/cometd"); 

				 this.init = function(...){
............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(this._isXD){r = dojo.io.script.get(bindArgs);}else{r = dojo.xhrPost(bindArgs);}
..............
}this._finishInit = function(data){
..................if(successful){........//ajax request insidethis.tunnelInit = transport.tunnelInit && dojo.hitch(transport,"tunnelInit");this.tunnelCollapse = transport.tunnelCollapse && dojo.hitch(transport,"tunnelCollapse");transport.startup(data);}else{if(!this._advice || this._advice["reconnect"] != "none"){setTimeout(dojo.hitch(this, "init", this.url, this._props),this._interval());}}
....................
}

				 dojox.cometd.publish("/service/echo", { msg: msg }); 

				 // 方法 1 cometd.batch(function() { cometd.publish('/channel1', { product: 'foo' }); cometd.publish('/channel2', { notificationType: 'all' }); cometd.publish('/channel3', { update: false }); }); // 方法 2 cometd.startBatch() cometd.publish('/channel1', { product: 'foo' }); cometd.publish('/channel2', { notificationType: 'all' }); cometd.publish('/channel3', { update: false }); cometd.endBatch() 

				 dojox.cometd.subscribe("/service/echo",echoRpcReturn); function echoRpcReturn(msg){ dojo.byId("responses").innerHTML += msg; } 

				 <web-app xmlns="http://java.sun.com/xml/ns/javaee"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://java.sun.com/xml/ns/javaee http://java.sun.com/xml/ns/javaee/web-app_2_5.xsd"version="2.5"> <servlet> <servlet-name>cometd</servlet-name> <servlet-class> org.cometd.server.continuation.ContinuationCometdServlet </servlet-class> <init-param> <param-name>timeout</param-name> <param-value>60000</param-value> </init-param> </servlet> <servlet-mapping> <servlet-name>cometd</servlet-name> <url-pattern>/cometd/*</url-pattern> </servlet-mapping> <filter> <filter-name>cross-origin</filter-name> <filter-class>org.eclipse.jetty.servlets.CrossOriginFilter</filter-class> </filter> <filter-mapping> <filter-name>cross-origin</filter-name> <url-pattern>/cometd/*</url-pattern> </filter-mapping> </web-app> 

				 <servlet> <servlet-name>cometd</servlet-name> <servlet-class>org.cometd.java.annotation.AnnotationCometdServlet</servlet-class> <init-param> <param-name>logLevel</param-name> <param-value>1</param-value> </init-param> <init-param> <param-name>services</param-name> <param-value>org.cometd.examples.ChatService</param-value> </init-param> <load-on-startup>1</load-on-startup> </servlet> <servlet-mapping> <servlet-name>cometd</servlet-name> <url-pattern>/cometd/*</url-pattern> </servlet-mapping> <servlet> <servlet-name>cometdDemo</servlet-name> <servlet-class>org.cometd.examples.CometdDemoServlet</servlet-class> <load-on-startup>2</load-on-startup> </servlet> 

				 public void init() throws ServletException { final BayeuxServerImpl bayeux = (BayeuxServerImpl)getServletContext().getAttribute(BayeuxServer.ATTRIBUTE); if (bayeux==null) throw new UnavailableException("No BayeuxServer!"); ................. // 创建扩展点bayeux.addExtension(new TimesyncExtension()); bayeux.addExtension(new AcknowledgedMessagesExtension()); // 设定握手连接权限bayeux.getChannel(ServerChannel.META_HANDSHAKE).addAuthorizer( GrantAuthorizer.GRANT_PUBLISH); // 启动服务渠道ServerAnnotationProcessor processor = new ServerAnnotationProcessor(bayeux); processor.process(new EchoRPC()); processor.process(new Monitor()); //processor.process(new ChatService()); bayeux.createIfAbsent("/foo/bar/baz",new ConfigurableServerChannel.Initializer() { public void configureChannel(ConfigurableServerChannel channel) { channel.setPersistent(true); } }); if (bayeux.getLogger().isDebugEnabled()) System.err.println(bayeux.dump()); ................. } 

				 @Service("echo")
public static class EchoRPC
{@Sessionprivate ServerSession _session;@SuppressWarnings("unused")@Configure("/service/echo")private void configureEcho(ConfigurableServerChannel channel){channel.addAuthorizer(GrantAuthorizer.GRANT_SUBSCRIBE_PUBLISH);}@Listener("/service/echo")public void doEcho(ServerSession session, ServerMessage message){Map<String,Object> data = message.getDataAsMap();Log.info("ECHO from "+session+" "+data);for(int i = 0; i < 50; i++){session.deliver(_session, message.getChannel(), data, null);}}
}

				 @Service("monitor") public static class Monitor { @Listener("/meta/subscribe") public void monitorSubscribe(ServerSession session, ServerMessage message) { Log.info("Monitored Subscribe from "+session+" for "+message.get(Message.SUBSCRIPTION_FIELD)); } @Listener("/meta/unsubscribe") public void monitorUnsubscribe(ServerSession session, ServerMessage message) { Log.info("Monitored Unsubscribe from "+session+" for "+message.get(Message.SUBSCRIPTION_FIELD)); } @Listener("/meta/*") public void monitorMeta(ServerSession session, ServerMessage message) { if (Log.isDebugEnabled()) Log.debug(message.toString()); } } 

				 <servlet> ............... <init-param> <param-name>services</param-name> <param-value>org.cometd.examples.ChatService</param-value> </init-param> <load-on-startup>1</load-on-startup> </servlet> 

				 @Service("chat") public class ChatService { .......................................... @Listener("/service/members") public void handleMembership(ServerSession client, ServerMessage message) { Map<String, Object> data = message.getDataAsMap(); final String room = ((String)data.get("room")).substring("/chat/".length()); Map<String, String> roomMembers = _members.get(room); if (roomMembers == null) { Map<String, String> new_room = new ConcurrentHashMap<String, String>(); roomMembers = _members.putIfAbsent(room, new_room); if (roomMembers == null) roomMembers = new_room; } final Map<String, String> members = roomMembers; String userName = (String)data.get("user"); members.put(userName, client.getId()); client.addListener(new ServerSession.RemoveListener() { public void removed(ServerSession session, boolean timeout) { members.values().remove(session.getId()); broadcastMembers(room,members.keySet()); } }); broadcastMembers(room,members.keySet()); } private void broadcastMembers(String room, Set<String> members) { // Broadcast the new members list ClientSessionChannel channel = _session.getLocalSession().getChannel("/members/"+room); channel.publish(members); } .......................................... @Listener("/service/privatechat") protected void privateChat(ServerSession client, ServerMessage message) { Map<String,Object> data = message.getDataAsMap(); String room = ((String)data.get("room")).substring("/chat/".length()); Map<String, String> membersMap = _members.get(room); if (membersMap==null) { Map<String,String>new_room=new ConcurrentHashMap<String, String>(); membersMap=_members.putIfAbsent(room,new_room); if (membersMap==null) membersMap=new_room; } String[] peerNames = ((String)data.get("peer")).split(","); ArrayList<ServerSession> peers = new ArrayList<ServerSession>(peerNames.length); ................. } }