src/php/lib/Grpc/BaseStub.php

   1 <?php
   2 /*
   3  *
   4  * Copyright 2015 gRPC authors.
   5  *
   6  * Licensed under the Apache License, Version 2.0 (the "License");
   7  * you may not use this file except in compliance with the License.
   8  * You may obtain a copy of the License at
   9  *
  10  *     http://www.apache.org/licenses/LICENSE-2.0
  11  *
  12  * Unless required by applicable law or agreed to in writing, software
  13  * distributed under the License is distributed on an "AS IS" BASIS,
  14  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  15  * See the License for the specific language governing permissions and
  16  * limitations under the License.
  17  *
  18  */
  19
  20 namespace Grpc;
  21
  22 /**
  23  * Base class for generated client stubs. Stub methods are expected to call
  24  * _simpleRequest or _streamRequest and return the result.
  25  */
  26 class BaseStub
  27 {
  28     private $hostname;
  29     private $hostname_override;
  30     private $channel;
  31     private $call_invoker;
  32
  33     // a callback function
  34     private $update_metadata;
  35
  36     /**
  37      * @param string  $hostname
  38      * @param array   $opts
  39      *  - 'update_metadata': (optional) a callback function which takes in a
  40      * metadata array, and returns an updated metadata array
  41      *  - 'grpc.primary_user_agent': (optional) a user-agent string
  42      * @param Channel|InterceptorChannel $channel An already created Channel or InterceptorChannel object (optional)
  43      */
  44     public function __construct($hostname, $opts, $channel = null)
  45     {
  46         $ssl_roots = file_get_contents(
  47             dirname(__FILE__).'/../../../../etc/roots.pem'
  48         );
  49         ChannelCredentials::setDefaultRootsPem($ssl_roots);
  50
  51         $this->hostname = $hostname;
  52         $this->update_metadata = null;
  53         if (isset($opts['update_metadata'])) {
  54             if (is_callable($opts['update_metadata'])) {
  55                 $this->update_metadata = $opts['update_metadata'];
  56             }
  57             unset($opts['update_metadata']);
  58         }
  59         if (!empty($opts['grpc.ssl_target_name_override'])) {
  60             $this->hostname_override = $opts['grpc.ssl_target_name_override'];
  61         }
  62         if (isset($opts['grpc_call_invoker'])) {
  63             $this->call_invoker = $opts['grpc_call_invoker'];
  64             unset($opts['grpc_call_invoker']);
  65             $channel_opts = $this->updateOpts($opts);
  66             // If the grpc_call_invoker is defined, use the channel created by the call invoker.
  67             $this->channel = $this->call_invoker->createChannelFactory($hostname, $channel_opts);
  68             return;
  69         }
  70         $this->call_invoker = new DefaultCallInvoker();
  71         if ($channel) {
  72             if (!is_a($channel, 'Grpc\Channel') &&
  73                 !is_a($channel, 'Grpc\Internal\InterceptorChannel')) {
  74                 throw new \Exception('The channel argument is not a Channel object '.
  75                     'or an InterceptorChannel object created by '.
  76                     'Interceptor::intercept($channel, Interceptor|Interceptor[] $interceptors)');
  77             }
  78             $this->channel = $channel;
  79             return;
  80         }
  81
  82         $this->channel = static::getDefaultChannel($hostname, $opts);
  83     }
  84
  85     private static function updateOpts($opts) {
  86         if (!file_exists($composerFile = __DIR__.'/../../composer.json')) {
  87             // for grpc/grpc-php subpackage
  88             $composerFile = __DIR__.'/../composer.json';
  89         }
  90         $package_config = json_decode(file_get_contents($composerFile), true);
  91         if (!empty($opts['grpc.primary_user_agent'])) {
  92             $opts['grpc.primary_user_agent'] .= ' ';
  93         } else {
  94             $opts['grpc.primary_user_agent'] = '';
  95         }
  96         $opts['grpc.primary_user_agent'] .=
  97             'grpc-php/'.$package_config['version'];
  98         if (!array_key_exists('credentials', $opts)) {
  99             throw new \Exception("The opts['credentials'] key is now ".
 100                 'required. Please see one of the '.
 101                 'ChannelCredentials::create methods');
 102         }
 103         return $opts;
 104     }
 105
 106     /**
 107      * Creates and returns the default Channel
 108      *
 109      * @param array $opts Channel constructor options
 110      *
 111      * @return Channel The channel
 112      */
 113     public static function getDefaultChannel($hostname, array $opts)
 114     {
 115         $channel_opts = self::updateOpts($opts);
 116         return new Channel($hostname, $opts);
 117     }
 118
 119     /**
 120      * @return string The URI of the endpoint
 121      */
 122     public function getTarget()
 123     {
 124         return $this->channel->getTarget();
 125     }
 126
 127     /**
 128      * @param bool $try_to_connect (optional)
 129      *
 130      * @return int The grpc connectivity state
 131      */
 132     public function getConnectivityState($try_to_connect = false)
 133     {
 134         return $this->channel->getConnectivityState($try_to_connect);
 135     }
 136
 137     /**
 138      * @param int $timeout in microseconds
 139      *
 140      * @return bool true if channel is ready
 141      * @throw Exception if channel is in FATAL_ERROR state
 142      */
 143     public function waitForReady($timeout)
 144     {
 145         $new_state = $this->getConnectivityState(true);
 146         if ($this->_checkConnectivityState($new_state)) {
 147             return true;
 148         }
 149
 150         $now = Timeval::now();
 151         $delta = new Timeval($timeout);
 152         $deadline = $now->add($delta);
 153
 154         while ($this->channel->watchConnectivityState($new_state, $deadline)) {
 155             // state has changed before deadline
 156             $new_state = $this->getConnectivityState();
 157             if ($this->_checkConnectivityState($new_state)) {
 158                 return true;
 159             }
 160         }
 161         // deadline has passed
 162         $new_state = $this->getConnectivityState();
 163
 164         return $this->_checkConnectivityState($new_state);
 165     }
 166
 167     /**
 168      * Close the communication channel associated with this stub.
 169      */
 170     public function close()
 171     {
 172         $this->channel->close();
 173     }
 174
 175     /**
 176      * @param $new_state Connect state
 177      *
 178      * @return bool true if state is CHANNEL_READY
 179      * @throw Exception if state is CHANNEL_FATAL_FAILURE
 180      */
 181     private function _checkConnectivityState($new_state)
 182     {
 183         if ($new_state == \Grpc\CHANNEL_READY) {
 184             return true;
 185         }
 186         if ($new_state == \Grpc\CHANNEL_FATAL_FAILURE) {
 187             throw new \Exception('Failed to connect to server');
 188         }
 189
 190         return false;
 191     }
 192
 193     /**
 194      * constructs the auth uri for the jwt.
 195      *
 196      * @param string $method The method string
 197      *
 198      * @return string The URL string
 199      */
 200     private function _get_jwt_aud_uri($method)
 201     {
 202         // TODO(jtattermusch): This is not the correct implementation
 203         // of extracting JWT "aud" claim. We should rely on
 204         // grpc_metadata_credentials_plugin which
 205         // also provides the correct value of "aud" claim
 206         // in the grpc_auth_metadata_context.service_url field.
 207         // Trying to do the construction of "aud" field ourselves
 208         // is bad.
 209         $last_slash_idx = strrpos($method, '/');
 210         if ($last_slash_idx === false) {
 211             throw new \InvalidArgumentException(
 212                 'service name must have a slash'
 213             );
 214         }
 215         $service_name = substr($method, 0, $last_slash_idx);
 216
 217         if ($this->hostname_override) {
 218             $hostname = $this->hostname_override;
 219         } else {
 220             $hostname = $this->hostname;
 221         }
 222
 223         // Remove the port if it is 443
 224         // See https://github.com/grpc/grpc/blob/07c9f7a36b2a0d34fcffebc85649cf3b8c339b5d/src/core/lib/security/transport/client_auth_filter.cc#L205
 225         if ((strlen($hostname) > 4) && (substr($hostname, -4) === ":443")) {
 226             $hostname = substr($hostname, 0, -4);
 227         }
 228
 229         return 'https://'.$hostname.$service_name;
 230     }
 231
 232     /**
 233      * validate and normalize the metadata array.
 234      *
 235      * @param array $metadata The metadata map
 236      *
 237      * @return array $metadata Validated and key-normalized metadata map
 238      * @throw InvalidArgumentException if key contains invalid characters
 239      */
 240     private function _validate_and_normalize_metadata($metadata)
 241     {
 242         $metadata_copy = [];
 243         foreach ($metadata as $key => $value) {
 244             if (!preg_match('/^[.A-Za-z\d_-]+$/', $key)) {
 245                 throw new \InvalidArgumentException(
 246                     'Metadata keys must be nonempty strings containing only '.
 247                     'alphanumeric characters, hyphens, underscores and dots'
 248                 );
 249             }
 250             $metadata_copy[strtolower($key)] = $value;
 251         }
 252
 253         return $metadata_copy;
 254     }
 255
 256     /**
 257      * Create a function which can be used to create UnaryCall
 258      *
 259      * @param Channel|InterceptorChannel   $channel
 260      * @param callable $deserialize A function that deserializes the response
 261      *
 262      * @return \Closure
 263      */
 264     private function _GrpcUnaryUnary($channel)
 265     {
 266         return function ($method,
 267                          $argument,
 268                          $deserialize,
 269                          array $metadata = [],
 270                          array $options = []) use ($channel) {
 271             $call = $this->call_invoker->UnaryCall(
 272                 $channel,
 273                 $method,
 274                 $deserialize,
 275                 $options
 276             );
 277             $jwt_aud_uri = $this->_get_jwt_aud_uri($method);
 278             if (is_callable($this->update_metadata)) {
 279                 $metadata = call_user_func(
 280                     $this->update_metadata,
 281                     $metadata,
 282                     $jwt_aud_uri
 283                 );
 284             }
 285             $metadata = $this->_validate_and_normalize_metadata(
 286                 $metadata
 287             );
 288             $call->start($argument, $metadata, $options);
 289             return $call;
 290         };
 291     }
 292
 293     /**
 294      * Create a function which can be used to create ServerStreamingCall
 295      *
 296      * @param Channel|InterceptorChannel   $channel
 297      * @param callable $deserialize A function that deserializes the response
 298      *
 299      * @return \Closure
 300      */
 301     private function _GrpcStreamUnary($channel)
 302     {
 303         return function ($method,
 304                          $deserialize,
 305                          array $metadata = [],
 306                          array $options = []) use ($channel) {
 307             $call = $this->call_invoker->ClientStreamingCall(
 308                 $channel,
 309                 $method,
 310                 $deserialize,
 311                 $options
 312             );
 313             $jwt_aud_uri = $this->_get_jwt_aud_uri($method);
 314             if (is_callable($this->update_metadata)) {
 315                 $metadata = call_user_func(
 316                     $this->update_metadata,
 317                     $metadata,
 318                     $jwt_aud_uri
 319                 );
 320             }
 321             $metadata = $this->_validate_and_normalize_metadata(
 322                 $metadata
 323             );
 324             $call->start($metadata);
 325             return $call;
 326         };
 327     }
 328
 329     /**
 330      * Create a function which can be used to create ClientStreamingCall
 331      *
 332      * @param Channel|InterceptorChannel   $channel
 333      * @param callable $deserialize A function that deserializes the response
 334      *
 335      * @return \Closure
 336      */
 337     private function _GrpcUnaryStream($channel)
 338     {
 339         return function ($method,
 340                          $argument,
 341                          $deserialize,
 342                          array $metadata = [],
 343                          array $options = []) use ($channel) {
 344             $call = $this->call_invoker->ServerStreamingCall(
 345                 $channel,
 346                 $method,
 347                 $deserialize,
 348                 $options
 349             );
 350             $jwt_aud_uri = $this->_get_jwt_aud_uri($method);
 351             if (is_callable($this->update_metadata)) {
 352                 $metadata = call_user_func(
 353                     $this->update_metadata,
 354                     $metadata,
 355                     $jwt_aud_uri
 356                 );
 357             }
 358             $metadata = $this->_validate_and_normalize_metadata(
 359                 $metadata
 360             );
 361             $call->start($argument, $metadata, $options);
 362             return $call;
 363         };
 364     }
 365
 366     /**
 367      * Create a function which can be used to create BidiStreamingCall
 368      *
 369      * @param Channel|InterceptorChannel   $channel
 370      * @param callable $deserialize A function that deserializes the response
 371      *
 372      * @return \Closure
 373      */
 374     private function _GrpcStreamStream($channel)
 375     {
 376         return function ($method,
 377                          $deserialize,
 378                          array $metadata = [],
 379                          array $options = []) use ($channel) {
 380             $call = $this->call_invoker->BidiStreamingCall(
 381                 $channel,
 382                 $method,
 383                 $deserialize,
 384                 $options
 385             );
 386             $jwt_aud_uri = $this->_get_jwt_aud_uri($method);
 387             if (is_callable($this->update_metadata)) {
 388                 $metadata = call_user_func(
 389                     $this->update_metadata,
 390                     $metadata,
 391                     $jwt_aud_uri
 392                 );
 393             }
 394             $metadata = $this->_validate_and_normalize_metadata(
 395                 $metadata
 396             );
 397             $call->start($metadata);
 398
 399             return $call;
 400         };
 401     }
 402
 403     /**
 404      * Create a function which can be used to create UnaryCall
 405      *
 406      * @param Channel|InterceptorChannel   $channel
 407      * @param callable $deserialize A function that deserializes the response
 408      *
 409      * @return \Closure
 410      */
 411     private function _UnaryUnaryCallFactory($channel)
 412     {
 413         if (is_a($channel, 'Grpc\Internal\InterceptorChannel')) {
 414             return function ($method,
 415                              $argument,
 416                              $deserialize,
 417                              array $metadata = [],
 418                              array $options = []) use ($channel) {
 419                 return $channel->getInterceptor()->interceptUnaryUnary(
 420                     $method,
 421                     $argument,
 422                     $deserialize,
 423                     $metadata,
 424                     $options,
 425                     $this->_UnaryUnaryCallFactory($channel->getNext())
 426                 );
 427             };
 428         }
 429         return $this->_GrpcUnaryUnary($channel);
 430     }
 431
 432     /**
 433      * Create a function which can be used to create ServerStreamingCall
 434      *
 435      * @param Channel|InterceptorChannel   $channel
 436      * @param callable $deserialize A function that deserializes the response
 437      *
 438      * @return \Closure
 439      */
 440     private function _UnaryStreamCallFactory($channel)
 441     {
 442         if (is_a($channel, 'Grpc\Internal\InterceptorChannel')) {
 443             return function ($method,
 444                              $argument,
 445                              $deserialize,
 446                              array $metadata = [],
 447                              array $options = []) use ($channel) {
 448                 return $channel->getInterceptor()->interceptUnaryStream(
 449                     $method,
 450                     $argument,
 451                     $deserialize,
 452                     $metadata,
 453                     $options,
 454                     $this->_UnaryStreamCallFactory($channel->getNext())
 455                 );
 456             };
 457         }
 458         return $this->_GrpcUnaryStream($channel);
 459     }
 460
 461     /**
 462      * Create a function which can be used to create ClientStreamingCall
 463      *
 464      * @param Channel|InterceptorChannel   $channel
 465      * @param callable $deserialize A function that deserializes the response
 466      *
 467      * @return \Closure
 468      */
 469     private function _StreamUnaryCallFactory($channel)
 470     {
 471         if (is_a($channel, 'Grpc\Internal\InterceptorChannel')) {
 472             return function ($method,
 473                              $deserialize,
 474                              array $metadata = [],
 475                              array $options = []) use ($channel) {
 476                 return $channel->getInterceptor()->interceptStreamUnary(
 477                     $method,
 478                     $deserialize,
 479                     $metadata,
 480                     $options,
 481                     $this->_StreamUnaryCallFactory($channel->getNext())
 482                 );
 483             };
 484         }
 485         return $this->_GrpcStreamUnary($channel);
 486     }
 487
 488     /**
 489      * Create a function which can be used to create BidiStreamingCall
 490      *
 491      * @param Channel|InterceptorChannel   $channel
 492      * @param callable $deserialize A function that deserializes the response
 493      *
 494      * @return \Closure
 495      */
 496     private function _StreamStreamCallFactory($channel)
 497     {
 498         if (is_a($channel, 'Grpc\Internal\InterceptorChannel')) {
 499             return function ($method,
 500                              $deserialize,
 501                              array $metadata = [],
 502                              array $options = []) use ($channel) {
 503                 return $channel->getInterceptor()->interceptStreamStream(
 504                     $method,
 505                     $deserialize,
 506                     $metadata,
 507                     $options,
 508                     $this->_StreamStreamCallFactory($channel->getNext())
 509                 );
 510             };
 511         }
 512         return $this->_GrpcStreamStream($channel);
 513     }
 514
 515     /* This class is intended to be subclassed by generated code, so
 516      * all functions begin with "_" to avoid name collisions. */
 517     /**
 518      * Call a remote method that takes a single argument and has a
 519      * single output.
 520      *
 521      * @param string   $method      The name of the method to call
 522      * @param mixed    $argument    The argument to the method
 523      * @param callable $deserialize A function that deserializes the response
 524      * @param array    $metadata    A metadata map to send to the server
 525      *                              (optional)
 526      * @param array    $options     An array of options (optional)
 527      *
 528      * @return UnaryCall The active call object
 529      */
 530     protected function _simpleRequest(
 531         $method,
 532         $argument,
 533         $deserialize,
 534         array $metadata = [],
 535         array $options = []
 536     ) {
 537         $call_factory = $this->_UnaryUnaryCallFactory($this->channel);
 538         $call = $call_factory($method, $argument, $deserialize, $metadata, $options);
 539         return $call;
 540     }
 541
 542     /**
 543      * Call a remote method that takes a stream of arguments and has a single
 544      * output.
 545      *
 546      * @param string   $method      The name of the method to call
 547      * @param callable $deserialize A function that deserializes the response
 548      * @param array    $metadata    A metadata map to send to the server
 549      *                              (optional)
 550      * @param array    $options     An array of options (optional)
 551      *
 552      * @return ClientStreamingCall The active call object
 553      */
 554     protected function _clientStreamRequest(
 555         $method,
 556         $deserialize,
 557         array $metadata = [],
 558         array $options = []
 559     ) {
 560         $call_factory = $this->_StreamUnaryCallFactory($this->channel);
 561         $call = $call_factory($method, $deserialize, $metadata, $options);
 562         return $call;
 563     }
 564
 565     /**
 566      * Call a remote method that takes a single argument and returns a stream
 567      * of responses.
 568      *
 569      * @param string   $method      The name of the method to call
 570      * @param mixed    $argument    The argument to the method
 571      * @param callable $deserialize A function that deserializes the responses
 572      * @param array    $metadata    A metadata map to send to the server
 573      *                              (optional)
 574      * @param array    $options     An array of options (optional)
 575      *
 576      * @return ServerStreamingCall The active call object
 577      */
 578     protected function _serverStreamRequest(
 579         $method,
 580         $argument,
 581         $deserialize,
 582         array $metadata = [],
 583         array $options = []
 584     ) {
 585         $call_factory = $this->_UnaryStreamCallFactory($this->channel);
 586         $call = $call_factory($method, $argument, $deserialize, $metadata, $options);
 587         return $call;
 588     }
 589
 590     /**
 591      * Call a remote method with messages streaming in both directions.
 592      *
 593      * @param string   $method      The name of the method to call
 594      * @param callable $deserialize A function that deserializes the responses
 595      * @param array    $metadata    A metadata map to send to the server
 596      *                              (optional)
 597      * @param array    $options     An array of options (optional)
 598      *
 599      * @return BidiStreamingCall The active call object
 600      */
 601     protected function _bidiRequest(
 602         $method,
 603         $deserialize,
 604         array $metadata = [],
 605         array $options = []
 606     ) {
 607         $call_factory = $this->_StreamStreamCallFactory($this->channel);
 608         $call = $call_factory($method, $deserialize, $metadata, $options);
 609         return $call;
 610     }
 611 }