BidiStream.php 5.8 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177
  1. <?php
  2. /*
  3. * Copyright 2016 Google LLC
  4. * All rights reserved.
  5. *
  6. * Redistribution and use in source and binary forms, with or without
  7. * modification, are permitted provided that the following conditions are
  8. * met:
  9. *
  10. * * Redistributions of source code must retain the above copyright
  11. * notice, this list of conditions and the following disclaimer.
  12. * * Redistributions in binary form must reproduce the above
  13. * copyright notice, this list of conditions and the following disclaimer
  14. * in the documentation and/or other materials provided with the
  15. * distribution.
  16. * * Neither the name of Google Inc. nor the names of its
  17. * contributors may be used to endorse or promote products derived from
  18. * this software without specific prior written permission.
  19. *
  20. * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
  21. * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
  22. * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
  23. * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
  24. * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
  25. * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
  26. * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
  27. * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
  28. * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
  29. * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
  30. * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
  31. */
  32. namespace Google\ApiCore;
  33. use Google\Rpc\Code;
  34. use Grpc\BidiStreamingCall;
  35. /**
  36. * BidiStream is the response object from a gRPC bidirectional streaming API call.
  37. */
  38. class BidiStream
  39. {
  40. private $call;
  41. private $isComplete = false;
  42. private $writesClosed = false;
  43. private $resourcesGetMethod = null;
  44. private $pendingResources = [];
  45. /**
  46. * BidiStream constructor.
  47. *
  48. * @param BidiStreamingCall $bidiStreamingCall The gRPC bidirectional streaming call object
  49. * @param array $streamingDescriptor
  50. */
  51. public function __construct(BidiStreamingCall $bidiStreamingCall, array $streamingDescriptor = [])
  52. {
  53. $this->call = $bidiStreamingCall;
  54. if (array_key_exists('resourcesGetMethod', $streamingDescriptor)) {
  55. $this->resourcesGetMethod = $streamingDescriptor['resourcesGetMethod'];
  56. }
  57. }
  58. /**
  59. * Write request to the server.
  60. *
  61. * @param mixed $request The request to write
  62. * @throws ValidationException
  63. */
  64. public function write($request)
  65. {
  66. if ($this->isComplete) {
  67. throw new ValidationException("Cannot call write() after streaming call is complete.");
  68. }
  69. if ($this->writesClosed) {
  70. throw new ValidationException("Cannot call write() after calling closeWrite().");
  71. }
  72. $this->call->write($request);
  73. }
  74. /**
  75. * Write all requests in $requests.
  76. *
  77. * @param iterable $requests An Iterable of request objects to write to the server
  78. *
  79. * @throws ValidationException
  80. */
  81. public function writeAll($requests = [])
  82. {
  83. foreach ($requests as $request) {
  84. $this->write($request);
  85. }
  86. }
  87. /**
  88. * Inform the server that no more requests will be written. The write() function cannot be
  89. * called after closeWrite() is called.
  90. * @throws ValidationException
  91. */
  92. public function closeWrite()
  93. {
  94. if ($this->isComplete) {
  95. throw new ValidationException(
  96. "Cannot call closeWrite() after streaming call is complete."
  97. );
  98. }
  99. if (!$this->writesClosed) {
  100. $this->call->writesDone();
  101. $this->writesClosed = true;
  102. }
  103. }
  104. /**
  105. * Read the next response from the server. Returns null if the streaming call completed
  106. * successfully. Throws an ApiException if the streaming call failed.
  107. *
  108. * @throws ValidationException
  109. * @throws ApiException
  110. * @return mixed
  111. */
  112. public function read()
  113. {
  114. if ($this->isComplete) {
  115. throw new ValidationException("Cannot call read() after streaming call is complete.");
  116. }
  117. $resourcesGetMethod = $this->resourcesGetMethod;
  118. if (!is_null($resourcesGetMethod)) {
  119. if (count($this->pendingResources) === 0) {
  120. $response = $this->call->read();
  121. if (!is_null($response)) {
  122. $pendingResources = [];
  123. foreach ($response->$resourcesGetMethod() as $resource) {
  124. $pendingResources[] = $resource;
  125. }
  126. $this->pendingResources = array_reverse($pendingResources);
  127. }
  128. }
  129. $result = array_pop($this->pendingResources);
  130. } else {
  131. $result = $this->call->read();
  132. }
  133. if (is_null($result)) {
  134. $status = $this->call->getStatus();
  135. $this->isComplete = true;
  136. if (!($status->code == Code::OK)) {
  137. throw ApiException::createFromStdClass($status);
  138. }
  139. }
  140. return $result;
  141. }
  142. /**
  143. * Call closeWrite(), and read all responses from the server, until the streaming call is
  144. * completed. Throws an ApiException if the streaming call failed.
  145. *
  146. * @throws ValidationException
  147. * @throws ApiException
  148. * @return \Generator|mixed[]
  149. */
  150. public function closeWriteAndReadAll()
  151. {
  152. $this->closeWrite();
  153. $response = $this->read();
  154. while (!is_null($response)) {
  155. yield $response;
  156. $response = $this->read();
  157. }
  158. }
  159. /**
  160. * Return the underlying gRPC call object
  161. *
  162. * @return \Grpc\BidiStreamingCall|mixed
  163. */
  164. public function getBidiStreamingCall()
  165. {
  166. return $this->call;
  167. }
  168. }