皮杜妮

[root@f2c node_work]# cat ServiceRoute.js
/*************************
* 服務路由接口
* @author
*
**************************/
var http = require('http'),
        url = require('url'),
        amqplib = require('amqplib'),
        async = require('async'),
        uuid = require('node-uuid');

var open = require('amqplib').connect('amqp://10.0.16.101:5672');
http.createServer(function(req, res){
        /*** 參數判斷 ***/
        var param = url.parse(req.url).query;
        if(param==null || param=="") {
                res.writeHead(200, {'Content-Type':'text/html'});
                res.write("no message", 'utf8');
                res.end();
        }
        else {
                /*** 參數處理 ***/
                console.log("*****************start*******************");
                var paramArr = param.split("&");
                var messageInfo = paramArr[0].split("=");
                var queueName = messageInfo[0];
                var b = new Buffer(messageInfo[1], 'base64');
                var mes = b.toString();
                console.log("*param="+new Date().toLocaleString());
                console.log("*param="+param);
                console.log("*request message = "+mes);
                var callBackInfo = null;
                if(paramArr.length>=2) callBackInfo = paramArr[1].split("=");
                /*** 返回隊列 ***/
                var uuIdStr = uuid.v1()+"_a";
                var common_options = {durable: false, autoDelete:true, exclusive:true};
                /*** 接收消息 ***/
                open.then(function(conn){
                        var  ok = conn.createChannel();
                        ok = ok.then(function(ch){
                                ch.assertQueue(uuIdStr, common_options);

                                var onSecond = 1000 * 1;
                                var timer=setTimeout(function(){
                                        console.log("*setTimeOut");
                                        res.write('{"s":-1, "error":"channel connect time out"}', 'utf8');
                                        ch.close();
                                }, onSecond);

                                ch.consume(uuIdStr, function(msg) {
                                        console.log("*response="+msg.content.toString());
                                        ch.close();
                                        clearTimeout(timer)
                                        /*** 返回信息到客戶端 ***/
                                        if(callBackInfo!=null)
                                                res.write(callBackInfo[1]+"('"+msg.content.toString().replace("'", "\'")+"')", 'utf8');
                                        else
                                                res.write(msg.content.toString(), 'utf8');
                                        res.end();
                                });
                                ch.on("error", function(err){
                                        console.log("*response error="+err);
                                });
                                ch.on("close", function(){
                                        console.log("*response close method is called");
                                });

                        });
                }).then(null, console.warn);
                /*** 發送消息 ***/
                open.then(function(conn) {
                var ok = conn.createChannel();
                ok = ok.then(function(ch){
                                ch.sendToQueue(queueName, new Buffer(mes), {correlationId:uuIdStr});
                                ch.close();
                                ok.then(ok.close.bind(ok));
                        });
                }).then(null, console.warn);

        }
}).listen(8081);