Comet Pipe Demo

[{"_1":2,"_3":-5,"_4":-5},"loaderData",{"_5":6,"_7":8},"actionData","errors","root",{"_11":-5},"routes/user.$name.collections",{"_9":10,"_11":12,"_13":14,"_15":16,"_17":18,"_19":16},"topics",[36],"user",{"_20":21,"_13":14,"_22":23,"_24":25,"_26":27,"_28":29,"_30":31,"_32":33,"_34":35},"loginname","inetfuture","page",1,"limit",20,"total","id","6127","avatar_url","http://www.gravatar.com/avatar/0893f12ed69264db6f7a0e81eb82ef2c?size=48","githubUsername","","create_at","2013-01-03 13:49:39.381","score",65,"topic_count",2,"reply_count",11,"is_star",false,{"_20":37,"_38":39,"_40":-5,"_41":42,"_43":44,"_45":46,"_47":35,"_48":35,"_32":49,"_50":51,"_26":52,"_53":54},"145","author_id","2","tab","content","

最近考虑把整个前端架构使用http streaming方式实现
\r\n
对这方面做了一下调研,并在在node上实现了一个简单的原型

\r\n

顺带提一下,
\r\n
楼下pengchun同学所提到的node chat使用的是longpoll的模型
\r\n
和httpstreaming同属与http comet的解决方案.
\r\n
不过在具体http连接的处理上有所不同
\r\n
long poll在数据通道每次接收到一个数据包后即关闭连接,并立即重新打开数据通道
\r\n
http streaming的数据通道始终处于打开状态.
\r\n
具体的介绍可以看这里 http://en.wikipedia.org/wiki/Comet_(programming)#Streaming

\r\n

一些细节:
\r\n
由于ie下在xhr readystate=3时无法取得responseText,
\r\n
因此在ie下改为通过使用htmlfile控件调用iframe实现
\r\n
另在输出正式数据前需现输出1k的噪音数据
\r\n
以解决浏览器的阻塞问题\r\n

\r\n

原型功能设计如下

\r\n

$\"pipe.png\"$

\r\n

具体代码如下

\r\n

pipe.js: 主服务程序

\r\n

\r\n
var http = require('http'),\r\n
\tfs = require('fs'),\r\n
\turl = require('url'),\r\n
\tpage = null;\r\n
// static files read & watch\r\n
var readFile = function(files) {\r\n
\tvar buffers = {};\r\n
\t// sync read\r\n
\tvar fread = function(file, cb){\r\n
\t\tfs.readFile(file, 'binary', function(err, data){\r\n
\t\t\tif (err) {\r\n
\t\t\t\tthrow err;\r\n
\t\t\t}\r\n
\t\t\tbuffers[file] = new Buffer(data, 'binary');\r\n
\t\t\tconsole.log('load', file)\r\n
\t\t});\r\n
\t}\r\n
\t// watch changes\r\n
\tvar watch = function watch(file) {\r\n
\t\tfs.watchFile(file, {persistent: true, interval: 100}, function (curr, prev) {\r\n
\t\t\tif (curr.mtime.getTime() != prev.mtime.getTime()) {\r\n
\t\t\t\tfread(file);\r\n
\t\t\t}\r\n
\t\t});\r\n
\t}\r\n
\t// run all files\r\n
\tfor (var i = 0; i < files.length; i++) {\r\n
\t\twatch(files[i]);\r\n
\t\tfread(files[i]);\r\n
\t}\r\n
\treturn buffers;\r\n
}\r\n
// http query\r\n
var httpQuery = function(u, cb){\r\n
\tconsole.log('http begin');\r\n
\t// parse url\r\n
\tvar uinfo = url.parse(u);\r\n
\t// create client\r\n
\tvar client = http.createClient(uinfo.port ? uinfo.port : 80, uinfo.hostname, false);\r\n
\tvar uri = uinfo.pathname + (uinfo.search ? uinfo.search : '');\r\n
\tvar req = client.request('GET', uri, {'host': uinfo.hostname});\r\n
\t// send request\r\n
\treq.end();\r\n
\tconsole.log('http request sent');\r\n
\r\n
\r\n
\tvar len = 4096;\r\n
\tvar pointer = 0;\r\n
\tvar extendFactor = 2;\r\n
\t// response start\r\n
\treq.on('response', function (res) {\r\n
\t\tif (res.headers['content-length']) {\r\n
\t\t\tlen = parseInt(res.headers['content-length']);\r\n
\t\t}\r\n
\t\t// body init\r\n
\t\tvar body = new Buffer(len);\r\n
\t\t// chunk recived\r\n
\t\tres.on('data', function(chunk){\r\n
\t\t\t// extends\r\n
\t\t\tif (pointer + chunk.length > len) {\r\n
\t\t\t\tlen *= extendFactor;\r\n
\t\t\t\tbody = body.copy(new Buffer(len), 0, 0);\r\n
\t\t\t\tconsole.log('proxy extend to', len);\r\n
\t\t\t}\r\n
\t\t\t// copy chunk to buf\r\n
\t\t\tchunk.copy(body, pointer, 0);\r\n
\t\t\t// move pointer\r\n
\t\t\tpointer += chunk.length;\r\n
\t\t})\r\n
\t\t// response end\r\n
\t\tres.on('end', function() {\r\n
\t\t\tcb(body.length > pointer ? body.slice(0, pointer) : body);\r\n
\t\t\tconsole.log('proxy end', pointer);\r\n
\t\t});\r\n
\t})\r\n
}\r\n
\r\n
// main server\r\n
var server = http.createServer(function (req, res){\r\n
\t// main page\r\n
\tif (req.url == '/') {\r\n
\t\tres.writeHeader(200);\r\n
\t\tres.end(page[\"pipe.html\"]);\r\n
\t// time serve\r\n
\t} else if (req.url == '/time') {\r\n
\t\tres.writeHeader(200);\r\n
\t\tres.end(new Date().toString());\r\n
\t// iframe recv\r\n
\t} else if (req.url.match(/^\\/iframe\\//)) {\r\n
\t\tvar clientid = parseInt(req.url.substr(8));\r\n
\t\tpipeClient.add(clientid, res, pipeClient.iframe);\r\n
\t\tconsole.log('iframe connect', clientid);\r\n
\t// ajax recv\r\n
\t} else if (req.url.match(/^\\/ajax\\//)) {\r\n
\t\tvar clientid = parseInt(req.url.substr(6));\r\n
\t\tpipeClient.add(clientid, res, pipeClient.ajax);\r\n
\t\tconsole.log('ajax connect', clientid);\r\n
\t// request listen\r\n
\t} else if (req.url.match(/^\\/req\\//)) {\r\n
\t\tres.writeHeader(200,{\r\n
\t\t\t'Cache-Control': 'no-cache, must-revalidate'\r\n
\t\t});\r\n
\t\tres.end();\r\n
\t\t// url parse\r\n
\t\tvar clientid = parseInt(req.url.substr(5, 13));\r\n
\t\t// get page\r\n
\t\thttpQuery(\"http://localhost:8000/time\", function (data){\r\n
\t\t\tconsole.log(data.toString());\r\n
\t\t\tpipeClient.write(clientid, data);\r\n
\t\t\tconsole.log(\"write\", clientid, data.length);\r\n
\t\t});\r\n
\t// error pages\r\n
\t} else {\r\n
\t\tres.writeHeader(404, {\"Content-Type\" : \"text/html\"});\r\n
\t\tres.end();\r\n
\t}\r\n
});\r\n
\r\n
var pipeClient = {\r\n
\ttimeout : 30000,\r\n
\tclient : {},\r\n
\tprefix : \"\",\r\n
\tiframe : 'iframe',\r\n
\tajax : 'ajax',\r\n
\tnoise : null,\r\n
\tnoiseSize : 1024,\r\n
\tpage : null,\r\n
\tinit : function(){\r\n
\t\tthis.noise = new Buffer(1024);\r\n
\t\tfor (var i = 0; i < this.noiseSize; i++) {\r\n
\t\t\tthis.noise[i] = 32;\r\n
\t\t}\r\n
\t\tthis.page = readFile(['iframe.html']);\r\n
\t},\r\n
\tadd : function(id, res, type) {\r\n
\r\n
\t\tif (type == this.ajax) {\r\n
\t\t\tres.writeHeader(200, {\r\n
\t\t\t\t'Cache-Control': 'no-cache, must-revalidate'\r\n
\t\t\t});\r\n
\t\t\tres.write(this.noise);\r\n
\t\t} else {\r\n
\t\t\tres.writeHeader(200, {\r\n
\t\t\t\t\"Content-Type\" : \"multipart/x-mixed-replace\",\r\n
\t\t\t\t'Cache-Control': 'no-cache, must-revalidate'\r\n
\t\t\t});\r\n
\t\t\tres.write(this.page['iframe.html']);\r\n
\t\t\tres.write(this.noise);\r\n
\t\t}\r\n
\t\tthis.client[id] = {\r\n
\t\t\tres : res,\r\n
\t\t\ttype : type,\r\n
\t\t\ttm : setTimeout(function(){\r\n
\t\t\t\tpipeClient.close(id);\r\n
\t\t\t}, this.timeout)\r\n
\t\t};\r\n
\t},\r\n
\tclose : function (id) {\r\n
\t\tconsole.log(\"client close\", id)\r\n
\t\tthis.client[id].res.end();\r\n
\t\tthis.client[id].res = null;\r\n
\t\tdelete this.client[id];\r\n
\t},\r\n
\twrite : function (id, data) {\r\n
\t\tclearTimeout(this.client[id].tm);\r\n
\t\tthis.client[id].tm = setTimeout(function(){\r\n
\t\t\tpipeClient.close(id);\r\n
\t\t}, this.timeout);\r\n
\t\tthis.client[id].res.write(this.format(data, this.client[id].type));\r\n
\r\n
\t},\r\n
\tformat : function(data, type) {\r\n
\t\t// with iframe\r\n
\t\tif (type == this.iframe) {\r\n
\t\t\tvar buf = new Buffer(this.prefix.length + data.length + this.suffix.length);\r\n
\t\t\tbuf.write(this.prefix, 0, 'binary');\r\n
\t\t\tdata.copy(buf, this.prefix.length, 0);\r\n
\t\t\tbuf.write(this.suffix, this.prefix.length + data.length);\r\n
\t\t// with ajax\r\n
\t\t} else {\r\n
\t\t\tvar buf = new Buffer(data.length + 8);\r\n
\t\t\t// set length\r\n
\t\t\tbuf.write(data.length.toString(16), 0, 'binary');\r\n
\t\t\t// space padding\r\n
\t\t\tfor (var i = data.length.toString(16).length; i < 8; i++) {\r\n
\t\t\t\tbuf[i] = 32;\r\n
\t\t\t}\r\n
\t\t\t// set data\r\n
\t\t\tdata.copy(buf, 8, 0);\r\n
\t\t}\r\n
\t\tconsole.log(buf.toString());\r\n
\t\treturn buf;\r\n
\t}\r\n
}\r\n
pipeClient.init();\r\n
\r\n
page = readFile(['pipe.html']);\r\n
setTimeout(function(){\r\n
\tserver.listen(8000);\r\n
}, 500);\r\n

\r\n
\r\n

pipe.html: 客户端程序

\r\n

\r\n
\r\n
\r\n
\r\n
\r\n
\r\n
Comet Pipe Demo\r\n
\r\n
\r\n
\r\n
\r\n
\r\n
\r\n
\r\n
\r\n
\r\n

\r\n

iframe.html: ie下iframe模式运行的输出头

\r\n

\r\n
\r\n
\r\n
\r\n
\r\n
\r\n
\r\n

","title","Node 下 Http Streaming 的跨浏览器实现","last_reply_at","2012-05-24 11:26:47.249","good","top",6,"visit_count",12093,"2011-01-21 07:28:52","author",{"_13":55,"_22":56},"qingdu","http://www.gravatar.com/avatar/2e26ae3b35ec2d132acc318e7b3a4abb?s=48"]