diff --git a/README.md b/README.md
index 2b12981..632d05b 100644
--- a/README.md
+++ b/README.md
@@ -10,7 +10,7 @@ npm install dubbo-node-client
# 注意
- 该项目只支持 [jsonrpc协议](https://github.com/ofpay/dubbo-rpc-jsonrpc), 不支持 dubbo协议的服务提供者
+ 该项目只支持 [jsonrpc协议](https://github.com/ofpay/dubbo-rpc-jsonrpc), 支持 dubbo协议的服务提供者
---
@@ -60,4 +60,25 @@ function doFoo(req, res){
});
}
+
+//新增 DUBBO方式调用如下：
+
+////获取serivce
+var GoodsSkuService = dubboClient.getService('com.xxx.dubbo.coach.api.goods.GoodsSkuService', '1.0');
+
+// dubbo协议方式调用 以私教为例
+var data = {
+ "$class": "com.xxx.dubbo.coach.api.goods.request.LessonSkuTypeReq",
+ "$": {"goodsNo": "01241", "coachId": 120, "userId": 105424}
+};
+GoodsSkuService.callRpc('goodsSkuDetail', data)
+ .then(function (r) {
+ console.info(r);
+ process.exit(0);
+ })
+ .catch(function (e) {
+ console.error(JSON.stringify(e));
+ process.exit(0);
+ });
+}
```
\ No newline at end of file
diff --git a/index.js b/index.js
index 36b7bf2..3b4c05f 100644
--- a/index.js
+++ b/index.js
@@ -22,7 +22,8 @@ module.exports = {
},
/**
- * rpc调用, 主要调用入口
+ * jsonrpc || dubbo 调用, 主要调用入口
+ * 不能同时使用jsonrpc或dubbo协议调用服务
*/
getService: function (serviceName, version, group) {
var invokerDesc = new InvokerDesc(serviceName, group, version),
diff --git a/lib/config/index.js b/lib/config/index.js
index 58b0e5e..a5b6c04 100644
--- a/lib/config/index.js
+++ b/lib/config/index.js
@@ -30,7 +30,8 @@ Config.prototype = {
this.option = option;
this.option.dubbo = this.option.dubbo || {
providerTimeout: 3,
- weight: 1
+ weight: 1,
+ protocol: "jsonrpc" //添加协议类型，支持rpc调用
};
},
@@ -73,6 +74,13 @@ Config.prototype = {
*/
getProviderTimeout: function(){
return this.option.dubbo.providerTimeout * 1000;
+ },
+
+ /**
+ * 获取protocol协议
+ */
+ getProtocol: function(){
+ return this.option.dubbo.protocol;
}
};
diff --git a/lib/registry/index.js b/lib/registry/index.js
index b744202..e8a177d 100644
--- a/lib/registry/index.js
+++ b/lib/registry/index.js
@@ -24,7 +24,7 @@ Registry.prototype.init = function () {
this.isInitializing = true;
this.zookeeper = Zookeeper.createClient(Config.getRegistryAddress(), Config.getRegistryOption());
this.zookeeper.once('connected', function () {
- self.initQueue.forEach(function (p) { //从队列中获取, 租个通知
+ self.initQueue.forEach(function (p) { //从队列中获取, 逐个通知
p.resolve(self.zookeeper);
});
self.isInitializing = false;
@@ -122,8 +122,12 @@ Registry.prototype.subscribe = function (invokerDesc) {
Registry.prototype.onMethodChangeHandler = function (invokerDesc, children) {
children.forEach(function (child) {
child = decodeURIComponent(child);
- var mHost = /^jsonrpc:\/\/([^\/]+)\//.exec(child),
- mVersion = /version=(.+)/.exec(child),
+ var mHost = /^jsonrpc:\/\/([^\/]+)\//.exec(child);
+ if (Config.getProtocol() === "dubbo") {
+ mHost = /^dubbo:\/\/([^\/]+)\//.exec(child);
+ }
+
+ var mVersion = /version=(.+)/.exec(child),
mGroup = /group=([^&]+)/.exec(child),
mMehtod = /methods=([^&]+)/.exec(child);
diff --git a/lib/rpc/api/invoker/desc.js b/lib/rpc/api/invoker/desc.js
index 71cff17..fbf638e 100644
--- a/lib/rpc/api/invoker/desc.js
+++ b/lib/rpc/api/invoker/desc.js
@@ -25,6 +25,9 @@ InvokerDesc.prototype.getService = function () {
return this.serviceName;
};
+InvokerDesc.prototype.getVersion = function() {
+ return this.version;
+};
//-----------------------------------------------------------------------------------------------
//
//-----------------------------------------------------------------------------------------------
diff --git a/lib/rpc/api/invoker/index.js b/lib/rpc/api/invoker/index.js
index 5cc28cb..b11c559 100644
--- a/lib/rpc/api/invoker/index.js
+++ b/lib/rpc/api/invoker/index.js
@@ -1,7 +1,8 @@
var Q = require('q'),
_ = require('underscore'),
Cluster = require('../../cluster/index'),
- HttpClient = require('../../util/Http');
+ HttpClient = require('../../util/Http'),
+ SocketClient = require('../../util/Socket');
//-----------------------------------------------------------------------------------------------
//
@@ -29,7 +30,7 @@ var time = new Date().getTime(),
//-----------------------------------------------------------------------------------------------
-// 对外只有一个方法, 和代理提供的方法
+// 对外只有一个方法, 和代理提供的call jsonrpc方法
//-----------------------------------------------------------------------------------------------
Invoker.prototype.call = function (methodName) {
var desc = this.invokerDesc,
@@ -44,6 +45,21 @@ Invoker.prototype.call = function (methodName) {
});
};
+//-----------------------------------------------------------------------------------------------
+// 对外只有一个方法, 和代理提供的callrpc方法
+//-----------------------------------------------------------------------------------------------
+Invoker.prototype.callRpc = function (methodName) {
+ console.log("request callRpc methodName [" + methodName + "]");
+ var desc = this.invokerDesc,
+ service = desc.serviceName,
+ methodArgs = _.toArray(arguments).slice(1);
+
+ return Cluster
+ .getProvider(desc)
+ .then(function (serverMap) {
+ return SocketClient.execute(serverMap, service, methodName, desc.getVersion(), methodArgs)
+ });
+};
//-----------------------------------------------------------------------------------------------
//
//
diff --git a/lib/rpc/util/Socket.js b/lib/rpc/util/Socket.js
new file mode 100644
index 0000000..fc7e2cb
--- /dev/null
+++ b/lib/rpc/util/Socket.js
@@ -0,0 +1,233 @@
+//引入hession.js
+const hessian = require('hessian.js');
+const net = require('net'); //网络库
+const url = require('url');
+const qs = require('querystring');
+const Q = require('q');
+const DEFAULT_LEN = 8388608; // 8 * 1024 * 1024 //默认参数buffer长度
+
+const ERROR = {
+ '100': 'create buffer failed ',
+ '102': '连接服务失败 ',
+ '103': '服务不可用',
+ '104': '未找到对应的method',
+ '105': 'socket连接错误',
+ '106': '远程服务无响应，请重试',
+ '107': '该服务未返回任何数据',
+ '108': '请求参数不合法或类型不一致',
+ '109': 'hessian decoder error ',
+ '110': 'socket已经关闭',
+};
+
+var SocketClient = function() {
+ this._dubboVersion = '2.5.3';
+ this._group = '';
+};
+
+SocketClient.prototype.execute = function(server, service, method, version, args) {
+ console.log("server: " + JSON.stringify(server) + " service = " + service + " method = " + method + " args = " + JSON.stringify(args));
+ var q = Q.defer();
+ var _this = this,
+ buffer;
+ try {
+ buffer = _this._createBuffer(service, method, version, args);
+ } catch (e) {
+ q.reject({
+ code: '100',
+ error: e.message || ERROR['100']
+ });
+ }
+
+ _this._fetchData(_this.serverIpAndPort(server), buffer, method)
+ .then(function(data) {
+ q.resolve(data);
+ })
+ .catch(function(err) {
+ q.reject(err);
+ });
+ return q.promise;
+};
+
+SocketClient.prototype.serverIpAndPort = function(server) {
+ var ips = server.split(":");
+ return {
+ host: ips[0],
+ port: ips[1]
+ };
+};
+
+SocketClient.prototype._fetchData = function(zoo, buffer, method) {
+ var q = Q.defer();
+
+ var bl = 16;
+ var host = zoo.host;
+ var port = zoo.port;
+ var ret = null;
+ var chunks = [];
+ var tryCount = 0;
+ var heap, ret;
+
+ var client = new net.Socket();
+
+ //connect
+ client.connect(port, host, function() {
+ client.write(buffer);
+ });
+
+ //发送数据
+ client.on('data', function(chunk) {
+
+ if (!chunks.length) {
+ var arr = Array.prototype.slice.call(chunk.slice(0, 16));
+ var i = 0;
+ while (i < 3) {
+ bl += arr.pop() * Math.pow(255, i++);
+ }
+ }
+
+ chunks.push(chunk);
+ heap = Buffer.concat(chunks);
+ (heap.length >= bl) && client.destroy();
+ });
+
+ // 105 socket connection error
+ client.on('error', function(err) {
+ client.destroy();
+ q.reject({
+ code: '105',
+ error: ERROR['105'] + (err.message || err)
+ });
+ });
+
+ // 110 socket closed
+ client.on('close', function(err) {
+
+ if (err) {
+ q.reject({
+ code: '110',
+ error: ERROR['110'] + (err.message || '')
+ });
+ }
+ // 106 service response error
+ if (heap[3] !== 20) {
+ ret = heap.slice(18, heap.length - 1).toString();
+ q.reject({
+ code: '106',
+ error: ERROR['106'] + ret
+ });
+ }
+
+ // 107 service void return
+ if (heap[3] === 20 && heap[15] === 1) {
+ q.resolve(true);
+ }
+
+ try {
+ var offset = heap[16] === 145 ? 17 : 18;
+ var buf = new hessian.DecoderV2(heap.slice(offset, heap.length));
+ var _ret = buf.read();
+ if (_ret instanceof Error || offset === 18) {
+ q.reject({
+ code: '108',
+ error: ERROR['108'] + (_ret.message || _ret)
+ }); //108 hessian read error
+ }
+ q.resolve(_ret);
+ } catch (e) {
+ q.reject({
+ code: '109',
+ error: ERROR['109'] + (e.message || e)
+ }); //109 hessian decoder error
+ }
+
+ });
+
+ client.on('timeout', function() {
+ client.destroy();
+ console.log('socket timeout');
+ });
+
+ return q.promise;
+};
+
+SocketClient.prototype._createBuffer = function(service, method, version, args) {
+ var typeRef, types, type, buffer;
+
+ typeRef = {
+ boolean: 'Z',
+ int: 'I',
+ short: 'S',
+ long: 'J',
+ double: 'D',
+ float: 'F'
+ };
+
+ if (args && args.length) {
+ for (var i = 0, l = args.length; i < l; i++) {
+ type = args[i]['$class'];
+ types += type && ~type.indexOf('.') ? 'L' + type.replace(/\./gi, '/') + ';' : typeRef[type];
+ }
+ buffer = this.buffer(service, method, version, types, args);
+ } else {
+ buffer = this.buffer(method, '');
+ }
+
+ return buffer;
+};
+
+SocketClient.prototype.buffer = function(service, method, version, type, args) {
+ var bufferBody = this.bufferBody(service, method, version, type, args);
+ var bufferHead = this.bufferHead(bufferBody.length);
+ return Buffer.concat([bufferHead, bufferBody]);
+};
+
+SocketClient.prototype.bufferHead = function(length) {
+ var head = [0xda, 0xbb, 0xc2, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0];
+ var i = 15;
+
+ if (length > DEFAULT_LEN) {
+ throw new Error(`Data length too large: ${length}, max payload: ${DEFAULT_LEN}`);
+ }
+ // 构造body长度信息
+ if (length - 256 < 0) {
+ head.splice(i, 1, length - 256);
+ } else {
+ while (length - 256 > 0) {
+ head.splice(i--, 1, length % 256);
+ length = length >> 8;
+ }
+ head.splice(i, 1, length);
+ }
+ return new Buffer(head);
+};
+
+SocketClient.prototype.bufferBody = function(service, method, version, type, args) {
+ var encoder = new hessian.EncoderV2();
+ encoder.write(this._dubboVersion);
+ encoder.write(service);
+ encoder.write(version);
+ encoder.write(method);
+ encoder.write(type);
+
+ if (args && args.length) {
+ for (var i = 0, len = args.length; i < len; ++i) {
+ encoder.write(args[i]);
+ }
+ }
+
+ encoder.write({
+ $class: 'java.util.HashMap',
+ $: {
+ interface: service,
+ version: version,
+ group: this._group,
+ path: service,
+ timeout: '60000'
+ }
+ });
+
+ encoder = encoder.byteBuffer._bytes.slice(0, encoder.byteBuffer._offset);
+ return encoder;
+};
+
+module.exports = new SocketClient();
diff --git a/package.json b/package.json
index 4f28664..b022362 100644
--- a/package.json
+++ b/package.json
@@ -20,6 +20,7 @@
"node-zookeeper-client":"0.2.2",
"q": "*",
"underscore": "*",
- "request": "*"
+ "request": "*",
+ "hessian.js": "^2.1.8"
}
}
diff --git a/test/dubbo.config.js b/test/dubbo.config.js
index 2b33b9f..174de43 100644
--- a/test/dubbo.config.js
+++ b/test/dubbo.config.js
@@ -17,7 +17,16 @@ module.exports = {
/**
* 注册中心
*/
- registry: '172.19.65.33:2181',
+ registry: '172.16.150.60:2181',
+
+ /**
+ * dubbo config
+ */
+ dubbo: {
+ providerTimeout: 3,
+ weight: 1,
+ protocol: "dubbo" //添加协议类型，支持rpc调用 默认为jsonrpc, 指定协议则初始化过程中只按照配置协议加载到缓存
+ },
/**
* 负载均衡规则, 目前只有轮询
@@ -27,5 +36,5 @@ module.exports = {
/**
* 懒加载, 用于开发阶段, 快速启动
*/
- lazy: true
+ lazy: false
};
\ No newline at end of file
diff --git a/test/test.js b/test/test.js
index 6d98be1..3898454 100644
--- a/test/test.js
+++ b/test/test.js
@@ -5,19 +5,32 @@ var _ = require('underscore'),
dubboClient.config(require('./dubbo.config.js'));
////获取serivce
-var catQueryProvider = dubboClient.getService('com.qianmi.pc.api.cat.StandardCatQueryProvider', '1.2.3');
+var catQueryProvider = dubboClient.getService('com.xxx.dubbo.coach.api.goods.GoodsSkuService', '1.0');
-//setTimeout(function(){
-// for(var k in catQueryProvider){
-// console.info(k);
-// }
-//}, 1000)
-
-catQueryProvider.call('listByParentId', 111)
+// dubbo协议方式调用
+var data = {
+ "$class": "com.xxx.dubbo.coach.api.goods.request.LessonSkuTypeReq",
+ "$": {"goodsNo": "01241", "coachId": 120, "userId": 105424}
+};
+catQueryProvider.callRpc('goodsSkuDetail', data)
.then(function (r) {
console.info(r);
+ process.exit(0);
})
.catch(function (e) {
console.error(JSON.stringify(e));
+ process.exit(0);
});
+//jsonrpc 方式调用
+// catQueryProvider.call('goodsSkuDetail', 111)
+// .then(function (r) {
+// console.info(r);
+// process.exit(0);
+// })
+// .catch(function (e) {
+// console.error(JSON.stringify(e));
+// process.exit(0);
+// });
+
+