Viewing file:  RpcServer.php (5.56 KB) -rw-r--r--
Select action/file-type:
 (+) |  (+) |  (+) | Code (+) | Session (+) |  (+) | SDB (+) |  (+) |  (+) |  (+) |  (+) |  (+) |
<?php
/*
*
* Copyright 2020 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 an experimental and incomplete implementation of gRPC server
* for PHP. APIs are _definitely_ going to be changed.
*
* DO NOT USE in production.
*/
/**
* Class RpcServer
* @package Grpc
*/
class RpcServer extends Server
{
protected $call;
// [ <String method_full_path> => [
// 'service' => <Object service>,
// 'method' => <String method_name>,
// 'request' => <Object request>,
// ] ]
protected $paths_map;
private function waitForNextEvent() {
return $this->requestCall();
}
private function loadRequest($request) {
if (!$this->call) {
throw new Exception("serverCall is not ready");
}
$event = $this->call->startBatch([
OP_RECV_MESSAGE => true,
]);
if (!$event->message) {
throw new Exception("Did not receive a proper message");
}
$request->mergeFromString($event->message);
return $request;
}
protected function sendOkResponse($response) {
if (!$this->call) {
throw new Exception("serverCall is not ready");
}
$this->call->startBatch([
OP_SEND_INITIAL_METADATA => [],
OP_SEND_MESSAGE => ['message' =>
$response->serializeToString()],
OP_SEND_STATUS_FROM_SERVER => [
'metadata' => [],
'code' => STATUS_OK,
'details' => 'OK',
],
OP_RECV_CLOSE_ON_SERVER => true,
]);
}
/**
* Add a service to this server
*
* @param Object $service The service to be added
*/
public function handle($service) {
$rf = new \ReflectionClass($service);
// If input does not have a parent class, which should be the
// generated stub, don't proceeed. This might change in the
// future.
if (!$rf->getParentClass()) return;
// The input class name needs to match the service name
$service_name = $rf->getName();
$namespace = $rf->getParentClass()->getNamespaceName();
$prefix = "";
if ($namespace) {
$parts = explode("\\", $namespace);
foreach ($parts as $part) {
$prefix .= lcfirst($part) . ".";
}
}
$base_path = "/" . $prefix . $service_name;
// Right now, assume all the methods in the class are RPC method
// implementations. Might change in the future.
$methods = $rf->getMethods();
foreach ($methods as $method) {
$method_name = $method->getName();
$full_path = $base_path . "/" . ucfirst($method_name);
$method_params = $method->getParameters();
// RPC should have exactly 1 request param
if (count($method_params) != 1) continue;
$request_param = $method_params[0];
// Method implementation must have type hint for request param
if (!$request_param->getType()) continue;
$request_type = $request_param->getType()->getName();
// $full_path needs to match the incoming event->method
// from requestCall() for us to know how to handle the request
$this->paths_map[$full_path] = [
'service' => $service,
'method' => $method_name,
'request' => new $request_type(),
];
}
}
public function run() {
$this->start();
while (true) {
// This blocks until the server receives a request
$event = $this->waitForNextEvent();
if (!$event) {
throw new Exception(
"Unexpected error: server->waitForNextEvent delivers"
. " an empty event");
}
if (!$event->call) {
throw new Exception(
"Unexpected error: server->waitForNextEvent delivers"
. " an event without a call");
}
$this->call = $event->call;
$full_path = $event->method;
// TODO: Can send a proper UNIMPLEMENTED response in the future
if (!array_key_exists($full_path, $this->paths_map)) continue;
$service = $this->paths_map[$full_path]['service'];
$method = $this->paths_map[$full_path]['method'];
$request = $this->paths_map[$full_path]['request'];
$request = $this->loadRequest($request);
if (!$request) {
throw new Exception("Unexpected error: fail to parse request");
}
if (!method_exists($service, $method)) {
// TODO: Can send a proper UNIMPLEMENTED response in the future
throw new Exception("Method not implemented");
}
// Dispatch to actual server logic
$response = $service->$method($request);
$this->sendOkResponse($response);
$this->call = null;
}
}
}
|