Elep

<?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\Gcp;

/**
 * GcpExtensionChannel maintains an array of channels for certain API.
 */
class GcpExtensionChannel
{
    public $max_size;
    public $max_concurrent_streams_low_watermark;
    public $target;
    public $options;
    public $affinity_by_method;
    public $affinity_key_to_channel_ref;
    public $channel_refs;
    public $credentials;
    public $affinity_conf;

private $is_closed;

/**
     * @return array An array of ChannelRefs created for certain API.
     */
    public function getChannelRefs()
    {
        return $this->channel_refs;
    }

/**
     * @param string $hostname
     * @param array $opts Options to create a \Grpc\Channel and affinity config
     */
    public function __construct($hostname = null, $opts = array())
    {
        if ($hostname == null || !is_array($opts)) {
            throw new \InvalidArgumentException("Expected hostname is empty");
        }
        $this->max_size = 10;
        $this->max_concurrent_streams_low_watermark = 100;
        if (isset($opts['affinity_conf'])) {
            if (isset($opts['affinity_conf']['channelPool'])) {
                if (isset($opts['affinity_conf']['channelPool']['maxSize'])) {
                    $this->max_size = $opts['affinity_conf']['channelPool']['maxSize'];
                }
                if (isset($opts['affinity_conf']['channelPool']['maxConcurrentStreamsLowWatermark'])) {
                    $this->max_concurrent_streams_low_watermark =
                        $opts['affinity_conf']['channelPool']['maxConcurrentStreamsLowWatermark'];
                }
            }
            $this->affinity_by_method = $opts['affinity_conf']['affinity_by_method'];
            $this->affinity_conf = $opts['affinity_conf'];
        }
        $this->target = $hostname;
        $this->affinity_key_to_channel_ref = array();
        $this->channel_refs = array();
        $this->updateOpts($opts);
        // Initiate a Grpc\Channel at the beginning in order to keep the same
        // behavior as the Grpc.
        $channel_ref = $this->getChannelRef();
        $channel_ref->getRealChannel($this->credentials);
    }

/**
     * @param array $opts Options to create a \Grpc\Channel
     */
    public function updateOpts($opts)
    {
        if (isset($opts['credentials'])) {
            $this->credentials = $opts['credentials'];
        }
        unset($opts['affinity_conf']);
        unset($opts['credentials']);
        $this->options = $opts;
        $this->is_closed = false;
    }

/**
     * Bind the ChannelRef with the affinity key. This is a private method.
     *
     * @param ChannelRef $channel_ref
     * @param string $affinity_key
     *
     * @return ChannelRef
     */
    public function bind($channel_ref, $affinity_key)
    {
        if (!array_key_exists($affinity_key, $this->affinity_key_to_channel_ref)) {
            $this->affinity_key_to_channel_ref[$affinity_key] = $channel_ref;
        }
        $channel_ref->affinityRefIncr();
        return $channel_ref;
    }

/**
     * Unbind the affinity key. This is a private method.
     *
     * @param string $affinity_key
     *
     * @return ChannelRef
     */
    public function unbind($affinity_key)
    {
        $channel_ref = null;
        if (array_key_exists($affinity_key, $this->affinity_key_to_channel_ref)) {
            $channel_ref =  $this->affinity_key_to_channel_ref[$affinity_key];
            $channel_ref->affinityRefDecr();
        }
        unset($this->affinity_key_to_channel_ref[$affinity_key]);
        return $channel_ref;
    }

public function cmp_by_active_stream_ref($a, $b)
    {
        return $a->getActiveStreamRef() - $b->getActiveStreamRef();
    }

/**
     * Pick or create a ChannelRef from the pool by affinity key.
     *
     * @param string $affinity_key
     *
     * @return ChannelRef
     */
    public function getChannelRef($affinity_key = null)
    {
        if ($affinity_key) {
            if (array_key_exists($affinity_key, $this->affinity_key_to_channel_ref)) {
                return $this->affinity_key_to_channel_ref[$affinity_key];
            }
            return $this->getChannelRef();
        }
        usort($this->channel_refs, array($this, 'cmp_by_active_stream_ref'));

if (count($this->channel_refs) > 0 && $this->channel_refs[0]->getActiveStreamRef() <
            $this->max_concurrent_streams_low_watermark) {
            return $this->channel_refs[0];
        }
        $num_channel_refs = count($this->channel_refs);
        if ($num_channel_refs < $this->max_size) {
            // grpc_target_persist_bound stands for how many channels can be persisted for
            // the same target in the C extension. It is possible that the user use the pure
            // gRPC and this GCP extension at the same time, which share the same target. In this case
            // pure gRPC channel may occupy positions in C extension, which deletes some channels created
            // by this GCP extension.
            // If that happens, it won't cause the script failure because we saves all arguments for creating
            // a channel instead of a channel itself. If we watch to fetch a GCP channel already deleted,
            // it will create a new channel. The only cons is the latency of the first RPC will high because
            // it will establish the connection again.
            if (!isset($this->options['grpc_target_persist_bound']) ||
                $this->options['grpc_target_persist_bound'] < $this->max_size) {
                $this->options['grpc_target_persist_bound'] = $this->max_size;
            }
            $cur_opts = array_merge($this->options,
                ['grpc_gcp_channel_id' => $num_channel_refs]);
            $channel_ref = new ChannelRef($this->target, $num_channel_refs, $cur_opts);
            array_unshift($this->channel_refs, $channel_ref);
        }
        return $this->channel_refs[0];
    }

/**
     * Get the connectivity state of the channel
     *
     * @param bool $try_to_connect try to connect on the channel
     *
     * @return int The grpc connectivity state
     * @throws \InvalidArgumentException
     */
    public function getConnectivityState($try_to_connect = false)
    {
        // Since getRealChannel is creating a PHP Channel object. However in gRPC, when a Channel
        // object is closed, we only mark this Object to be invalid. Thus, we need a global variable
        // to mark whether this GCPExtensionChannel is close or not.
        if ($this->is_closed) {
            throw new \RuntimeException("Channel has already been closed");
        }
        $ready = 0;
        $idle = 0;
        $connecting = 0;
        $transient_failure = 0;
        $shutdown = 0;
        foreach ($this->channel_refs as $channel_ref) {
            $state = $channel_ref->getRealChannel($this->credentials)->getConnectivityState($try_to_connect);
            switch ($state) {
                case \Grpc\CHANNEL_READY:
                    $ready += 1;
                    break 2;
                case \Grpc\CHANNEL_FATAL_FAILURE:
                    $shutdown += 1;
                    break;
                case \Grpc\CHANNEL_CONNECTING:
                    $connecting += 1;
                    break;
                case \Grpc\CHANNEL_TRANSIENT_FAILURE:
                    $transient_failure += 1;
                    break;
                case \Grpc\CHANNEL_IDLE:
                    $idle += 1;
                    break;
            }
        }
        if ($ready > 0) {
            return \Grpc\CHANNEL_READY;
        } elseif ($idle > 0) {
            return \Grpc\CHANNEL_IDLE;
        } elseif ($connecting > 0) {
            return \Grpc\CHANNEL_CONNECTING;
        } elseif ($transient_failure > 0) {
            return \Grpc\CHANNEL_TRANSIENT_FAILURE;
        } elseif ($shutdown > 0) {
            return \Grpc\CHANNEL_SHUTDOWN;
        }
    }

/**
     * Watch the connectivity state of the channel until it changed
     *
     * @param int     $last_state   The previous connectivity state of the channel
     * @param Timeval $deadline_obj The deadline this function should wait until
     *
     * @return bool If the connectivity state changes from last_state
     *              before deadline
     * @throws \InvalidArgumentException
     */
    public function watchConnectivityState($last_state, $deadline_obj = null)
    {
        if ($deadline_obj == null || !is_a($deadline_obj, '\Grpc\Timeval')) {
            throw new \InvalidArgumentException("");
        }
        // Since getRealChannel is creating a PHP Channel object. However in gRPC, when a Channel
        // object is closed, we only mark this Object to be invalid. Thus, we need a global variable
        // to mark whether this GCPExtensionChannel is close or not.
        if ($this->is_closed) {
            throw new \RuntimeException("Channel has already been closed");
        }
        $state = 0;
        foreach ($this->channel_refs as $channel_ref) {
            $state = $channel_ref->getRealChannel($this->credentials)->watchConnectivityState($last_state, $deadline_obj);
        }
        return $state;
    }

/**
     * Get the endpoint this call/stream is connected to
     *
     * @return string The URI of the endpoint
     */
    public function getTarget()
    {
        return $this->target;
    }

/**
     * Close the channel
     */
    public function close()
    {
        foreach ($this->channel_refs as $channel_ref) {
            $channel_ref->getRealChannel($this->credentials)->close();
        }
        $this->is_closed = true;
    }
}