GcpBaseCall.php 6.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223
  1. <?php
  2. /*
  3. *
  4. * Copyright 2018 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. namespace Grpc\Gcp;
  20. abstract class GcpBaseCall
  21. {
  22. const BOUND = 'BOUND';
  23. const UNBIND = 'UNBIND';
  24. const BIND = 'BIND';
  25. protected $gcp_channel;
  26. // It has the Grpc\Channel and related ref_count information for this RPC.
  27. protected $channel_ref;
  28. // If this RPC is 'UNBIND', use it instead of the one from response.
  29. protected $affinity_key;
  30. // Array of [affinity_key, command]
  31. protected $_affinity;
  32. // Information needed to create Grpc\Call object when the RPC starts.
  33. protected $method;
  34. protected $argument;
  35. protected $metadata;
  36. protected $options;
  37. protected $deserialize;
  38. // In GCP extension, it is when a RPC calls "start", we pick a channel.
  39. // Thus we need to save the $me
  40. protected $metadata_rpc = array();
  41. // first_rpc is used to check whether the first request is sent for client
  42. // streaming RPC.
  43. protected $has_real_call = null;
  44. protected $real_call;
  45. /**
  46. * Create a new Call wrapper object.
  47. *
  48. * @param Channel $channel The channel to communicate on
  49. * @param string $method The method to call on the
  50. * remote server
  51. * @param callback $deserialize A callback function to deserialize
  52. * the response
  53. * @param array $options Call options (optional)
  54. */
  55. public function __construct($channel, $method, $deserialize, $options)
  56. {
  57. $this->gcp_channel = $channel;
  58. $this->method = $method;
  59. $this->deserialize = $deserialize;
  60. $this->options = $options;
  61. $this->_affinity = null;
  62. if (isset($this->gcp_channel->affinity_conf['affinity_by_method'][$method])) {
  63. $this->_affinity = $this->gcp_channel->affinity_conf['affinity_by_method'][$method];
  64. }
  65. }
  66. /**
  67. * Pick a ChannelRef from the channel pool based on the request and
  68. * the affinity config.
  69. *
  70. * @param mixed $argument Requests.
  71. *
  72. * @return ChannelRef
  73. */
  74. protected function _rpcPreProcess($argument)
  75. {
  76. $this->affinity_key = null;
  77. if ($this->_affinity) {
  78. $command = $this->_affinity['command'];
  79. if ($command == self::BOUND || $command == self::UNBIND) {
  80. $this->affinity_key = $this->getAffinityKeyFromProto($argument);
  81. }
  82. }
  83. $this->channel_ref = $this->gcp_channel->getChannelRef($this->affinity_key);
  84. $this->channel_ref->activeStreamRefIncr();
  85. return $this->channel_ref;
  86. }
  87. /**
  88. * Update ChannelRef when RPC finishes.
  89. *
  90. * @param \stdClass $status The status object, with integer $code, string
  91. * $details, and array $metadata members
  92. * @param mixed $response Response.
  93. */
  94. protected function _rpcPostProcess($status, $response)
  95. {
  96. if ($this->_affinity) {
  97. $command = $this->_affinity['command'];
  98. if ($command == self::BIND) {
  99. if ($status->code != \Grpc\STATUS_OK) {
  100. return;
  101. }
  102. $affinity_key = $this->getAffinityKeyFromProto($response);
  103. $this->gcp_channel->bind($this->channel_ref, $affinity_key);
  104. } elseif ($command == self::UNBIND) {
  105. $this->gcp_channel->unbind($this->affinity_key);
  106. }
  107. }
  108. $this->channel_ref->activeStreamRefDecr();
  109. }
  110. /**
  111. * Get the affinity key based on the affinity config.
  112. *
  113. * @param mixed $proto Objects may contain the affinity key.
  114. *
  115. * @return string Affinity key.
  116. */
  117. protected function getAffinityKeyFromProto($proto)
  118. {
  119. if ($this->_affinity) {
  120. $names = $this->_affinity['affinityKey'];
  121. $names_arr = explode(".", $names);
  122. foreach ($names_arr as $name) {
  123. $getAttrMethod = 'get' . ucfirst($name);
  124. $proto = call_user_func_array(array($proto, $getAttrMethod), array());
  125. }
  126. return $proto;
  127. }
  128. echo "Cannot find the field in the proto\n";
  129. }
  130. /**
  131. * @return mixed The metadata sent by the server
  132. */
  133. public function getMetadata()
  134. {
  135. if (!$this->has_real_call) {
  136. $this->createRealCall();
  137. $this->has_real_call = true;
  138. }
  139. return $this->real_call->getMetadata();
  140. }
  141. /**
  142. * @return mixed The trailing metadata sent by the server
  143. */
  144. public function getTrailingMetadata()
  145. {
  146. if (!$this->has_real_call) {
  147. $this->createRealCall();
  148. $this->has_real_call = true;
  149. }
  150. return $this->real_call->getTrailingMetadata();
  151. }
  152. /**
  153. * @return string The URI of the endpoint
  154. */
  155. public function getPeer()
  156. {
  157. if (!$this->has_real_call) {
  158. $this->createRealCall();
  159. $this->has_real_call = true;
  160. }
  161. return $this->real_call->getPeer();
  162. }
  163. /**
  164. * Cancels the call.
  165. */
  166. public function cancel()
  167. {
  168. if (!$this->has_real_call) {
  169. $this->has_real_call = true;
  170. $this->createRealCall();
  171. }
  172. $this->real_call->cancel();
  173. }
  174. /**
  175. * Serialize a message to the protobuf binary format.
  176. *
  177. * @param mixed $data The Protobuf message
  178. *
  179. * @return string The protobuf binary format
  180. */
  181. protected function _serializeMessage($data)
  182. {
  183. return $this->real_call->_serializeMessage($data);
  184. }
  185. /**
  186. * Deserialize a response value to an object.
  187. *
  188. * @param string $value The binary value to deserialize
  189. *
  190. * @return mixed The deserialized value
  191. */
  192. protected function _deserializeResponse($value)
  193. {
  194. return $this->real_call->_deserializeResponse($value);
  195. }
  196. /**
  197. * Set the CallCredentials for the underlying Call.
  198. *
  199. * @param CallCredentials $call_credentials The CallCredentials object
  200. */
  201. public function setCallCredentials($call_credentials)
  202. {
  203. $this->call->setCredentials($call_credentials);
  204. }
  205. }