学习Pushlet(二)：看源码了解pushlet

系统管理员 2022-01-06 03:09 251阅读 0赞

首先我们需要去了解一下pushlet的代码，从而进行我们自己的自定义需求，首先我们找到我们写的后台服务，有两个方法一个设置睡眠时间和生成事件，那么到底有什么用呢，哪里用到了呢。我们将从官网下载的源码也放入到项目中去，我们进入到HelloWorldPlushlet继承的EventPullSource类里面，原来它继承了Runnable，是个线程，那么查看run方法：发现这里用到了我们自定义的两个方法，通过设置的睡眠时间让线程睡眠，一个通过生成事件的方法生成事件并且发布出去，那么到这里我们大概明白了，原来pushlet后台是生成了一个线程，这个线程不断生成事件并且发布出去，然后睡眠，一直循环。这就是生产信息的那个线程。

/**
         * Main loop: sleep, generate event and publish.
         */
        public void run() {
            Log.debug(getClass().getName() + ": starting...");
            alive = true;
            while (alive) {
                try {
    
                    Thread.sleep(getSleepTime());
    
                    // Stopped during sleep: end loop.
                    if (!alive) {
                        break;
                    }
    
                    // If passivated wait until we get
                    // get notify()-ied. If there are no subscribers
                    // it wasts CPU to remain producing events...
                    synchronized (this) {
                        while (!active) {
                            Log.debug(getClass().getName() + ": waiting...");
                            wait();
                        }
                    }
    
                } catch (InterruptedException e) {
                    break;
                }
    
                try {
                    // Derived class should produce an event.
                    Event event = pullEvent();
    
                    // Let the publisher push it to subscribers.
                    Dispatcher.getInstance().multicast(event);
                } catch (Throwable t) {
                    Log.warn("EventPullSource exception while multicasting ", t);
                    t.printStackTrace();
                }
            }
            Log.debug(getClass().getName() + ": stopped");
        }
    }

那么前台是怎么访问的呢，我们打开浏览器f12看看前台都发送了哪些请求:

![1663969-20190516230013100-973607636.png][]

第三个和第四个都有pushlet.srv，是不是很眼熟，这不就是我们前面web.xml配置的链接，这两个链接都是pushlet servlet管理了，第三个链接就是前台告诉pushlet我监控了什么事件了，第四链接是每次经过定时时间，都会发送这个链接，很明显这是用来接收后台的生产线程消息的。好到这里我们大体了解了流程，后台有一个随着我们项目启动就会启动的生产事件并广播事件的线程，前台会通国joinListen监听方法建立连接，监听事件，然后每隔一定时间就会发送一个refresh事件到后台，从而获得后台产生的消息。这也就是为什么我前面说pushlet表面是服务器推，但是实际上还是前台拉的原因。

接下来继续详细的看代码，那么上面说的第三和第四个请求是怎么发出去的，需要看我们的jsajax-pushlet-client.js，我们先在jsp上调用了iPL.init()方法。看看init方法都干了什么：

_init: function () {
            PL._showStatus();
            PL._setStatus('initializing...');
            /*
                Setup Cross-Browser XMLHttpRequest v1.2
               Emulate Gecko 'XMLHttpRequest()' functionality in IE and Opera. Opera requires
               the Sun Java Runtime Environment <http://www.java.com/>.
    
               by Andrew Gregory
               http://www.scss.com.au/family/andrew/webdesign/xmlhttprequest/
    
               This work is licensed under the Creative Commons Attribution License. To view a
               copy of this license, visit http://creativecommons.org/licenses/by-sa/2.5/ or
               send a letter to Creative Commons, 559 Nathan Abbott Way, Stanford, California
               94305, USA.
    
               */
            // IE support
            if (window.ActiveXObject && !window.XMLHttpRequest) {
                window.XMLHttpRequest = function() {
                    var msxmls = new Array(
                            'Msxml2.XMLHTTP.5.0',
                            'Msxml2.XMLHTTP.4.0',
                            'Msxml2.XMLHTTP.3.0',
                            'Msxml2.XMLHTTP',
                            'Microsoft.XMLHTTP');
                    for (var i = 0; i < msxmls.length; i++) {
                        try {
                            return new ActiveXObject(msxmls[i]);
                        } catch (e) {
                        }
                    }
                    return null;
                };
            }
    
            // ActiveXObject emulation
            if (!window.ActiveXObject && window.XMLHttpRequest) {
                window.ActiveXObject = function(type) {
                    switch (type.toLowerCase()) {
                        case 'microsoft.xmlhttp':
                        case 'msxml2.xmlhttp':
                        case 'msxml2.xmlhttp.3.0':
                        case 'msxml2.xmlhttp.4.0':
                        case 'msxml2.xmlhttp.5.0':
                            return new XMLHttpRequest();
                    }
                    return null;
                };
            }
    
            PL.pushletURL = PL._getWebRoot() + 'pushlet.srv';
            PL._setStatus('initialized');
            PL.state = PL.STATE_READY;
        },

