Skip to content

Commit 3a5a790

Browse files
stephenplusplusstephenplusplus
committed
pubsub: convert to grpc
1 parent f69d6ec commit 3a5a790

File tree

12 files changed

+871
-368
lines changed

12 files changed

+871
-368
lines changed

‎lib/common/grpc-service-object.js

+100
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,100 @@
1+
/*!
2+
* Copyright 2015 Google Inc. All Rights Reserved.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
/*!
18+
* @module common/grpcServiceObject
19+
*/
20+
21+
'use strict';
22+
23+
var extend = require('extend');
24+
var nodeutil = require('util');
25+
26+
/**
27+
* @type {module:common/serviceObject}
28+
* @private
29+
*/
30+
var ServiceObject = require('./service-object.js');
31+
32+
/**
33+
* @type {module:common/util}
34+
* @private
35+
*/
36+
var util = require('./util.js');
37+
38+
/**
39+
* GrpcServiceObject is a base class, meant to be inherited from by a service
40+
* object that uses the gRPC protobuf API.
41+
*
42+
* @private
43+
*
44+
* @param {object} config - Configuration object.
45+
*/
46+
function GrpcServiceObject(config) {
47+
ServiceObject.call(this, config);
48+
}
49+
50+
nodeutil.inherits(GrpcServiceObject, ServiceObject);
51+
52+
/**
53+
* Delete the object.
54+
*
55+
* @param {function=} callback - The callback function.
56+
* @param {?error} callback.err - An error returned while making this request.
57+
*/
58+
GrpcServiceObject.prototype.delete = function(callback) {
59+
var protoOpts = this.methods.delete.protoOpts;
60+
var reqOpts = this.methods.delete.reqOpts;
61+
62+
this.request(protoOpts, reqOpts, callback || util.noop);
63+
};
64+
65+
/**
66+
* Get the metadata of this object.
67+
*
68+
* @param {function} callback - The callback function.
69+
* @param {?error} callback.err - An error returned while making this request.
70+
* @param {object} callback.metadata - The metadata for this object.
71+
*/
72+
GrpcServiceObject.prototype.getMetadata = function(callback) {
73+
var protoOpts = this.methods.getMetadata.protoOpts;
74+
var reqOpts = this.methods.getMetadata.reqOpts;
75+
76+
this.request(protoOpts, reqOpts, callback);
77+
};
78+
79+
/**
80+
* Set the metadata for this object.
81+
*
82+
* @param {object} metadata - The metadata to set on this object.
83+
* @param {function=} callback - The callback function.
84+
* @param {?error} callback.err - An error returned while making this request.
85+
*/
86+
GrpcServiceObject.prototype.setMetadata = function(metadata, callback) {
87+
var protoOpts = this.methods.setMetadata.protoOpts;
88+
var reqOpts = extend(true, {}, this.methods.setMetadata.reqOpts, metadata);
89+
90+
this.request(protoOpts, reqOpts, callback || util.noop);
91+
};
92+
93+
/**
94+
* Patch a request to the GrpcService object.
95+
*/
96+
GrpcServiceObject.prototype.request = function(protoOpts, reqOpts, callback) {
97+
this.parent.request(protoOpts, reqOpts, callback);
98+
};
99+
100+
module.exports = GrpcServiceObject;

‎lib/common/grpc-service.js

