Merge pull request #3181 An error occurred from stanley-cheung/php_per_message_compression

PHP: support per message compression disable

src/php/ext/grpc/README.md

@@ -4,7 +4,7 @@ gRPC PHP Extension
 # Requirements
 
  * PHP 5.5+
- * [gRPC core library](https://github.com/grpc/grpc) 0.10.0
+ * [gRPC core library](https://github.com/grpc/grpc) 0.11.0
 
 # Installation
 
@@ -22,26 +22,21 @@ Clone the gRPC source code repository
 $ git clone https://github.com/grpc/grpc.git
 ```
 
-Build and install the Protocol Buffers compiler (protoc)
+Build and install the gRPC C core libraries
 
-```
-$ # from grpc
-$ git checkout --track origin/release-0_9
+```sh
+$ cd grpc
+$ git checkout --track origin/release-0_11
 $ git pull --recurse-submodules && git submodule update --init --recursive
-$ cd third_party/protobuf
-$ ./autogen.sh
-$ ./configure
 $ make
-$ make check
 $ sudo make install
 ```
 
-Build and install the gRPC C core library
+Note: you may encounter a warning about the Protobuf compiler `protoc` 3.0.0+ not being installed. The following might help, and will be useful later on when we need to compile the `protoc-gen-php` tool.
 
 ```sh
-$ # from grpc
-$ make
-$ sudo make install
+$ cd grpc/third_party/protobuf
+$ sudo make install   # 'make' should have been run by core grpc
 ```
 
 ## Install the gRPC PHP extension
@@ -55,7 +50,7 @@ $ sudo pecl install grpc
 Note: before a stable release, you may need to do
 
 ```sh
-$ sudo pecl install grpc-0.5.1
+$ sudo pecl install grpc-beta
 ```
 
 OR

src/php/ext/grpc/call.c

@@ -265,6 +265,9 @@ PHP_METHOD(Call, startBatch) {
   HashTable *array_hash;
   HashPosition array_pointer;
   HashTable *status_hash;
+  HashTable *message_hash;
+  zval **message_value;
+  zval **message_flags;
   char *key;
   uint key_len;
   ulong index;
@@ -319,13 +322,33 @@ PHP_METHOD(Call, startBatch) {
             metadata.metadata;
         break;
       case GRPC_OP_SEND_MESSAGE:
-        if (Z_TYPE_PP(value) != IS_STRING) {
+        if (Z_TYPE_PP(value) != IS_ARRAY) {
+          zend_throw_exception(spl_ce_InvalidArgumentException,
+                               "Expected an array for send message",
+                               1 TSRMLS_CC);
+          goto cleanup;
+        }
+        message_hash = Z_ARRVAL_PP(value);
+        if (zend_hash_find(message_hash, "flags", sizeof("flags"),
+                           (void **)&message_flags) == SUCCESS) {
+          if (Z_TYPE_PP(message_flags) != IS_LONG) {
+            zend_throw_exception(spl_ce_InvalidArgumentException,
+                                 "Expected an int for message flags",
+                                 1 TSRMLS_CC);
+          }
+          ops[op_num].flags = Z_LVAL_PP(message_flags) & GRPC_WRITE_USED_MASK;
+        }
+        if (zend_hash_find(message_hash, "message", sizeof("message"),
+                           (void **)&message_value) != SUCCESS ||
+            Z_TYPE_PP(message_value) != IS_STRING) {
           zend_throw_exception(spl_ce_InvalidArgumentException,
                                "Expected a string for send message",
                                1 TSRMLS_CC);
+          goto cleanup;
         }
         ops[op_num].data.send_message =
-            string_to_byte_buffer(Z_STRVAL_PP(value), Z_STRLEN_PP(value));
+            string_to_byte_buffer(Z_STRVAL_PP(message_value),
+                                  Z_STRLEN_PP(message_value));
         break;
       case GRPC_OP_SEND_CLOSE_FROM_CLIENT:
         break;

src/php/ext/grpc/package.xml

@@ -10,43 +10,48 @@
   <email>grpc-packages@google.com</email>
   <active>yes</active>
  </lead>
- <date>2015-07-09</date>
- <time>21:47:27</time>
+ <date>2015-09-01</date>
+ <time>13:37:33</time>
  <version>
-  <release>0.5.1</release>
-  <api>0.5.1</api>
+  <release>0.6.0</release>
+  <api>0.6.0</api>
  </version>
  <stability>
-  <release>alpha</release>
-  <api>alpha</api>
+  <release>beta</release>
+  <api>beta</api>
  </stability>
  <license>BSD</license>
  <notes>
-Update to wrap gRPC C Core version 0.10.0
+ - support per message compression disable
+ - expose per-call host override option
+ - expose connectivity API
+ - expose channel target and call peer
+ - add user-agent
+ - update to wrap gRPC C core library beta version 0.11.0
  </notes>
  <contents>
   <dir baseinstalldir="/" name="/">
    <file baseinstalldir="/" md5sum="6f19828fb869b7b8a590cbb76b4f996d" name="byte_buffer.c" role="src" />
    <file baseinstalldir="/" md5sum="c8de0f819499c48adfc8d7f472c0196b" name="byte_buffer.h" role="src" />
-   <file baseinstalldir="/" md5sum="cb45b62f767ae7b4377761df696649fc" name="call.c" role="src" />
+   <file baseinstalldir="/" md5sum="d64c9005993de02abac55664b0b9e0b2" name="call.c" role="src" />
    <file baseinstalldir="/" md5sum="26acbf04c30162c2d2aca4688bb2aec8" name="call.h" role="src" />
-   <file baseinstalldir="/" md5sum="50837fbdb2892795f1871b22e5979762" name="channel.c" role="src" />
-   <file baseinstalldir="/" md5sum="f1b66029daeced20b47cf00cc6523fc8" name="channel.h" role="src" />
-   <file baseinstalldir="/" md5sum="81a1193e93d8b6602add8ac360de565b" name="completion_queue.c" role="src" />
+   <file baseinstalldir="/" md5sum="15e56239b32c803f073e8a2b9f96e8c3" name="channel.c" role="src" />
+   <file baseinstalldir="/" md5sum="ed4b00c0cf3702b115d0cfa87450dc09" name="channel.h" role="src" />
+   <file baseinstalldir="/" md5sum="55ab7a42f9dd9bfc7e28a61cfc5fca63" name="completion_queue.c" role="src" />
    <file baseinstalldir="/" md5sum="f10b5bb232d74a6878e829e2e76cdaa2" name="completion_queue.h" role="src" />
    <file baseinstalldir="/" md5sum="a22f8eac0164761058cc4d9eb2ceb069" name="config.m4" role="src" />
-   <file baseinstalldir="/" md5sum="8c3f1e11dac623001378bfd53b554f08" name="credentials.c" role="src" />
+   <file baseinstalldir="/" md5sum="588752c908f7bc1663f7b8fc922ae661" name="credentials.c" role="src" />
    <file baseinstalldir="/" md5sum="6988d6e97c19c8f8e3eb92371cf8246b" name="credentials.h" role="src" />
    <file baseinstalldir="/" md5sum="38a1bc979d810c36ebc2a52d4b7b5319" name="CREDITS" role="doc" />
    <file baseinstalldir="/" md5sum="3f35b472bbdef5a788cd90617d7d0847" name="LICENSE" role="doc" />
-   <file baseinstalldir="/" md5sum="6aaa7a290122d230f2d8c4e4e05da4a9" name="php_grpc.c" role="src" />
+   <file baseinstalldir="/" md5sum="6a550516a1423def0786851c76f87c85" name="php_grpc.c" role="src" />
    <file baseinstalldir="/" md5sum="673b07859d9f69232f8a754c56780686" name="php_grpc.h" role="src" />
    <file baseinstalldir="/" md5sum="c1d0b42fd77b7d6740bf7744bee90af5" name="README.md" role="doc" />
-   <file baseinstalldir="/" md5sum="30997dd423403e1f8ad09dcee598e5c4" name="server.c" role="src" />
+   <file baseinstalldir="/" md5sum="3e4e960454ebb2fc7b78a840493f5315" name="server.c" role="src" />
    <file baseinstalldir="/" md5sum="4b730f06d14cbbb0642bdbd194749595" name="server.h" role="src" />
-   <file baseinstalldir="/" md5sum="f6930beafb6c0e061899262f2f077e98" name="server_credentials.c" role="src" />
+   <file baseinstalldir="/" md5sum="34ea881f1fe960d190d0713422cf8916" name="server_credentials.c" role="src" />
    <file baseinstalldir="/" md5sum="9c4b4cc06356a8a39a16a085a9b85996" name="server_credentials.h" role="src" />
-   <file baseinstalldir="/" md5sum="c89c623cd17177ebde18313fc5c17122" name="timeval.c" role="src" />
+   <file baseinstalldir="/" md5sum="7646ec78cb133f66ba59e03c6f451e39" name="timeval.c" role="src" />
    <file baseinstalldir="/" md5sum="496e27a100b4d93ca3fb35c924c5e163" name="timeval.h" role="src" />
   </dir>
  </contents>
@@ -93,5 +98,25 @@ First alpha release
 Update to wrap gRPC C Core version 0.10.0
    </notes>
   </release>
+  <release>
+   <version>
+    <release>0.6.0</release>
+    <api>0.6.0</api>
+   </version>
+   <stability>
+    <release>beta</release>
+    <api>beta</api>
+   </stability>
+   <date>2015-09-01</date>
+   <license>BSD</license>
+   <notes>
+ - support per message compression disable
+ - expose per-call host override option
+ - expose connectivity API
+ - expose channel target and call peer
+ - add user-agent
+ - update to wrap gRPC C core library beta version 0.11.0
+   </notes>
+  </release>
  </changelog>
 </package>

src/php/lib/Grpc/AbstractCall.php

@@ -92,4 +92,4 @@ abstract class AbstractCall {
     }
     return call_user_func($this->deserialize, $value);
   }
-}
+}

src/php/lib/Grpc/BaseStub.php

@@ -168,7 +168,8 @@ class BaseStub {
   public function _simpleRequest($method,
                                  $argument,
                                  callable $deserialize,
-                                 $metadata = array()) {
+                                 $metadata = array(),
+                                 $options = array()) {
     list($actual_metadata, $timeout)  = $this->_extract_timeout_from_metadata($metadata);
     $call = new UnaryCall($this->channel, $method, $deserialize, $timeout);
     $jwt_aud_uri = $this->_get_jwt_aud_uri($method);
@@ -177,7 +178,7 @@ class BaseStub {
                                         $actual_metadata,
                                         $jwt_aud_uri);
     }
-    $call->start($argument, $actual_metadata);
+    $call->start($argument, $actual_metadata, $options);
     return $call;
   }
 
@@ -193,7 +194,6 @@ class BaseStub {
    * @return ClientStreamingSurfaceActiveCall The active call object
    */
   public function _clientStreamRequest($method,
-                                       $arguments,
                                        callable $deserialize,
                                        $metadata = array()) {
     list($actual_metadata, $timeout)  = $this->_extract_timeout_from_metadata($metadata);
@@ -204,7 +204,7 @@ class BaseStub {
                                         $actual_metadata,
                                         $jwt_aud_uri);
     }
-    $call->start($arguments, $actual_metadata);
+    $call->start($actual_metadata);
     return $call;
   }
 
@@ -221,7 +221,8 @@ class BaseStub {
   public function _serverStreamRequest($method,
                                        $argument,
                                        callable $deserialize,
-                                       $metadata = array()) {
+                                       $metadata = array(),
+                                       $options = array()) {
     list($actual_metadata, $timeout)  = $this->_extract_timeout_from_metadata($metadata);
     $call = new ServerStreamingCall($this->channel, $method, $deserialize, $timeout);
     $jwt_aud_uri = $this->_get_jwt_aud_uri($method);
@@ -230,7 +231,7 @@ class BaseStub {
                                         $actual_metadata,
                                         $jwt_aud_uri);
     }
-    $call->start($argument, $actual_metadata);
+    $call->start($argument, $actual_metadata, $options);
     return $call;
   }
 

src/php/lib/Grpc/BidiStreamingCall.php

@@ -42,7 +42,7 @@ class BidiStreamingCall extends AbstractCall {
    * Start the call
    * @param array $metadata Metadata to send with the call, if applicable
    */
-  public function start($metadata) {
+  public function start($metadata = array()) {
     $this->call->startBatch([OP_SEND_INITIAL_METADATA => $metadata]);
   }
 
@@ -66,9 +66,15 @@ class BidiStreamingCall extends AbstractCall {
    * Write a single message to the server. This cannot be called after
    * writesDone is called.
    * @param ByteBuffer $data The data to write
+   * @param array $options an array of options, possible keys:
+   *              'flags' => a number
    */
-  public function write($data) {
-    $this->call->startBatch([OP_SEND_MESSAGE => $data->serialize()]);
+  public function write($data, $options = array()) {
+    $message_array = ['message' => $data->serialize()];
+    if (isset($options['flags'])) {
+      $message_array['flags'] = $options['flags'];
+    }
+    $this->call->startBatch([OP_SEND_MESSAGE => $message_array]);
   }
 
   /**
@@ -86,7 +92,7 @@ class BidiStreamingCall extends AbstractCall {
   public function getStatus() {
     $status_event = $this->call->startBatch([
         OP_RECV_STATUS_ON_CLIENT => true
-                                              ]);
+    ]);
     return $status_event->status;
   }
 }

src/php/lib/Grpc/ClientStreamingCall.php

@@ -40,15 +40,25 @@ namespace Grpc;
 class ClientStreamingCall extends AbstractCall {
   /**
    * Start the call.
-   * @param Traversable $arg_iter The iterator of arguments to send
    * @param array $metadata Metadata to send with the call, if applicable
    */
-  public function start($arg_iter, $metadata = array()) {
-    $event = $this->call->startBatch([OP_SEND_INITIAL_METADATA => $metadata]);
-    foreach($arg_iter as $arg) {
-      $this->call->startBatch([OP_SEND_MESSAGE => $arg->serialize()]);
+  public function start($metadata = array()) {
+    $this->call->startBatch([OP_SEND_INITIAL_METADATA => $metadata]);
+  }
+
+  /**
+   * Write a single message to the server. This cannot be called after
+   * wait is called.
+   * @param ByteBuffer $data The data to write
+   * @param array $options an array of options, possible keys:
+   *              'flags' => a number
+   */
+  public function write($data, $options = array()) {
+    $message_array = ['message' => $data->serialize()];
+    if (isset($options['flags'])) {
+      $message_array['flags'] = $options['flags'];
     }
-    $this->call->startBatch([OP_SEND_CLOSE_FROM_CLIENT => true]);
+    $this->call->startBatch([OP_SEND_MESSAGE => $message_array]);
   }
 
   /**
@@ -57,6 +67,7 @@ class ClientStreamingCall extends AbstractCall {
    */
   public function wait() {
     $event = $this->call->startBatch([
+        OP_SEND_CLOSE_FROM_CLIENT => true,
         OP_RECV_INITIAL_METADATA => true,
         OP_RECV_MESSAGE => true,
         OP_RECV_STATUS_ON_CLIENT => true]);

src/php/lib/Grpc/ServerStreamingCall.php

@@ -40,14 +40,20 @@ namespace Grpc;
 class ServerStreamingCall extends AbstractCall {
   /**
    * Start the call
-   * @param $arg The argument to send
+   * @param $data The data to send
    * @param array $metadata Metadata to send with the call, if applicable
+   * @param array $options an array of options, possible keys:
+   *              'flags' => a number
    */
-  public function start($arg, $metadata = array()) {
+  public function start($data, $metadata = array(), $options = array()) {
+    $message_array = ['message' => $data->serialize()];
+    if (isset($options['flags'])) {
+      $message_array['flags'] = $options['flags'];
+    }
     $event = $this->call->startBatch([
         OP_SEND_INITIAL_METADATA => $metadata,
         OP_RECV_INITIAL_METADATA => true,
-        OP_SEND_MESSAGE => $arg->serialize(),
+        OP_SEND_MESSAGE => $message_array,
         OP_SEND_CLOSE_FROM_CLIENT => true]);
     $this->metadata = $event->metadata;
   }
@@ -71,7 +77,7 @@ class ServerStreamingCall extends AbstractCall {
   public function getStatus() {
     $status_event = $this->call->startBatch([
         OP_RECV_STATUS_ON_CLIENT => true
-                                              ]);
+    ]);
     return $status_event->status;
   }
-}
+}

src/php/lib/Grpc/UnaryCall.php

@@ -40,14 +40,20 @@ namespace Grpc;
 class UnaryCall extends AbstractCall {
   /**
    * Start the call
-   * @param $arg The argument to send
+   * @param $data The data to send
    * @param array $metadata Metadata to send with the call, if applicable
+   * @param array $options an array of options, possible keys:
+   *              'flags' => a number
    */
-  public function start($arg, $metadata = array()) {
+  public function start($data, $metadata = array(), $options = array()) {
+    $message_array = ['message' => $data->serialize()];
+    if (isset($options['flags'])) {
+      $message_array['flags'] = $options['flags'];
+    }
     $event = $this->call->startBatch([
         OP_SEND_INITIAL_METADATA => $metadata,
         OP_RECV_INITIAL_METADATA => true,
-        OP_SEND_MESSAGE => $arg->serialize(),
+        OP_SEND_MESSAGE => $message_array,
         OP_SEND_CLOSE_FROM_CLIENT => true]);
     $this->metadata = $event->metadata;
   }
@@ -62,4 +68,4 @@ class UnaryCall extends AbstractCall {
         OP_RECV_STATUS_ON_CLIENT => true]);
     return array($this->deserializeResponse($event->message), $event->status);
   }
-}
+}

src/php/tests/generated_code/AbstractGeneratedCodeTest.php

@@ -51,6 +51,18 @@ abstract class AbstractGeneratedCodeTest extends PHPUnit_Framework_TestCase {
     $this->assertTrue(is_string(self::$client->getTarget()));
   }
 
+  public function testWriteFlags() {
+    $div_arg = new math\DivArgs();
+    $div_arg->setDividend(7);
+    $div_arg->setDivisor(4);
+    $call = self::$client->Div($div_arg, array(), array('flags' => Grpc\WRITE_NO_COMPRESS));
+    $this->assertTrue(is_string($call->getPeer()));
+    list($response, $status) = $call->wait();
+    $this->assertSame(1, $response->getQuotient());
+    $this->assertSame(3, $response->getRemainder());
+    $this->assertSame(\Grpc\STATUS_OK, $status->code);
+  }
+
   public function testSimpleRequest() {
     $div_arg = new math\DivArgs();
     $div_arg->setDividend(7);
@@ -79,15 +91,13 @@ abstract class AbstractGeneratedCodeTest extends PHPUnit_Framework_TestCase {
   }
 
   public function testClientStreaming() {
-    $num_iter = function() {
-      for ($i = 0; $i < 7; $i++) {
-        $num = new math\Num();
-        $num->setNum($i);
-        yield $num;
-      }
-    };
-    $call = self::$client->Sum($num_iter());
+    $call = self::$client->Sum();
     $this->assertTrue(is_string($call->getPeer()));
+    for ($i = 0; $i < 7; $i++) {
+      $num = new math\Num();
+      $num->setNum($i);
+      $call->write($num);
+    }
     list($response, $status) = $call->wait();
     $this->assertSame(21, $response->getNum());
     $this->assertSame(\Grpc\STATUS_OK, $status->code);

src/php/tests/interop/interop_client.php

@@ -173,7 +173,11 @@ function clientStreaming($stub) {
         return $request;
       }, $request_lengths);
 
-  list($result, $status) = $stub->StreamingInputCall($requests)->wait();
+  $call = $stub->StreamingInputCall();
+  foreach ($requests as $request) {
+    $call->write($request);
+  }
+  list($result, $status) = $call->wait();
   hardAssert($status->code === Grpc\STATUS_OK, 'Call did not complete successfully');
   hardAssert($result->getAggregatedPayloadSize() === 74922,
               'aggregated_payload_size was incorrect');
@@ -246,6 +250,19 @@ function pingPong($stub) {
               'Call did not complete successfully');
 }
 
+/**
+ * Run the cancel_after_begin test.
+ * Passes when run against the Node server as of 2015-08-28
+ * @param $stub Stub object that has service methods.
+ */
+function cancelAfterBegin($stub) {
+  $call = $stub->StreamingInputCall();
+  $call->cancel();
+  list($result, $status) = $call->wait();
+  hardAssert($status->code === Grpc\STATUS_CANCELLED,
+             'Call status was not CANCELLED');
+}
+
 /**
  * Run the cancel_after_first_response test.
  * Passes when run against the Node server as of 2015-04-30
@@ -353,6 +370,9 @@ switch ($args['test_case']) {
   case 'ping_pong':
     pingPong($stub);
     break;
+  case 'cancel_after_begin':
+    cancelAfterBegin($stub);
+    break;
   case 'cancel_after_first_response':
     cancelAfterFirstResponse($stub);
     break;
@@ -368,11 +388,7 @@ switch ($args['test_case']) {
   case 'jwt_token_creds':
     jwtTokenCreds($stub, $args);
     break;
-  case 'cancel_after_begin':
-    // Currently unimplementable with the current API design
-    // Specifically, in the ClientStreamingCall->start() method, the
-    // messages are sent immediately after metadata is sent. There is
-    // currently no way to cancel before messages are sent.
   default:
+    echo "Unsupported test case $args[test_case]\n";
     exit(1);
 }

src/php/tests/unit_tests/EndToEndTest.php

@@ -91,6 +91,51 @@ class EndToEndTest extends PHPUnit_Framework_TestCase{
     unset($server_call);
   }
 
+  public function testMessageWriteFlags() {
+    $deadline = Grpc\Timeval::infFuture();
+    $req_text = 'message_write_flags_test';
+    $status_text = 'xyz';
+    $call = new Grpc\Call($this->channel,
+                          'dummy_method',
+                          $deadline);
+
+    $event = $call->startBatch([
+        Grpc\OP_SEND_INITIAL_METADATA => [],
+        Grpc\OP_SEND_MESSAGE => ['message' => $req_text,
+                                 'flags' => Grpc\WRITE_NO_COMPRESS],
+        Grpc\OP_SEND_CLOSE_FROM_CLIENT => true
+                                       ]);
+
+    $this->assertTrue($event->send_metadata);
+    $this->assertTrue($event->send_close);
+
+    $event = $this->server->requestCall();
+    $this->assertSame('dummy_method', $event->method);
+    $server_call = $event->call;
+
+    $event = $server_call->startBatch([
+        Grpc\OP_SEND_INITIAL_METADATA => [],
+        Grpc\OP_SEND_STATUS_FROM_SERVER => [
+            'metadata' => [],
+            'code' => Grpc\STATUS_OK,
+            'details' => $status_text
+        ],
+    ]);
+
+    $event = $call->startBatch([
+        Grpc\OP_RECV_INITIAL_METADATA => true,
+        Grpc\OP_RECV_STATUS_ON_CLIENT => true
+                                 ]);
+
+    $status = $event->status;
+    $this->assertSame([], $status->metadata);
+    $this->assertSame(Grpc\STATUS_OK, $status->code);
+    $this->assertSame($status_text, $status->details);
+
+    unset($call);
+    unset($server_call);
+  }
+
   public function testClientServerFullRequestResponse() {
     $deadline = Grpc\Timeval::infFuture();
     $req_text = 'client_server_full_request_response';
@@ -104,7 +149,7 @@ class EndToEndTest extends PHPUnit_Framework_TestCase{
     $event = $call->startBatch([
         Grpc\OP_SEND_INITIAL_METADATA => [],
         Grpc\OP_SEND_CLOSE_FROM_CLIENT => true,
-        Grpc\OP_SEND_MESSAGE => $req_text
+        Grpc\OP_SEND_MESSAGE => ['message' => $req_text],
                                        ]);
 
     $this->assertTrue($event->send_metadata);
@@ -117,7 +162,7 @@ class EndToEndTest extends PHPUnit_Framework_TestCase{
 
     $event = $server_call->startBatch([
         Grpc\OP_SEND_INITIAL_METADATA => [],
-        Grpc\OP_SEND_MESSAGE => $reply_text,
+        Grpc\OP_SEND_MESSAGE => ['message' => $reply_text],
         Grpc\OP_SEND_STATUS_FROM_SERVER => [
             'metadata' => [],
             'code' => Grpc\STATUS_OK,

src/php/tests/unit_tests/SecureEndToEndTest.php

@@ -107,6 +107,53 @@ class SecureEndToEndTest extends PHPUnit_Framework_TestCase{
     unset($server_call);
   }
 
+  public function testMessageWriteFlags() {
+    $deadline = Grpc\Timeval::infFuture();
+    $req_text = 'message_write_flags_test';
+    $status_text = 'xyz';
+    $call = new Grpc\Call($this->channel,
+                          'dummy_method',
+                          $deadline,
+                          $this->host_override);
+
+    $event = $call->startBatch([
+        Grpc\OP_SEND_INITIAL_METADATA => [],
+        Grpc\OP_SEND_MESSAGE => ['message' => $req_text,
+                                 'flags' => Grpc\WRITE_NO_COMPRESS],
+        Grpc\OP_SEND_CLOSE_FROM_CLIENT => true
+                                       ]);
+
+    $this->assertTrue($event->send_metadata);
+    $this->assertTrue($event->send_close);
+
+    $event = $this->server->requestCall();
+    $this->assertSame('dummy_method', $event->method);
+    $server_call = $event->call;
+
+    $event = $server_call->startBatch([
+        Grpc\OP_SEND_INITIAL_METADATA => [],
+        Grpc\OP_SEND_STATUS_FROM_SERVER => [
+            'metadata' => [],
+            'code' => Grpc\STATUS_OK,
+            'details' => $status_text
+        ],
+    ]);
+
+    $event = $call->startBatch([
+        Grpc\OP_RECV_INITIAL_METADATA => true,
+        Grpc\OP_RECV_STATUS_ON_CLIENT => true
+    ]);
+
+    $this->assertSame([], $event->metadata);
+    $status = $event->status;
+    $this->assertSame([], $status->metadata);
+    $this->assertSame(Grpc\STATUS_OK, $status->code);
+    $this->assertSame($status_text, $status->details);
+
+    unset($call);
+    unset($server_call);
+  }
+
   public function testClientServerFullRequestResponse() {
     $deadline = Grpc\Timeval::infFuture();
     $req_text = 'client_server_full_request_response';
@@ -121,7 +168,7 @@ class SecureEndToEndTest extends PHPUnit_Framework_TestCase{
     $event = $call->startBatch([
         Grpc\OP_SEND_INITIAL_METADATA => [],
         Grpc\OP_SEND_CLOSE_FROM_CLIENT => true,
-        Grpc\OP_SEND_MESSAGE => $req_text
+        Grpc\OP_SEND_MESSAGE => ['message' => $req_text]
                                        ]);
 
     $this->assertTrue($event->send_metadata);
@@ -134,7 +181,7 @@ class SecureEndToEndTest extends PHPUnit_Framework_TestCase{
 
     $event = $server_call->startBatch([
         Grpc\OP_SEND_INITIAL_METADATA => [],
-        Grpc\OP_SEND_MESSAGE => $reply_text,
+        Grpc\OP_SEND_MESSAGE => ['message' => $reply_text],
         Grpc\OP_SEND_STATUS_FROM_SERVER => [
             'metadata' => [],
             'code' => Grpc\STATUS_OK,