主要是创建了XMLHttpRequest对象用来发送请求，然后注意到在拼接url的时候，拼接了'pushlet.srv'，是不是很眼熟，就是我们在web.xml中配置的过滤url，所以说web.xml要是修改的话，js也要修改。然后我们看看PL.\_getWebRoot()方法：

_getWebRoot: function() {
            /** Return directory of this relative to document URL. */
            if (PL.webRoot != null) {
                return PL.webRoot;
            }
            //derive the baseDir value by looking for the script tag that loaded this file
            var head = document.getElementsByTagName('head')[0];
            var nodes = head.childNodes;
            for (var i = 0; i < nodes.length; ++i) {
                var src = nodes.item(i).src;
                if (src) {
                    var index = src.indexOf("ajax-pushlet-client.js");
                    if (index >= 0) {
                        index = src.indexOf("lib");
                        PL.webRoot = src.substring(0, index);
                        break;
                    }
                }
            }
            return PL.webRoot;

其实就是获取路径的，实际项目中我们一般把js不放到webcontent下面的时候，这个时候这个方法往往给我拼接错，从而浏览器控制台会报4040错误，所以我们最好把这个方法修改如下：

_getWebRoot: function() {
    
    /** Return directory of this relative to document URL. */
    
    if (PL.webRoot != null) {
    
    return PL.webRoot;
    
    }
    
    //derive the baseDir value by looking for the script tag that loaded this file
    
    //获取当前网址，如： http://localhost:8080/PushletNote/index.jsp
    
    var webPath = window.document.location.href;
    
    //获取主机地址之后的目录，如： /PushletNote/index.jsp
    
    var pathName = window.document.location.pathname;
    
    //获取主机地址，如： http://localhost:8080
    
    var hostPaht = webPath.substring(0,webPath.indexOf(pathName));
    
    //获取带"/"的项目名，如：/Pushlet
    
    var projectName = pathName.substring(0,pathName.substr(1).indexOf('/')+1);
    
    PL.webRoot = hostPaht + projectName + "/";
    
    return PL.webRoot;
    
    },

然后是执行PL.joinListen(),开始调用PL.\_doRequest('join-listen', query);发送请求。声明了是join-listen事件。

joinListen: function(aSubject) {
            PL._setStatus('join-listen ' + aSubject);
            // PL.join();
            // PL.listen(aSubject);
    
            PL.sessionId = null;
            // Create event URI for listen
            var query = PL.NV_P_FORMAT + '&' + PL.NV_P_MODE;
    
            // Optional subject to subscribe to
            if (aSubject) {
                query = query + '&p_subject=' + aSubject;
            }
    
            PL._doRequest('join-listen', query);
    
        },

然后执行了PL.\_doRequest方法，看看代码：主要是拼接了请求参数，然后发送请求，所有日后我们扩展请求参数的时候就修改此方法。

_doRequest: function(anEvent, aQuery) {
            // Check if we are not in any error state
            if (PL.state < 0) {
                PL._setStatus('died (' + PL.state + ')');
                return;
            }
    
            // We may have (async) requests outstanding and thus
            // may have to wait for them to complete and change state.
            var waitForState = false;
            if (anEvent == 'join' || anEvent == 'join-listen') {
                // We can only join after initialization
                waitForState = (PL.state < PL.STATE_READY);
            } else if (anEvent == 'leave') {
                PL.state = PL.STATE_READY;
            } else if (anEvent == 'refresh') {
                // We must be in the listening state
                if (PL.state != PL.STATE_LISTENING) {
                    return;
                }
            } else if (anEvent == 'listen') {
                // We must have joined before we can listen
                waitForState = (PL.state < PL.STATE_JOINED);
            } else if (anEvent == 'subscribe' || anEvent == 'unsubscribe') {
                // We must be listeing for subscription mgmnt
                waitForState = (PL.state < PL.STATE_LISTENING);
            } else {
                // All other requests require that we have at least joined
                waitForState = (PL.state < PL.STATE_JOINED);
            }
    
            // May have to wait for right state to issue request
            if (waitForState == true) {
                PL._setStatus(anEvent + ' , waiting... state=' + PL.state);
                setTimeout(function() {
                    PL._doRequest(anEvent, aQuery);
                }, 100);
                return;
            }
    
            // ASSERTION: PL.state is OK for this request
    
            // Construct base URL for GET
            var url = PL.pushletURL + '?p_event=' + anEvent;
    
            // Optionally attach query string
            if (aQuery) {
                url = url + '&' + aQuery;
            }
    
            PL.debug('_doRequest', url);
            PL._getXML(url, PL._onResponse);
    
            // uncomment to use synchronous XmlHttpRequest
            //var rsp = PL._getXML(url);
            //PL._onResponse(rsp);  */
        },

方法的最后调用了getXML方法，传入了参数是url和回调函数名，可以看出是通过xmlhttp.open和xmlhttp.send方法发送请求的；如果我们自己写发送请求就可以仿照这个方法写。此方法会接受到后台返回的xml格式，然后交给回调函数处理。

_getXML: function(url, callback) {
    
            // Obtain XMLHttpRequest object
            var xmlhttp = new XMLHttpRequest();
            if (!xmlhttp || xmlhttp == null) {
                alert('No browser XMLHttpRequest (AJAX) support');
                return;
            }
    
            // Setup optional async response handling via callback
            var cb = callback;
            var async = false;
    
            if (cb) {
                // Async mode
                async = true;
                xmlhttp.onreadystatechange = function() {
                    if (xmlhttp.readyState == 4) {
                        if (xmlhttp.status == 200) {
                            // Processing statements go here...
                            cb(xmlhttp.responseXML);
    
                            // Avoid memory leaks in IE
                            // 12.may.2007 thanks to Julio Santa Cruz
                            xmlhttp = null;
                        } else {
                            var event = new PushletEvent();
                            event.put('p_event', 'error')
                            event.put('p_reason', '[pushlet] problem retrieving XML data:\n' + xmlhttp.statusText);
                            PL._onEvent(event);
                        }
                    }
                };
            }
            // Open URL
            xmlhttp.open('GET', url, async);
    
            // Send XML to KW server
            xmlhttp.send(null);
    
            if (!cb) {
                if (xmlhttp.status != 200) {
                    var event = new PushletEvent();
                    event.put('p_event', 'error')
                    event.put('p_reason', '[pushlet] problem retrieving XML data:\n' + xmlhttp.statusText);
                    PL._onEvent(event)
                    return null;
                }
                // Sync mode (no callback)
                // alert(xmlhttp.responseText);
    
                return xmlhttp.responseXML;
            }
        },

这个方法执行完，会执行回调方法\_onResponse方法：这个方法经过PL.\_rsp2Events(xml);方法将返回的xml解析，然后交给onevent（）方法处理

_onResponse: function(xml) {
            PL.debug('_onResponse', xml);
            var events = PL._rsp2Events(xml);
            if (events == null) {
                PL._setStatus('null events')
                return;
            }
    
            delete xml;
    
            PL.debug('_onResponse eventCnt=', events.length);
            // Go through all <event/> elements
            for (i = 0; i < events.length; i++) {
                PL._onEvent(events[i]);
            }
        },

PL._onEvent主要通过eventType来判断后端返回的事件类型，然后执行什么方法，经过是这样的：当我们第一次发送join-listen：“http://localhost:8080/pushlet.srv?p_event=join-listen&p_format=xml-strict&p_mode=pull&p_subject=/test”的时候，后台返回的事件类型是“join-listen-ack”确认监听成功，
    然后又返回了“refresh”事件类型，前台会通过定时器不断的发送请求，前台发送refresh事件：“http://localhost:8080/pushlet.srv?p_event=refresh&p_id=qahozocywy”，后台返回的是“refresh-ack”事件类型，然后返回“data”类型，并且返回了数据。这时会调用window.onData（）方法处理返回事件，这就是为什么我们要用ondata函数处理返回信息了。这就是为什么打开f12,会看到不断的定时的发送refresh事件的请求了。

_onEvent: function (event) {
            // Create a PushletEvent object from the arguments passed in
            // push.arguments is event data coming from the Server
    
            PL.debug('_onEvent()', event.toString());
    
            // Do action based on event type
            var eventType = event.getEvent();
    
            if (eventType == 'data') {
                PL._setStatus('data');
                PL._doCallback(event, window.onData);
            } else if (eventType == 'refresh') {
                if (PL.state < PL.STATE_LISTENING) {
                    PL._setStatus('not refreshing state=' + PL.STATE_LISTENING);
                }
                var timeout = event.get('p_wait');
                setTimeout(function () {
                    PL._doRequest('refresh');
                }, timeout);
                return;
            } else if (eventType == 'error') {
                PL.state = PL.STATE_ERROR;
                PL._setStatus('server error: ' + event.get('p_reason'));
                PL._doCallback(event, window.onError);
            } else if (eventType == 'join-ack') {
                PL.state = PL.STATE_JOINED;
                PL.sessionId = event.get('p_id');
                PL._setStatus('connected');
                PL._doCallback(event, window.onJoinAck);
            } else if (eventType == 'join-listen-ack') {
                PL.state = PL.STATE_LISTENING;
                PL.sessionId = event.get('p_id');
                PL._setStatus('join-listen-ack');
                PL._doCallback(event, window.onJoinListenAck);
            } else if (eventType == 'listen-ack') {
                PL.state = PL.STATE_LISTENING;
                PL._setStatus('listening');
                PL._doCallback(event, window.onListenAck);
            } else if (eventType == 'hb') {
                PL._setStatus('heartbeat');
                PL._doCallback(event, window.onHeartbeat);
            } else if (eventType == 'hb-ack') {
                PL._doCallback(event, window.onHeartbeatAck);
            } else if (eventType == 'leave-ack') {
                PL._setStatus('disconnected');
                PL._doCallback(event, window.onLeaveAck);
            } else if (eventType == 'refresh-ack') {
                PL._doCallback(event, window.onRefreshAck);
            } else if (eventType == 'subscribe-ack') {
                PL._setStatus('subscribed to ' + event.get('p_subject'));
                PL._doCallback(event, window.onSubscribeAck);
            } else if (eventType == 'unsubscribe-ack') {
                PL._setStatus('unsubscribed');
                PL._doCallback(event, window.onUnsubscribeAck);
            } else if (eventType == 'abort') {
                PL.state = PL.STATE_ERROR;
                PL._setStatus('abort');
                PL._doCallback(event, window.onAbort);
            } else if (eventType.match(/nack$/)) {
                PL._setStatus('error response: ' + event.get('p_reason'));
                PL._doCallback(event, window.onNack);
            }
        },

好，代码了解的差不多了，pushlet的原理及代码经过也有了大致的了解了，下一篇会进行后台源码的阅读学习，当我们了解了源码后，我们开始介绍实际项目中的使用，这样才能在实际项目中应用的更加灵活。

转载于:https://www.cnblogs.com/liuyuan1227/p/10878606.html

[1663969-20190516230013100-973607636.png]: /images/20211228/b77b7e47c3ea4e3889d868c0d1243d4e.png