<?php
/*
*
* Copyright 2015 gRPC authors.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*
*/
namespace Grpc;
/**
* Base class for generated client stubs. Stub methods are expected to call
* _simpleRequest or _streamRequest and return the result.
*/
class BaseStub
{
private $hostname;
private $hostname_override;
private $channel;
private $call_invoker;
// a callback function
private $update_metadata;
/**
* @param string $hostname
* @param array $opts
* - 'update_metadata': (optional) a callback function which takes in a
* metadata array, and returns an updated metadata array
* - 'grpc.primary_user_agent': (optional) a user-agent string
* @param Channel|InterceptorChannel $channel An already created Channel or InterceptorChannel object (optional)
*/
public function __construct($hostname, $opts, $channel = null)
{
if (!method_exists('ChannelCredentials', 'isDefaultRootsPemSet') ||
!ChannelCredentials::isDefaultRootsPemSet()) {
$ssl_roots = file_get_contents(
dirname(__FILE__).'/../../etc/roots.pem'
);
ChannelCredentials::setDefaultRootsPem($ssl_roots);
}
$this->hostname = $hostname;
$this->update_metadata = null;
if (isset($opts['update_metadata'])) {
if (is_callable($opts['update_metadata'])) {
$this->update_metadata = $opts['update_metadata'];
}
unset($opts['update_metadata']);
}
if (!empty($opts['grpc.ssl_target_name_override'])) {
$this->hostname_override = $opts['grpc.ssl_target_name_override'];
}
if (isset($opts['grpc_call_invoker'])) {
$this->call_invoker = $opts['grpc_call_invoker'];
unset($opts['grpc_call_invoker']);
$channel_opts = $this->updateOpts($opts);
// If the grpc_call_invoker is defined, use the channel created by the call invoker.
$this->channel = $this->call_invoker->createChannelFactory($hostname, $channel_opts);
return;
}
$this->call_invoker = new DefaultCallInvoker();
if ($channel) {
if (!is_a($channel, 'Grpc\Channel') &&
!is_a($channel, 'Grpc\Internal\InterceptorChannel')) {
throw new \Exception('The channel argument is not a Channel object '.
'or an InterceptorChannel object created by '.
'Interceptor::intercept($channel, Interceptor|Interceptor[] $interceptors)');
}
$this->channel = $channel;
return;
}
$this->channel = static::getDefaultChannel($hostname, $opts);
}
private static function updateOpts($opts) {
if (!empty($opts['grpc.primary_user_agent'])) {
$opts['grpc.primary_user_agent'] .= ' ';
} else {
$opts['grpc.primary_user_agent'] = '';
}
if (defined('\Grpc\VERSION')) {
$version_str = \Grpc\VERSION;
} else {
if (!file_exists($composerFile = __DIR__.'/../../composer.json')) {
// for grpc/grpc-php subpackage
$composerFile = __DIR__.'/../composer.json';
}
$package_config = json_decode(file_get_contents($composerFile), true);
$version_str = $package_config['version'];
}
$opts['grpc.primary_user_agent'] .= 'grpc-php/'.$version_str;
if (!array_key_exists('credentials', $opts)) {
throw new \Exception("The opts['credentials'] key is now ".
'required. Please see one of the '.
'ChannelCredentials::create methods');
}
return $opts;
}
/**
* Creates and returns the default Channel
*
* @param array $opts Channel constructor options
*
* @return Channel The channel
*/
public static function getDefaultChannel($hostname, array $opts)
{
$channel_opts = self::updateOpts($opts);
return new Channel($hostname, $opts);
}
/**
* @return string The URI of the endpoint
*/
public function getTarget()
{
return $this->channel->getTarget();
}
/**
* @param bool $try_to_connect (optional)
*
* @return int The grpc connectivity state
*/
public function getConnectivityState($try_to_connect = false)
{
return $this->channel->getConnectivityState($try_to_connect);
}
/**
* @param int $timeout in microseconds
*
* @return bool true if channel is ready
* @throw Exception if channel is in FATAL_ERROR state
*/
public function waitForReady($timeout)
{
$new_state = $this->getConnectivityState(true);
if ($this->_checkConnectivityState($new_state)) {
return true;
}
$now = Timeval::now();
$delta = new Timeval($timeout);
$deadline = $now->add($delta);
while ($this->channel->watchConnectivityState($new_state, $deadline)) {
// state has changed before deadline
$new_state = $this->getConnectivityState();
if ($this->_checkConnectivityState($new_state)) {
return true;
}
}
// deadline has passed
$new_state = $this->getConnectivityState();
return $this->_checkConnectivityState($new_state);
}
/**
* Close the communication channel associated with this stub.
*/
public function close()
{
$this->channel->close();
}
/**
* @param $new_state Connect state
*
* @return bool true if state is CHANNEL_READY
* @throw Exception if state is CHANNEL_FATAL_FAILURE
*/
private function _checkConnectivityState($new_state)
{
if ($new_state == \Grpc\CHANNEL_READY) {
return true;
}
if ($new_state == \Grpc\CHANNEL_FATAL_FAILURE) {
throw new \Exception('Failed to connect to server');
}
return false;
}
/**
* constructs the auth uri for the jwt.
*
* @param string $method The method string
*
* @return string The URL string
*/
private function _get_jwt_aud_uri($method)
{
// TODO(jtattermusch): This is not the correct implementation
// of extracting JWT "aud" claim. We should rely on
// grpc_metadata_credentials_plugin which
// also provides the correct value of "aud" claim
// in the grpc_auth_metadata_context.service_url field.
// Trying to do the construction of "aud" field ourselves
// is bad.
$last_slash_idx = strrpos($method, '/');
if ($last_slash_idx === false) {
throw new \InvalidArgumentException(
'service name must have a slash'
);
}
$service_name = substr($method, 0, $last_slash_idx);
if ($this->hostname_override) {
$hostname = $this->hostname_override;
} else {
$hostname = $this->hostname;
}
// Remove the port if it is 443
// See https://github.com/grpc/grpc/blob/07c9f7a36b2a0d34fcffebc85649cf3b8c339b5d/src/core/lib/security/transport/client_auth_filter.cc#L205
if ((strlen($hostname) > 4) && (substr($hostname, -4) === ":443")) {
$hostname = substr($hostname, 0, -4);
}
return 'https://'.$hostname.$service_name;
}
/**
* validate and normalize the metadata array.
*
* @param array $metadata The metadata map
*
* @return array $metadata Validated and key-normalized metadata map
* @throw InvalidArgumentException if key contains invalid characters
*/
private function _validate_and_normalize_metadata($metadata)
{
$metadata_copy = [];
foreach ($metadata as $key => $value) {
if (!preg_match('/^[.A-Za-z\d_-]+$/', $key)) {
throw new \InvalidArgumentException(
'Metadata keys must be nonempty strings containing only '.
'alphanumeric characters, hyphens, underscores and dots'
);
}
$metadata_copy[strtolower($key)] = $value;
}
return $metadata_copy;
}
/**
* Create a function which can be used to create UnaryCall
*
* @param Channel|InterceptorChannel $channel
* @param callable $deserialize A function that deserializes the response
*
* @return \Closure
*/
private function _GrpcUnaryUnary($channel)
{
return function ($method,
$argument,
$deserialize,
array $metadata = [],
array $options = []) use ($channel) {
$call = $this->call_invoker->UnaryCall(
$channel,
$method,
$deserialize,
$options
);
$jwt_aud_uri = $this->_get_jwt_aud_uri($method);
if (is_callable($this->update_metadata)) {
$metadata = call_user_func(
$this->update_metadata,
$metadata,
$jwt_aud_uri
);
}
$metadata = $this->_validate_and_normalize_metadata(
$metadata
);
$call->start($argument, $metadata, $options);
return $call;
};
}
/**
* Create a function which can be used to create ServerStreamingCall
*
* @param Channel|InterceptorChannel $channel
* @param callable $deserialize A function that deserializes the response
*
* @return \Closure
*/
private function _GrpcStreamUnary($channel)
{
return function ($method,
$deserialize,
array $metadata = [],
array $options = []) use ($channel) {
$call = $this->call_invoker->ClientStreamingCall(
$channel,
$method,
$deserialize,
$options
);
$jwt_aud_uri = $this->_get_jwt_aud_uri($method);
if (is_callable($this->update_metadata)) {
$metadata = call_user_func(
$this->update_metadata,
$metadata,
$jwt_aud_uri
);
}
$metadata = $this->_validate_and_normalize_metadata(
$metadata
);
$call->start($metadata);
return $call;
};
}
/**
* Create a function which can be used to create ClientStreamingCall
*
* @param Channel|InterceptorChannel $channel
* @param callable $deserialize A function that deserializes the response
*
* @return \Closure
*/
private function _GrpcUnaryStream($channel)
{
return function ($method,
$argument,
$deserialize,
array $metadata = [],
array $options = []) use ($channel) {
$call = $this->call_invoker->ServerStreamingCall(
$channel,
$method,
$deserialize,
$options
);
$jwt_aud_uri = $this->_get_jwt_aud_uri($method);
if (is_callable($this->update_metadata)) {
$metadata = call_user_func(
$this->update_metadata,
$metadata,
$jwt_aud_uri
);
}
$metadata = $this->_validate_and_normalize_metadata(
$metadata
);
$call->start($argument, $metadata, $options);
return $call;
};
}
/**
* Create a function which can be used to create BidiStreamingCall
*
* @param Channel|InterceptorChannel $channel
* @param callable $deserialize A function that deserializes the response
*
* @return \Closure
*/
private function _GrpcStreamStream($channel)
{
return function ($method,
$deserialize,
array $metadata = [],
array $options = []) use ($channel) {
$call = $this->call_invoker->BidiStreamingCall(
$channel,
$method,
$deserialize,
$options
);
$jwt_aud_uri = $this->_get_jwt_aud_uri($method);
if (is_callable($this->update_metadata)) {
$metadata = call_user_func(
$this->update_metadata,
$metadata,
$jwt_aud_uri
);
}
$metadata = $this->_validate_and_normalize_metadata(
$metadata
);
$call->start($metadata);
return $call;
};
}
/**
* Create a function which can be used to create UnaryCall
*
* @param Channel|InterceptorChannel $channel
* @param callable $deserialize A function that deserializes the response
*
* @return \Closure
*/
private function _UnaryUnaryCallFactory($channel)
{
if (is_a($channel, 'Grpc\Internal\InterceptorChannel')) {
return function ($method,
$argument,
$deserialize,
array $metadata = [],
array $options = []) use ($channel) {
return $channel->getInterceptor()->interceptUnaryUnary(
$method,
$argument,
$deserialize,
$this->_UnaryUnaryCallFactory($channel->getNext()),
$metadata,
$options
);
};
}
return $this->_GrpcUnaryUnary($channel);
}
/**
* Create a function which can be used to create ServerStreamingCall
*
* @param Channel|InterceptorChannel $channel
* @param callable $deserialize A function that deserializes the response
*
* @return \Closure
*/
private function _UnaryStreamCallFactory($channel)
{
if (is_a($channel, 'Grpc\Internal\InterceptorChannel')) {
return function ($method,
$argument,
$deserialize,
array $metadata = [],
array $options = []) use ($channel) {
return $channel->getInterceptor()->interceptUnaryStream(
$method,
$argument,
$deserialize,
$this->_UnaryStreamCallFactory($channel->getNext()),
$metadata,
$options
);
};
}
return $this->_GrpcUnaryStream($channel);
}
/**
* Create a function which can be used to create ClientStreamingCall
*
* @param Channel|InterceptorChannel $channel
* @param callable $deserialize A function that deserializes the response
*
* @return \Closure
*/
private function _StreamUnaryCallFactory($channel)
{
if (is_a($channel, 'Grpc\Internal\InterceptorChannel')) {
return function ($method,
$deserialize,
array $metadata = [],
array $options = []) use ($channel) {
return $channel->getInterceptor()->interceptStreamUnary(
$method,
$deserialize,
$this->_StreamUnaryCallFactory($channel->getNext()),
$metadata,
$options
);
};
}
return $this->_GrpcStreamUnary($channel);
}
/**
* Create a function which can be used to create BidiStreamingCall
*
* @param Channel|InterceptorChannel $channel
* @param callable $deserialize A function that deserializes the response
*
* @return \Closure
*/
private function _StreamStreamCallFactory($channel)
{
if (is_a($channel, 'Grpc\Internal\InterceptorChannel')) {
return function ($method,
$deserialize,
array $metadata = [],
array $options = []) use ($channel) {
return $channel->getInterceptor()->interceptStreamStream(
$method,
$deserialize,
$this->_StreamStreamCallFactory($channel->getNext()),
$metadata,
$options
);
};
}
return $this->_GrpcStreamStream($channel);
}
/* This class is intended to be subclassed by generated code, so
* all functions begin with "_" to avoid name collisions. */
/**
* Call a remote method that takes a single argument and has a
* single output.
*
* @param string $method The name of the method to call
* @param mixed $argument The argument to the method
* @param callable $deserialize A function that deserializes the response
* @param array $metadata A metadata map to send to the server
* (optional)
* @param array $options An array of options (optional)
*
* @return UnaryCall The active call object
*/
protected function _simpleRequest(
$method,
$argument,
$deserialize,
array $metadata = [],
array $options = []
) {
$call_factory = $this->_UnaryUnaryCallFactory($this->channel);
$call = $call_factory($method, $argument, $deserialize, $metadata, $options);
return $call;
}
/**
* Call a remote method that takes a stream of arguments and has a single
* output.
*
* @param string $method The name of the method to call
* @param callable $deserialize A function that deserializes the response
* @param array $metadata A metadata map to send to the server
* (optional)
* @param array $options An array of options (optional)
*
* @return ClientStreamingCall The active call object
*/
protected function _clientStreamRequest(
$method,
$deserialize,
array $metadata = [],
array $options = []
) {
$call_factory = $this->_StreamUnaryCallFactory($this->channel);
$call = $call_factory($method, $deserialize, $metadata, $options);
return $call;
}
/**
* Call a remote method that takes a single argument and returns a stream
* of responses.
*
* @param string $method The name of the method to call
* @param mixed $argument The argument to the method
* @param callable $deserialize A function that deserializes the responses
* @param array $metadata A metadata map to send to the server
* (optional)
* @param array $options An array of options (optional)
*
* @return ServerStreamingCall The active call object
*/
protected function _serverStreamRequest(
$method,
$argument,
$deserialize,
array $metadata = [],
array $options = []
) {
$call_factory = $this->_UnaryStreamCallFactory($this->channel);
$call = $call_factory($method, $argument, $deserialize, $metadata, $options);
return $call;
}
/**
* Call a remote method with messages streaming in both directions.
*
* @param string $method The name of the method to call
* @param callable $deserialize A function that deserializes the responses
* @param array $metadata A metadata map to send to the server
* (optional)
* @param array $options An array of options (optional)
*
* @return BidiStreamingCall The active call object
*/
protected function _bidiRequest(
$method,
$deserialize,
array $metadata = [],
array $options = []
) {
$call_factory = $this->_StreamStreamCallFactory($this->channel);
$call = $call_factory($method, $deserialize, $metadata, $options);
return $call;
}
}