+244
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,244 @@
1+
/*!
2+
* Copyright 2015 Google Inc. All Rights Reserved.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
/*!
18+
* @module common/grpcService
19+
*/
20+
21+
'use strict';
22+
23+
var camelize = require('camelize');
24+
var googleProtoFiles = require('google-proto-files');
25+
var grpc = require('grpc');
26+
var is = require('is');
27+
var nodeutil = require('util');
28+
var path = require('path');
29+
var snakeize = require('snakeize');
30+
31+
/**
32+
* @type {module:common/service}
33+
* @private
34+
*/
35+
var Service = require('./service.js');
36+
37+
/**
38+
* @const {object} - A map of protobuf codes to HTTP status codes.
39+
* @private
40+
*/
41+
var HTTP_ERROR_CODE_MAP = {
42+
0: {
43+
code: 200,
44+
message: 'OK'
45+
},
46+
47+
1: {
48+
code: 499,
49+
message: 'Client Closed Request'
50+
},
51+
52+
2: {
53+
code: 500,
54+
message: 'Internal Server Error'
55+
},
56+
57+
3: {
58+
code: 400,
59+
message: 'Bad Request'
60+
},
61+
62+
4: {
63+
code: 504,
64+
message: 'Gateway Timeout'
65+
},
66+
67+
5: {
68+
code: 404,
69+
message: 'Not Found'
70+
},
71+
72+
6: {
73+
code: 409,
74+
message: 'Conflict'
75+
},
76+
77+
7: {
78+
code: 403,
79+
message: 'Forbidden'
80+
},
81+
82+
8: {
83+
code: 429,
84+
message: 'Too Many Requests'
85+
},
86+
87+
9: {
88+
code: 412,
89+
message: 'Precondition Failed'
90+
},
91+
92+
10: {
93+
code: 409,
94+
message: 'Conflict'
95+
},
96+
97+
11: {
98+
code: 400,
99+
message: 'Bad Request'
100+
},
101+
102+
12: {
103+
code: 501,
104+
message: 'Not Implemented'
105+
},
106+
107+
13: {
108+
code: 500,
109+
message: 'Internal Server Error'
110+
},
111+
112+
14: {
113+
code: 503,
114+
message: 'Service Unavailable'
115+
},
116+
117+
15: {
118+
code: 500,
119+
message: 'Internal Server Error'
120+
},
121+
122+
16: {
123+
code: 401,
124+
message: 'Unauthorized'
125+
}
126+
};
127+
128+
/**
129+
* Service is a base class, meant to be inherited from by a "service," like
130+
* BigQuery or Storage.
131+
*
132+
* This handles making authenticated requests by exposing a `makeReq_` function.
133+
*
134+
* @param {object} config - Configuration object.
135+
* @param {string} config.baseUrl - The base URL to make API requests to.
136+
* @param {string[]} config.scopes - The scopes required for the request.
137+
* @param {object} options - [Configuration object](#/docs/?method=gcloud).
138+
*/
139+
function GrpcService(config, options) {
140+
Service.call(this, config, options);
141+
142+
var service = config.service;
143+
var apiVersion = config.apiVersion;
144+
var rootDir = googleProtoFiles('..');
145+
146+
this.protoOpts = config.proto;
147+
this.proto = grpc.load({
148+
root: rootDir,
149+
file: path.relative(rootDir, googleProtoFiles[service][apiVersion])
150+
}).google[service][apiVersion];
151+
}
152+
153+
nodeutil.inherits(GrpcService, Service);
154+
155+
/**
156+
* Make an authenticated request with gRPC.
157+
*
158+
* @param {object} protoOpts - The proto options.
159+
* @param {string} protoOpts.service - The service name.
160+
* @param {string} protoOpts.method - The method name.
161+
* @param {number=} protoOpts.timeout - After how many milliseconds should the
162+
* request cancel.
163+
* @param {object} reqOpts - The request options.
164+
* @param {function=} callback - The callback function.
165+
*/
166+
GrpcService.prototype.request = function(protoOpts, reqOpts, callback) {
167+
var self = this;
168+
var proto = this.proto;
169+
170+
if (!this.grpcCredentials) {
171+
// We must establish an authClient to give to grpc.
172+
this.authClient.getCredentials(function(err) {
173+
if (err) {
174+
callback(err);
175+
return;
176+
}
177+
178+
var authClient = self.authClient.authClient;
179+
180+
self.grpcCredentials = grpc.credentials.combineChannelCredentials(
181+
grpc.credentials.createSsl(),
182+
grpc.credentials.createFromGoogleCredential(authClient)
183+
);
184+
185+
self.request.call(self, protoOpts, reqOpts, callback);
186+
});
187+
return;
188+
}
189+
190+
var grpcOpts = {};
191+
192+
if (is.number(protoOpts.timeout)) {
193+
grpcOpts.deadline = new Date(Date.now() + protoOpts.timeout);
194+
}
195+
196+
var service = new proto[protoOpts.service](
197+
this.baseUrl,
198+
this.grpcCredentials
199+
);
200+
201+
// snakeize and camelize are used to transform camelCase request options to
202+
// snake_case. This is what ProtoBuf.js (via gRPC) expects. Similarly, the
203+
// response is in snake_case, which is why we use camelize to return it to
204+
// camelCase. An option will be added to gRPC to allow us to skip this step:
205+
// https://github.com/GoogleCloudPlatform/gcloud-node/pull/1070#discussion_r51285492
206+
service[protoOpts.method](snakeize(reqOpts), function(err, resp) {
207+
if (err) {
208+
if (HTTP_ERROR_CODE_MAP[err.code]) {
209+
var httpError = HTTP_ERROR_CODE_MAP[err.code];
210+
err.code = httpError.code;
211+
}
212+
213+
callback(err);
214+
return;
215+
}
216+
217+
function convertBuffers(data) {
218+
if (is.array(data)) {
219+
return data.map(convertBuffers);
220+
}
221+
222+
if (is.object(data)) {
223+
for (var prop in data) {
224+
var value = data[prop];
225+
226+
// An option will be added to gRPC to expose a setting which will
227+
// replace this function (convertBuffers).
228+
// https://github.com/GoogleCloudPlatform/gcloud-node/pull/1070#discussion_r51285492
229+
if (is.object(value) && value.length) {
230+
data[prop] = new Buffer(value).toString('base64');
231+
} else if (is.object(value)) {
232+
data[prop] = convertBuffers(value);
233+
}
234+
}
235+
}
236+
237+
return data;
238+
}
239+
240+
callback(null, convertBuffers(camelize(resp)));
241+
}, null, grpcOpts);
242+
};
243+
244+
module.exports = GrpcService;

‎lib/common/service-object.js

+1-2
Original file line numberDiff line numberDiff line change
@@ -137,14 +137,13 @@ ServiceObject.prototype.create = function(options, callback) {
137137
*/
138138
ServiceObject.prototype.delete = function(callback) {
139139
var methodConfig = this.methods.delete || {};
140+
callback = callback || util.noop;
140141

141142
var reqOpts = extend({
142143
method: 'DELETE',
143144
uri: ''
144145
}, methodConfig.reqOpts);
145146

146-
callback = callback || util.noop;
147-
148147
// The `request` method may have been overridden to hold any special behavior.
149148
// Ensure we call the original `request` method.
150149
ServiceObject.prototype.request.call(this, reqOpts, function(err, resp) {

0 commit comments

Comments
 (0)