allow vendord

vendor/google/grpc-gcp/src/GcpExtensionChannel.php

@@ -0,0 +1,281 @@
+<?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;
+    }
+}