blob: 6246266d81977f1c6e3288f1856f0dbc90e3af3e [file] [log] [blame]
// Copyright 2015 The Vanadium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.
/**
* @fileoverview Client for the vanadium service.
*
* Usage:
* var cl = new client(proxyConnection);
* var service = cl.bindTo('EndpointAddress', 'ServiceName');
* resultPromise = service.MethodName(arg);
* @private
*/
var actions = require('../verror/actions');
var byteUtil = require('../vdl/byte-util');
var Controller =
require('../gen-vdl/v.io/x/ref/services/wspr/internal/app').Controller;
var context = require('../context');
var Deferred = require('../lib/deferred');
var emitStreamError = require('../lib/emit-stream-error');
var Incoming = require('../proxy/message-type').Incoming;
var makeError = require('../verror/make-errors');
var Outgoing = require('../proxy/message-type').Outgoing;
var Promise = require('../lib/promise');
var ReservedSignature =
require('../gen-vdl/v.io/v23/rpc').ReservedSignature.val;
var RpcCallOption =
require('../gen-vdl/v.io/x/ref/services/wspr/internal/app').RpcCallOption;
var RpcRequest =
require('../gen-vdl/v.io/x/ref/services/wspr/internal/app').RpcRequest;
var Stream = require('../proxy/stream');
var time = require('../gen-vdl/v.io/v23/vdlroot/time');
var uncapitalize = require('../vdl/util').uncapitalize;
var unwrap = require('../vdl/type-util').unwrap;
var vdl = require('../vdl');
var verror = require('../gen-vdl/v.io/v23/verror');
var vlog = require('../lib/vlog');
var SharedContextKeys = require('../runtime/shared-context-keys');
var vtrace = require('../vtrace');
var Blessings = require('../security/blessings');
var BlessingsId =
require('../gen-vdl/v.io/x/ref/services/wspr/internal/principal').BlessingsId;
var ByteStreamMessageReader = require('../vom/byte-stream-message-reader');
var ByteStreamMessageWriter = require('../vom/byte-stream-message-writer');
var Encoder = require('../vom/encoder');
var Decoder = require('../vom/decoder');
var TaskSequence = require('../lib/task-sequence');
var runtimeFromContext = require('../runtime/runtime-from-context');
var vom = require('../vom');
var OutstandingRPC = function(ctx, options, cb) {
this._ctx = ctx;
this._controller = ctx.value(SharedContextKeys.RUNTIME)._controller;
this._proxy = options.proxy;
this._id = -1;
this._name = options.name;
this._methodName = options.methodName,
this._args = options.args;
this._outArgTypes = options.outArgTypes;
this._numOutParams = options.numOutParams;
this._isStreaming = options.isStreaming || false;
this._inStreamingType = options.inStreamingType;
this._outStreamingType = options.outStreamingType;
this._callOptions = options.callOptions;
this._cb = cb;
this._encoder = options.encoder;
this._decoder = options.decoder;
this._def = null;
this._tasks = new TaskSequence();
};
// Helper function to convert an out argument to the given type.
function convertOutArg(ctx, arg, type, controller) {
if (arg instanceof BlessingsId) {
var runtime = runtimeFromContext(ctx);
return runtime.blessingsManager.blessingsFromId(arg)
.then(function(blessings) {
if (blessings) {
blessings.retain();
}
return blessings;
});
}
// There's no protection against bad out args if it's a JSValue.
// Otherwise, convert to the out arg type to ensure type correctness.
if (!type.equals(vdl.types.JSVALUE)) {
try {
return Promise.resolve(unwrap(vdl.canonicalize.reduce(arg, type)));
} catch(err) {
return Promise.reject(err);
}
}
return Promise.resolve(unwrap(arg));
}
OutstandingRPC.prototype.start = function() {
this._id = this._proxy.nextId();
var ctx = this._ctx;
var self = this;
var cb;
var outArgTypes = this._outArgTypes;
if (this._cb) {
// Wrap the callback to call with multiple arguments cb(err, a, b, c)
// rather than cb(err, [a, b, c]).
var origCb = this._cb;
cb = function convertToMultiArgs(err, results) { // jshint ignore:line
// If called from a deferred, the results are undefined.
if (err) {
origCb(err);
return;
}
// Each out argument should also be unwrapped. (results was []any)
results = results || [];
var resultPromises = results.map(function(res, i) {
return convertOutArg(ctx, res, outArgTypes[i], self._controller);
});
Promise.all(resultPromises)
.then(function(results) {
results.unshift(null);
origCb.apply(null, results);
}).catch(origCb);
};
}
var def = new Deferred(cb);
if (!this._cb) {
// If we are using a promise, strip single args out of the arg array.
// e.g. [ arg1 ] -> arg1
def.promise = def.promise.then(function(args) {
if (!Array.isArray(args)) {
throw new verror.InternalError(ctx,
'Internal error: incorrectly formatted out args in client');
}
// Each out argument should also be unwrapped. (args was []any)
var unwrappedArgPromises = args.map(function(outArg, i) {
return convertOutArg(ctx, outArg, outArgTypes[i], self._controller);
});
return Promise.all(unwrappedArgPromises).then(function(unwrappedArgs) {
// We expect:
// 0 args - return; // NOT return [];
// 1 args - return a; // NOT return [a];
// 2 args - return [a, b] ;
//
// Convert the results from array style to the expected return style.
// undefined, a, [a, b], [a, b, c] etc
switch(unwrappedArgs.length) {
case 0:
return undefined;
case 1:
return unwrappedArgs[0];
default:
return unwrappedArgs;
}
});
});
}
var streamingDeferred = null;
if (this._isStreaming) {
streamingDeferred = new Deferred();
// Clients read data of type outStreamingType and write data of type
// inStreamingType.
def.stream = new Stream(this._id, streamingDeferred.promise, true,
this._outStreamingType, this._inStreamingType);
def.promise.stream = def.stream;
}
var message = this.constructMessage();
this._def = def;
this._proxy.cancelFromContext(this._ctx, this._id);
this._proxy.sendRequest(message, Outgoing.REQUEST, this, this._id);
if (streamingDeferred) {
this._proxy.senderPromise.then(function(ws) {
streamingDeferred.resolve(ws);
}, function(err) {
streamingDeferred.reject(err);
});
}
return def.promise;
};
OutstandingRPC.prototype.handleResponse = function(type, data) {
var rpc = this;
switch (type) {
case Incoming.FINAL_RESPONSE:
this._tasks.addTask(function() {
return rpc.handleCompletion(data);
});
break;
case Incoming.STREAM_RESPONSE:
this._tasks.addTask(function() {
return rpc.handleStreamData(data);
});
break;
case Incoming.ERROR_RESPONSE:
this._tasks.addTask(function() {
return rpc.handleError(data);
});
break;
case Incoming.STREAM_CLOSE:
this._tasks.addTask(function() {
return rpc.handleStreamClose();
});
break;
default:
this._tasks.addTask(function() {
return rpc.handleError(
new verror.InternalError(
rpc._ctx, 'Received unknown response type from wspr'));
});
break;
}
};
OutstandingRPC.prototype.handleCompletion = function(data) {
try {
var bytes = byteUtil.hex2Bytes(data);
this._decoder._messageReader.addBytes(bytes);
} catch (e) {
this.handleError(
new verror.InternalError(this._ctx, 'Failed to decode result: ', e));
return Promise.resolve();
}
var rpc = this;
return this._decoder.decode().then(function(response) {
vtrace.getStore(rpc._ctx).merge(response.traceResponse);
vtrace.getSpan(rpc._ctx).finish();
rpc._def.resolve(response.outArgs);
if (rpc._def.stream) {
rpc._def.stream._queueClose();
}
rpc._proxy.dequeue(this._id);
}).catch(function(e) {
rpc.handleError(
new verror.InternalError(rpc._ctx, 'Failed to decode result: ', e));
return;
});
};
OutstandingRPC.prototype.handleStreamData = function(data) {
if (!this._def.stream) {
vlog.logger.warn('Ignoring streaming message for non-streaming flow : ' +
this._id);
return Promise.resolve();
}
try {
data = byteUtil.hex2Bytes(data);
} catch (e) {
this.handleError(
new verror.InternalError(this._ctx, 'Failed to decode result: ', e));
return;
}
var rpc = this;
return vom.decode(data).then(function(data) {
rpc._def.stream._queueRead(data);
}).catch(function(e) {
rpc.handleError(
new verror.InternalError(rpc._ctx, 'Failed to decode result: ', e));
});
};
OutstandingRPC.prototype.handleStreamClose = function() {
if (this._def.stream) {
this._def.stream._queueClose();
}
return Promise.resolve();
};
OutstandingRPC.prototype.handleError = function(err) {
if (this._def.stream) {
emitStreamError(this._def.stream, err);
this._def.stream._queueClose();
}
this._def.reject(err);
this._proxy.dequeue(this._id);
return Promise.resolve();
};
/**
* Construct a message to send to the vanadium native code
* @private
* @return {string} json string to send to jspr
*/
OutstandingRPC.prototype.constructMessage = function() {
var deadline = this._ctx.deadline();
var timeout = new time.WireDeadline();
if (deadline !== null) {
var millis = deadline - Date.now();
var seconds = Math.floor(millis / 1000);
timeout.fromNow = new time.Duration({
seconds: seconds,
nanos: (millis - seconds * 1000) * 1000000
});
} else {
timeout.noDeadline = true;
}
var language = this._ctx.value(SharedContextKeys.LANG_KEY) || '';
var jsonMessage = {
name: this._name,
method: this._methodName,
numInArgs: this._args.length,
// TODO(bprosnitz) Is || 0 needed?
numOutArgs: this._numOutParams || 0,
isStreaming: this._isStreaming,
traceRequest: vtrace.request(this._ctx),
deadline: timeout,
callOptions: this._callOptions,
context: {
language: language,
}
};
var header = new RpcRequest(jsonMessage);
var encoder = this._encoder;
if (!encoder._messageWriter) {
encoder._messageWriter = new ByteStreamMessageWriter();
}
encoder.encode(header);
for (var i = 0; i < this._args.length; i++) {
var o = this._args[i];
if (o instanceof Blessings) {
o = o.convertToJsBlessings();
}
encoder.encode(o);
}
return byteUtil.bytes2Hex(encoder._messageWriter.consumeBytes());
};
/**
* @summary Client represents the interface for making RPC calls.
* There may be multiple outstanding Calls associated with a single Client.
*
* @description
* <p>Private Constructor, use
* [Runtime#newClient]{@link module:vanadium~Runtime#newClient}</p>
* @inner
* @constructor
* @memberof module:vanadium.rpc
*/
function Client(proxyConnection) {
if (!(this instanceof Client)) {
return new Client(proxyConnection);
}
this._proxyConnection = proxyConnection;
if (proxyConnection && proxyConnection.clientEncoder) {
this._encoder = proxyConnection.clientEncoder;
}
if (proxyConnection && proxyConnection.clientDecoder) {
this._decoder = proxyConnection.clientDecoder;
}
this._controller = this.bindWithSignature(
'__controller', [Controller.prototype._serviceDescription]);
}
// TODO(bprosnitz) v.io/core/javascript.IncorrectArgCount.
var IncorrectArgCount = makeError(
'v.io/core/javascript.IncorrectArgCount',
actions.NO_RETRY,
'{1:}{2:} Client RPC call {3}({4}) had an incorrect number of ' +
'arguments. Expected format: {5}({6})');
/**
* A callback that is called when
* [bindTo]{@link module:vanadium.rpc~Client#bindTo} finishes.
* @callback module:vanadium.rpc~Client~bindToCb
* @param {Error} err If set the error that occurred.
* @param {object} service The stub object containing the exported
* methods of the remote service.
*/
/**
* <p>Performs client side binding of a remote service to a native JavaScript
* stub object.</p>
*
* Usage:
* <pre>
* client.bindTo(context, 'Service/Name').then(function(service) {
* service.fooMethod(fooArgs).then(function(methodCallResult) {
* // Do stuff with results.
* }).catch(function(err) {
* // Calling fooMethod failed.
* });
* }).catch(function(err) {
* // Binding to Service/Name failed.
* });
* </pre>
* @param {module:vanadium.context.Context} ctx A context.
* @param {string} name The vanadium name of the service to bind to.
* @param {module:vanadium.rpc~Client~bindToCb} [cb] If given, this function
* will be called on completion of the bind.
* @return {Promise<object>} Promise that resolves to the stub object containing
* the exported methods of the remote service.
*/
Client.prototype.bindTo = function(ctx, name, cb) {
var client = this;
var last = arguments.length - 1;
// grab the callback
if (typeof arguments[last] === 'function') {
cb = arguments[last];
}
var def = new Deferred(cb);
// Require first arg to be a Context
if (! (ctx instanceof context.Context)) {
var err = new Error('First argument must be a Context object.');
def.reject(err);
return def.promise;
}
client.signature(ctx, name).then(function(serviceSignature) {
vlog.logger.debug('Received signature for:', name, serviceSignature);
def.resolve(client.bindWithSignature(name, serviceSignature));
}).catch(function(err) {
def.reject(err);
});
return def.promise;
};
/**
* <p>Performs client side binding of a remote service to a native JavaScript
* stub object when you already have the service signature.</p>
*
* Usage:
* <pre>
* var service = client.bindWithSignature('Service/Name', signature);
* service.fooMethod(fooArgs).then(function(methodCallResult) {
* // Do stuff with results.
* }).catch(function(err) {
* // Calling fooMethod failed.
* });
* </pre>
*
* @param {string} name The vanadium name of the service to bind to.
* @param {module:vanadium.vdl.signature.Interface} signature The service
* signature of a vanadium service.
* @return {object} The stub object containing
* the exported methods of the remote service.
*/
Client.prototype.bindWithSignature = function(name, signature) {
var client = this;
var boundObject = {};
function bindMethod(methodSig) {
var method = uncapitalize(methodSig.name);
boundObject[method] = function(ctx /*, arg1, arg2, ..., callback*/) {
var args = Array.prototype.slice.call(arguments, 0);
var callback;
var err;
// Callback is the last function argument, pull it out of the args
var lastType = typeof args[args.length - 1];
if (lastType === 'function' || lastType === 'undefined') {
callback = args.pop();
}
// Require first arg to be a Context
if (args.length >= 1 && args[0] instanceof context.Context) {
ctx = args.shift();
} else {
err = new Error('First argument must be a Context object.');
if (callback) {
return callback(err);
} else {
return Promise.reject(err);
}
}
// Remove ClientCallOptions from args and build array of callOptions.
var callOptions = [];
args = args.filter(function(arg) {
if (arg instanceof ClientCallOption) {
callOptions = callOptions.concat(
arg._toRpcCallOption(ctx, client._proxyConnection));
return false;
}
return true;
});
ctx = vtrace.withNewSpan(ctx, '<jsclient>"'+name+'".'+method);
if (args.length !== methodSig.inArgs.length) {
var expectedArgs = methodSig.inArgs.map(function(arg) {
return arg.name;
});
// TODO(jasoncampbell): Create an constructor for this error so it
// can be created with less ceremony and checked in a
// programatic way:
//
// service
// .foo('bar')
// .catch(ArgumentsArityError, function(err) {
// console.error('invalid number of arguments')
// })
//
// The given arguments exclude the ctx and (optional) cb.
var givenArgs = Array.prototype.slice.call(arguments, 1);
if (typeof givenArgs[givenArgs.length - 1] === 'function') {
givenArgs.pop();
}
err = new IncorrectArgCount(
ctx,
methodSig.name,
givenArgs,
methodSig.name,
expectedArgs
);
if (callback) {
return callback(err);
} else {
return Promise.reject(err);
}
}
// The inArgs need to be converted to the signature's inArg types.
var canonArgs = new Array(args.length);
try {
for (var i = 0; i < args.length; i++) {
canonArgs[i] = vdl.canonicalize.fill(args[i],
methodSig.inArgs[i].type);
}
} catch(err) {
vlog.logger.error('rpc failed - invalid arg(s)', err);
if (callback) {
return callback(err);
} else {
return Promise.reject(err);
}
}
// The OutstandingRPC needs to know streaming information.
var inStreaming = (typeof methodSig.inStream === 'object' &&
methodSig.inStream !== null);
var outStreaming = (typeof methodSig.outStream === 'object' &&
methodSig.outStream !== null);
var isStreaming = inStreaming || outStreaming;
// The OutstandingRPC needs to know the out arg types.
var outArgTypes = methodSig.outArgs.map(function(outArg) {
return outArg.type;
});
var rpc = new OutstandingRPC(ctx, {
proxy: client._proxyConnection,
name: name,
methodName: methodSig.name,
args: canonArgs,
outArgTypes: outArgTypes,
numOutParams: methodSig.outArgs.length,
isStreaming: isStreaming,
inStreamingType: inStreaming ? methodSig.inStream.type : null,
outStreamingType: outStreaming ? methodSig.outStream.type : null,
callOptions: callOptions,
// If there isn't an encoder or decoder cached, we just use a new one.
// This only really happens in unit tests.
encoder: client._encoder || new Encoder(),
decoder: client._decoder || new Decoder(new ByteStreamMessageReader()),
}, callback);
return rpc.start();
};
}
// Setup the bindings to every method in the service signature list.
signature.forEach(function(sig) {
sig.methods.forEach(function(meth) {
bindMethod(meth);
});
});
Object.defineProperty(boundObject, '__signature', {
value: signature,
writable: false,
});
return boundObject;
};
/**
* A callback that is called with either signature interfaces or an error.
* @callback module:vanadium.rpc~Client~signatureCb
* @param {Error} err If set, the error that occurred.
* @param {module:vanadium.vdl.signature.Interface[]} signature The signature
* interfaces.
*/
/**
* Returns the object signatures for a given object name.
* @param {module:vanadium.context.Context} ctx A context.
* @param {string} name The vanadium name of the service to bind to.
* @param {module:vanadium.rpc~Client~signatureCb} [cb] If given, this
* function will be called on completion.
* @return {Promise<module:vanadium.vdl.signature.Interface[]>} Promise that
* will be resolved with the signature interfaces or rejected with an error
* if there is one.
*/
Client.prototype.signature = function(ctx, name, cb) {
var last = arguments.length - 1;
// grab the callback
if (typeof arguments[last] === 'function') {
cb = arguments[last];
}
var deferred = new Deferred(cb);
if (!(ctx instanceof context.Context)) {
deferred.reject(new Error('First argument must be a Context object.'));
return deferred.promise;
}
var cache = this._proxyConnection.signatureCache;
var cacheEntry = cache.get(name);
if (cacheEntry) {
deferred.resolve(cacheEntry);
return deferred.promise;
}
this._controller.signature(ctx, name).then(function(signature){
cache.set(name, signature);
deferred.resolve(signature);
}).catch(function(err) {
deferred.reject(err);
});
return deferred.promise;
};
/**
* A callback that will be called on completion of the
* [remoteBlessings]{@link module:vanadium.rpc~Client#remoteBlessings}
* function.
* @callback module:vanadium.rpc~Client~remoteBlessingsCb
* @param {Error} err If set, the error that occurred.
* @param {string[]} blessingNames The blessings of the remote server.
*/
/**
* Returns the remote blessings of a server at the given name.
* @param {module:vanadium.context.Context} ctx A context.
* @param {string} name The vanadium name of the service to get the remote
* blessings of.
* @param {string} [method] The name of the rpc method that will be started in
* order to read the blessings. Defaults to 'Signature'. This only matters in
* the case when a server responds to different method calls with different
* blessings.
* @param {module:vanadium.rpc~Client~remoteBlessingsCb} [cb] If given, this
* function will be called on completion.
* @return {Promise<string[]>} Promise that will be resolved with the
* blessing names or rejected with an error if there is one.
*/
Client.prototype.remoteBlessings = function(ctx, name, method, cb) {
var last = arguments.length - 1;
// grab the callback
if (typeof arguments[last] === 'function') {
cb = arguments[last];
}
// method defaults to Signature.
if (typeof method !== 'string') {
method = ReservedSignature;
}
return this._controller.remoteBlessings(ctx, name, method, cb);
};
/**
* @summary Create a ClientCallOption object.
*
* @description <p>Client call options can be passed to a service method and
* are used to configure the RPC call. They are not passed to the Vanadium RPC
* service.</p>
*
* <p>Supported keys are 'allowedServersPolicy' and 'granter'.</p>
*
* <p>Example of allowedServersPolicy option:</p>
* <pre>
* var callOpt = client.callOption({
* allowedServersPolicy: ['alice/home/tv']
* });
* service.get(ctx, 'foo', callOpt, function(err) {
* // err will be non-null if service's blessings do not match
* // ['alice/home/tv'].
* });
* </pre>
*
* <p>Example of granter option:</p>
* <pre>
* var ctx = runtime.getContext();
* var granter = function(ctx, call, callback) {
* // Bless the server's public key with the extension 'ext' and 5 second
* // expiration caveat.
* var expCaveat = caveats.createExpiryCaveat(new Date(Date.now() + 5000));
* runtime.principal.bless(ctx, call.remoteBlessings.publicKey,
* call.localBlessings, 'ext', expCaveat, callback);
* };
*
* var callOpt = client.callOption({
* granter: granter
* });
*
* // Make a call on to the service. Server will be granted blessing.
* service.get(ctx, 'foo', callOpt, cb);
* </pre>
* @param {object} opts Map of call options.
* @param {string[]} opts.allowedServersPolicy <p>Array of blessing patterns
* that the allowed server must match in order for the RPC to be initiated.</p>
* @param {module:vanadium.security~GranterFunction} opts.granter <p>A granter
* function.</p>
* @return {module:vanadium.rpc~Client~ClientCallOption}
*/
Client.prototype.callOption = function(opts) {
// TODO(nlacasse): Support other CallOption types.
var allowedOptions = ['allowedServersPolicy', 'granter'];
// Validate opts.
var keys = Object.keys(opts);
keys.forEach(function(key) {
if (allowedOptions.indexOf(key) < 0) {
throw new verror.BadArgError(null, 'Invalid call option ' + key);
}
});
return new ClientCallOption(opts);
};
/**
* @summary ClientCallOption represents different configurations that can be
* specified when making an RPC call.
* @description
* Private constructor, use
* [client.callOption(opts)]{@link module:vanadium.rpc~Client#callOption}
* to construct an instance.
* @constructor
* @inner
* @memberof module:vanadium.rpc~Client
*/
function ClientCallOption(opts) {
this.opts = opts;
}
/**
* Convert ClientCallOption object to array of RpcCallOption VDL values.
* @private
* @return {Array} Array of RpcCallOption VDL values.
*/
ClientCallOption.prototype._toRpcCallOption = function(ctx, proxy) {
var rpcCallOptions = [];
var keys = Object.keys(this.opts);
keys.forEach(function(key) {
var opt = {};
if (key === 'granter') {
var runtime = ctx.value(SharedContextKeys.RUNTIME);
var granterRouter = runtime._getGranterRouter();
var fn = this.opts[key];
var granterId = granterRouter.addGranter(fn);
opt[key] = granterId;
} else {
opt[key] = this.opts[key];
}
rpcCallOptions.push(new RpcCallOption(opt));
}, this);
return rpcCallOptions;
};
/**
* Export the module
*/
module.exports = Client;