mirror of https://github.com/grpc/grpc.git
move InterceptorChannel to Internal subdir; change yearpull/13342/head
ZhouyihaiDing
7 years ago
parent
aef957950a
commit
52b95cf1fb
4 changed files with 888 additions and 90 deletions
@ -0,0 +1,86 @@ |
||||
<?php |
||||
/* |
||||
* |
||||
* Copyright 2018 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; |
||||
|
||||
/** |
||||
* Represents an interceptor that intercept RPC invocations before call starts. |
||||
* This is an EXPERIMENTAL API. |
||||
*/ |
||||
class Interceptor |
||||
{ |
||||
public function interceptUnaryUnary( |
||||
$method, |
||||
$argument, |
||||
array $metadata = [], |
||||
array $options = [], |
||||
$continuation |
||||
) { |
||||
return $continuation($method, $argument, $metadata, $options); |
||||
} |
||||
|
||||
public function interceptStreamUnary( |
||||
$method, |
||||
array $metadata = [], |
||||
array $options = [], |
||||
$continuation |
||||
) { |
||||
return $continuation($method, $metadata, $options); |
||||
} |
||||
|
||||
public function interceptUnaryStream( |
||||
$method, |
||||
$argument, |
||||
array $metadata = [], |
||||
array $options = [], |
||||
$continuation |
||||
) { |
||||
return $continuation($method, $argument, $metadata, $options); |
||||
} |
||||
|
||||
public function interceptStreamStream( |
||||
$method, |
||||
array $metadata = [], |
||||
array $options = [], |
||||
$continuation |
||||
) { |
||||
return $continuation($method, $metadata, $options); |
||||
} |
||||
|
||||
/** |
||||
* Intercept the methods with Channel |
||||
* |
||||
* @param Channel|InterceptorChannel $channel An already created Channel or InterceptorChannel object (optional) |
||||
* @param Interceptor|Interceptor[] $interceptors interceptors to be added |
||||
* |
||||
* @return InterceptorChannel |
||||
*/ |
||||
public static function intercept($channel, $interceptors) |
||||
{ |
||||
if (is_array($interceptors)) { |
||||
for ($i = count($interceptors) - 1; $i >= 0; $i--) { |
||||
$channel = new InterceptorChannel($channel, $interceptors[$i]); |
||||
} |
||||
} else { |
||||
$channel = new InterceptorChannel($channel, $interceptors); |
||||
} |
||||
return $channel; |
||||
} |
||||
} |
||||
|
||||
@ -0,0 +1,76 @@ |
||||
<?php |
||||
/* |
||||
* |
||||
* Copyright 2018 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; |
||||
|
||||
/** |
||||
* This is a PRIVATE API and can change without notice. |
||||
*/ |
||||
class InterceptorChannel |
||||
{ |
||||
private $next = null; |
||||
private $interceptor; |
||||
|
||||
/** |
||||
* @param Channel|InterceptorChannel $channel An already created Channel |
||||
* or InterceptorChannel object (optional) |
||||
* @param Interceptor $interceptor |
||||
*/ |
||||
public function __construct($channel, $interceptor) |
||||
{ |
||||
if (!is_a($channel, 'Grpc\Channel') && |
||||
!is_a($channel, 'Grpc\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->interceptor = $interceptor; |
||||
$this->next = $channel; |
||||
} |
||||
|
||||
public function getNext() |
||||
{ |
||||
return $this->next; |
||||
} |
||||
|
||||
public function getInterceptor() |
||||
{ |
||||
return $this->interceptor; |
||||
} |
||||
|
||||
public function getTarget() |
||||
{ |
||||
return $this->getNext()->getTarget(); |
||||
} |
||||
|
||||
public function watchConnectivityState($new_state, $deadline) |
||||
{ |
||||
return $this->getNext()->watchConnectivityState($new_state, $deadline); |
||||
} |
||||
|
||||
public function getConnectivityState($try_to_connect = false) |
||||
{ |
||||
return $this->getNext()->getConnectivityState($try_to_connect); |
||||
} |
||||
|
||||
public function close() |
||||
{ |
||||
return $this->getNext()->close(); |
||||
} |
||||
} |
||||
@ -0,0 +1,427 @@ |
||||
<?php |
||||
/* |
||||
* |
||||
* Copyright 2018 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. |
||||
* |
||||
*/ |
||||
/** |
||||
* Interface exported by the server. |
||||
*/ |
||||
require_once(dirname(__FILE__).'/../../lib/Grpc/BaseStub.php'); |
||||
require_once(dirname(__FILE__).'/../../lib/Grpc/AbstractCall.php'); |
||||
require_once(dirname(__FILE__).'/../../lib/Grpc/UnaryCall.php'); |
||||
require_once(dirname(__FILE__).'/../../lib/Grpc/ClientStreamingCall.php'); |
||||
require_once(dirname(__FILE__).'/../../lib/Grpc/Interceptor.php'); |
||||
require_once(dirname(__FILE__).'/../../lib/Grpc/Internal/InterceptorChannel.php'); |
||||
|
||||
class SimpleRequest |
||||
{ |
||||
private $data; |
||||
public function __construct($data) |
||||
{ |
||||
$this->data = $data; |
||||
} |
||||
public function setData($data) |
||||
{ |
||||
$this->data = $data; |
||||
} |
||||
public function serializeToString() |
||||
{ |
||||
return $this->data; |
||||
} |
||||
} |
||||
|
||||
class InterceptorClient extends Grpc\BaseStub |
||||
{ |
||||
|
||||
/** |
||||
* @param string $hostname hostname |
||||
* @param array $opts channel options |
||||
* @param Channel|InterceptorChannel $channel (optional) re-use channel object |
||||
*/ |
||||
public function __construct($hostname, $opts, $channel = null) |
||||
{ |
||||
parent::__construct($hostname, $opts, $channel); |
||||
} |
||||
|
||||
/** |
||||
* A simple RPC. |
||||
* @param \Routeguide\Point $argument input argument |
||||
* @param array $metadata metadata |
||||
* @param array $options call options |
||||
*/ |
||||
public function UnaryCall( |
||||
SimpleRequest $argument, |
||||
$metadata = [], |
||||
$options = [] |
||||
) { |
||||
return $this->_simpleRequest( |
||||
'/dummy_method', |
||||
$argument, |
||||
[], |
||||
$metadata, |
||||
$options |
||||
); |
||||
} |
||||
|
||||
/** |
||||
* A client-to-server streaming RPC. |
||||
* @param array $metadata metadata |
||||
* @param array $options call options |
||||
*/ |
||||
public function StreamCall( |
||||
$metadata = [], |
||||
$options = [] |
||||
) { |
||||
return $this->_clientStreamRequest('/dummy_method', [], $metadata, $options); |
||||
} |
||||
} |
||||
|
||||
|
||||
class ChangeMetadataInterceptor extends Grpc\Interceptor |
||||
{ |
||||
public function interceptUnaryUnary($method, |
||||
$argument, |
||||
array $metadata = [], |
||||
array $options = [], |
||||
$continuation) |
||||
{ |
||||
$metadata["foo"] = array('interceptor_from_unary_request'); |
||||
return $continuation($method, $argument, $metadata, $options); |
||||
} |
||||
public function interceptStreamUnary($method, array $metadata = [], array $options = [], $continuation) |
||||
{ |
||||
$metadata["foo"] = array('interceptor_from_stream_request'); |
||||
return $continuation($method, $metadata, $options); |
||||
} |
||||
} |
||||
|
||||
class ChangeMetadataInterceptor2 extends Grpc\Interceptor |
||||
{ |
||||
public function interceptUnaryUnary($method, |
||||
$argument, |
||||
array $metadata = [], |
||||
array $options = [], |
||||
$continuation) |
||||
{ |
||||
if (array_key_exists('foo', $metadata)) { |
||||
$metadata['bar'] = array('ChangeMetadataInterceptor should be executed first'); |
||||
} else { |
||||
$metadata["bar"] = array('interceptor_from_unary_request'); |
||||
} |
||||
return $continuation($method, $argument, $metadata, $options); |
||||
} |
||||
public function interceptStreamUnary($method, |
||||
array $metadata = [], |
||||
array $options = [], |
||||
$continuation) |
||||
{ |
||||
if (array_key_exists('foo', $metadata)) { |
||||
$metadata['bar'] = array('ChangeMetadataInterceptor should be executed first'); |
||||
} else { |
||||
$metadata["bar"] = array('interceptor_from_stream_request'); |
||||
} |
||||
return $continuation($method, $metadata, $options); |
||||
} |
||||
} |
||||
|
||||
class ChangeRequestCall |
||||
{ |
||||
private $call; |
||||
|
||||
public function __construct($call) |
||||
{ |
||||
$this->call = $call; |
||||
} |
||||
public function getCall() |
||||
{ |
||||
return $this->call; |
||||
} |
||||
|
||||
public function write($request) |
||||
{ |
||||
$request->setData('intercepted_stream_request'); |
||||
$this->getCall()->write($request); |
||||
} |
||||
|
||||
public function wait() |
||||
{ |
||||
return $this->getCall()->wait(); |
||||
} |
||||
} |
||||
|
||||
class ChangeRequestInterceptor extends Grpc\Interceptor |
||||
{ |
||||
public function interceptUnaryUnary($method, |
||||
$argument, |
||||
array $metadata = [], |
||||
array $options = [], |
||||
$continuation) |
||||
{ |
||||
$argument->setData('intercepted_unary_request'); |
||||
return $continuation($method, $argument, $metadata, $options); |
||||
} |
||||
public function interceptStreamUnary($method, array $metadata = [], array $options = [], $continuation) |
||||
{ |
||||
return new ChangeRequestCall( |
||||
$continuation($method, $metadata, $options) |
||||
); |
||||
} |
||||
} |
||||
|
||||
class StopCallInterceptor extends Grpc\Interceptor |
||||
{ |
||||
public function interceptUnaryUnary($method, |
||||
$argument, |
||||
array $metadata = [], |
||||
array $options = [], |
||||
$continuation) |
||||
{ |
||||
$metadata["foo"] = array('interceptor_from_request_response'); |
||||
} |
||||
public function interceptStreamUnary($method, |
||||
array $metadata = [], |
||||
array $options = [], |
||||
$continuation) |
||||
{ |
||||
$metadata["foo"] = array('interceptor_from_request_response'); |
||||
} |
||||
} |
||||
|
||||
class InterceptorTest extends PHPUnit_Framework_TestCase |
||||
{ |
||||
public function setUp() |
||||
{ |
||||
$this->server = new Grpc\Server([]); |
||||
$this->port = $this->server->addHttp2Port('0.0.0.0:0'); |
||||
$this->channel = new Grpc\Channel('localhost:'.$this->port, ['credentials' => Grpc\ChannelCredentials::createInsecure()]); |
||||
$this->server->start(); |
||||
} |
||||
|
||||
public function tearDown() |
||||
{ |
||||
$this->channel->close(); |
||||
} |
||||
|
||||
|
||||
public function testClientChangeMetadataOneInterceptor() |
||||
{ |
||||
$req_text = 'client_request'; |
||||
$channel_matadata_interceptor = new ChangeMetadataInterceptor(); |
||||
$intercept_channel = Grpc\Interceptor::intercept($this->channel, $channel_matadata_interceptor); |
||||
echo "create Client\n"; |
||||
$client = new InterceptorClient('localhost:'.$this->port, [ |
||||
'credentials' => Grpc\ChannelCredentials::createInsecure(), |
||||
], $intercept_channel); |
||||
echo "create Call\n"; |
||||
$req = new SimpleRequest($req_text); |
||||
echo "Call created\n"; |
||||
$unary_call = $client->UnaryCall($req); |
||||
echo "start call\n"; |
||||
$event = $this->server->requestCall(); |
||||
$this->assertSame('/dummy_method', $event->method); |
||||
$this->assertSame(['interceptor_from_unary_request'], $event->metadata['foo']); |
||||
|
||||
$stream_call = $client->StreamCall(); |
||||
$stream_call->write($req); |
||||
$event = $this->server->requestCall(); |
||||
$this->assertSame('/dummy_method', $event->method); |
||||
$this->assertSame(['interceptor_from_stream_request'], $event->metadata['foo']); |
||||
|
||||
unset($unary_call); |
||||
unset($stream_call); |
||||
unset($server_call); |
||||
} |
||||
|
||||
public function testClientChangeMetadataTwoInterceptor() |
||||
{ |
||||
$req_text = 'client_request'; |
||||
$channel_matadata_interceptor = new ChangeMetadataInterceptor(); |
||||
$channel_matadata_intercepto2 = new ChangeMetadataInterceptor2(); |
||||
// test intercept separately. |
||||
$intercept_channel1 = Grpc\Interceptor::intercept($this->channel, $channel_matadata_interceptor); |
||||
$intercept_channel2 = Grpc\Interceptor::intercept($intercept_channel1, $channel_matadata_intercepto2); |
||||
$client = new InterceptorClient('localhost:'.$this->port, [ |
||||
'credentials' => Grpc\ChannelCredentials::createInsecure(), |
||||
], $intercept_channel2); |
||||
|
||||
$req = new SimpleRequest($req_text); |
||||
$unary_call = $client->UnaryCall($req); |
||||
$event = $this->server->requestCall(); |
||||
$this->assertSame('/dummy_method', $event->method); |
||||
$this->assertSame(['interceptor_from_unary_request'], $event->metadata['foo']); |
||||
$this->assertSame(['interceptor_from_unary_request'], $event->metadata['bar']); |
||||
|
||||
$stream_call = $client->StreamCall(); |
||||
$stream_call->write($req); |
||||
$event = $this->server->requestCall(); |
||||
$this->assertSame('/dummy_method', $event->method); |
||||
$this->assertSame(['interceptor_from_stream_request'], $event->metadata['foo']); |
||||
$this->assertSame(['interceptor_from_stream_request'], $event->metadata['bar']); |
||||
|
||||
unset($unary_call); |
||||
unset($stream_call); |
||||
unset($server_call); |
||||
|
||||
// test intercept by array. |
||||
$intercept_channel3 = Grpc\Interceptor::intercept($this->channel, |
||||
[$channel_matadata_intercepto2, $channel_matadata_interceptor]); |
||||
$client = new InterceptorClient('localhost:'.$this->port, [ |
||||
'credentials' => Grpc\ChannelCredentials::createInsecure(), |
||||
], $intercept_channel3); |
||||
|
||||
$req = new SimpleRequest($req_text); |
||||
$unary_call = $client->UnaryCall($req); |
||||
$event = $this->server->requestCall(); |
||||
$this->assertSame('/dummy_method', $event->method); |
||||
$this->assertSame(['interceptor_from_unary_request'], $event->metadata['foo']); |
||||
$this->assertSame(['interceptor_from_unary_request'], $event->metadata['bar']); |
||||
|
||||
$stream_call = $client->StreamCall(); |
||||
$stream_call->write($req); |
||||
$event = $this->server->requestCall(); |
||||
$this->assertSame('/dummy_method', $event->method); |
||||
$this->assertSame(['interceptor_from_stream_request'], $event->metadata['foo']); |
||||
$this->assertSame(['interceptor_from_stream_request'], $event->metadata['bar']); |
||||
|
||||
unset($unary_call); |
||||
unset($stream_call); |
||||
unset($server_call); |
||||
} |
||||
|
||||
public function testClientChangeRequestInterceptor() |
||||
{ |
||||
$req_text = 'client_request'; |
||||
$change_request_interceptor = new ChangeRequestInterceptor(); |
||||
$intercept_channel = Grpc\Interceptor::intercept($this->channel, |
||||
$change_request_interceptor); |
||||
$client = new InterceptorClient('localhost:'.$this->port, [ |
||||
'credentials' => Grpc\ChannelCredentials::createInsecure(), |
||||
], $intercept_channel); |
||||
|
||||
$req = new SimpleRequest($req_text); |
||||
$unary_call = $client->UnaryCall($req); |
||||
|
||||
$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' => '', |
||||
], |
||||
Grpc\OP_RECV_MESSAGE => true, |
||||
Grpc\OP_RECV_CLOSE_ON_SERVER => true, |
||||
]); |
||||
$this->assertSame('intercepted_unary_request', $event->message); |
||||
|
||||
$stream_call = $client->StreamCall(); |
||||
$stream_call->write($req); |
||||
$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' => '', |
||||
], |
||||
Grpc\OP_RECV_MESSAGE => true, |
||||
Grpc\OP_RECV_CLOSE_ON_SERVER => true, |
||||
]); |
||||
$this->assertSame('intercepted_stream_request', $event->message); |
||||
|
||||
unset($unary_call); |
||||
unset($stream_call); |
||||
unset($server_call); |
||||
} |
||||
|
||||
public function testClientChangeStopCallInterceptor() |
||||
{ |
||||
$req_text = 'client_request'; |
||||
$channel_request_interceptor = new StopCallInterceptor(); |
||||
$intercept_channel = Grpc\Interceptor::intercept($this->channel, |
||||
$channel_request_interceptor); |
||||
$client = new InterceptorClient('localhost:'.$this->port, [ |
||||
'credentials' => Grpc\ChannelCredentials::createInsecure(), |
||||
], $intercept_channel); |
||||
|
||||
$req = new SimpleRequest($req_text); |
||||
$unary_call = $client->UnaryCall($req); |
||||
$this->assertNull($unary_call); |
||||
|
||||
|
||||
$stream_call = $client->StreamCall(); |
||||
$this->assertNull($stream_call); |
||||
|
||||
unset($unary_call); |
||||
unset($stream_call); |
||||
unset($server_call); |
||||
} |
||||
|
||||
public function testGetInterceptorChannelConnectivityState() |
||||
{ |
||||
$channel = new Grpc\Channel( |
||||
'localhost:0', |
||||
['credentials' => Grpc\ChannelCredentials::createInsecure()] |
||||
); |
||||
$interceptor_channel = Grpc\Interceptor::intercept($channel, new Grpc\Interceptor()); |
||||
$state = $interceptor_channel->getConnectivityState(); |
||||
$this->assertEquals(0, $state); |
||||
$channel->close(); |
||||
} |
||||
|
||||
public function testInterceptorChannelWatchConnectivityState() |
||||
{ |
||||
$channel = new Grpc\Channel( |
||||
'localhost:0', |
||||
['credentials' => Grpc\ChannelCredentials::createInsecure()] |
||||
); |
||||
$interceptor_channel = Grpc\Interceptor::intercept($channel, new Grpc\Interceptor()); |
||||
$now = Grpc\Timeval::now(); |
||||
$deadline = $now->add(new Grpc\Timeval(100*1000)); |
||||
$state = $interceptor_channel->watchConnectivityState(1, $deadline); |
||||
$this->assertTrue($state); |
||||
unset($time); |
||||
unset($deadline); |
||||
$channel->close(); |
||||
} |
||||
|
||||
public function testInterceptorChannelClose() |
||||
{ |
||||
$channel = new Grpc\Channel( |
||||
'localhost:0', |
||||
['credentials' => Grpc\ChannelCredentials::createInsecure()] |
||||
); |
||||
$interceptor_channel = Grpc\Interceptor::intercept($channel, new Grpc\Interceptor()); |
||||
$this->assertNotNull($interceptor_channel); |
||||
$channel->close(); |
||||
} |
||||
|
||||
public function testInterceptorChannelGetTarget() |
||||
{ |
||||
$channel = new Grpc\Channel( |
||||
'localhost:8888', |
||||
['credentials' => Grpc\ChannelCredentials::createInsecure()] |
||||
); |
||||
$interceptor_channel = Grpc\Interceptor::intercept($channel, new Grpc\Interceptor()); |
||||
$target = $interceptor_channel->getTarget(); |
||||
$this->assertTrue(is_string($target)); |
||||
$channel->close(); |
||||
} |
||||
} |
||||
Loading…
Reference in new issue