# Copyright 2016 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.
"""Test making many calls and immediately cancelling most of them."""

import threading
import unittest

from grpc._cython import cygrpc
from grpc.framework.foundation import logging_pool
from tests.unit.framework.common import test_constants

_INFINITE_FUTURE = cygrpc.Timespec(float('+inf'))
_EMPTY_FLAGS = 0
_EMPTY_METADATA = ()

_SERVER_SHUTDOWN_TAG = 'server_shutdown'
_REQUEST_CALL_TAG = 'request_call'
_RECEIVE_CLOSE_ON_SERVER_TAG = 'receive_close_on_server'
_RECEIVE_MESSAGE_TAG = 'receive_message'
_SERVER_COMPLETE_CALL_TAG = 'server_complete_call'

_SUCCESS_CALL_FRACTION = 1.0 / 8.0


class _State(object):

    def __init__(self):
        self.condition = threading.Condition()
        self.handlers_released = False
        self.parked_handlers = 0
        self.handled_rpcs = 0


def _is_cancellation_event(event):
    return (event.tag is _RECEIVE_CLOSE_ON_SERVER_TAG and
            event.batch_operations[0].received_cancelled)


class _Handler(object):

    def __init__(self, state, completion_queue, rpc_event):
        self._state = state
        self._lock = threading.Lock()
        self._completion_queue = completion_queue
        self._call = rpc_event.operation_call

    def __call__(self):
        with self._state.condition:
            self._state.parked_handlers += 1
            if self._state.parked_handlers == test_constants.THREAD_CONCURRENCY:
                self._state.condition.notify_all()
            while not self._state.handlers_released:
                self._state.condition.wait()

        with self._lock:
            self._call.start_server_batch(
                (cygrpc.operation_receive_close_on_server(_EMPTY_FLAGS),),
                _RECEIVE_CLOSE_ON_SERVER_TAG)
            self._call.start_server_batch(
                (cygrpc.operation_receive_message(_EMPTY_FLAGS),),
                _RECEIVE_MESSAGE_TAG)
        first_event = self._completion_queue.poll()
        if _is_cancellation_event(first_event):
            self._completion_queue.poll()
        else:
            with self._lock:
                operations = (
                    cygrpc.operation_send_initial_metadata(_EMPTY_METADATA,
                                                           _EMPTY_FLAGS),
                    cygrpc.operation_send_message(b'\x79\x57', _EMPTY_FLAGS),
                    cygrpc.operation_send_status_from_server(
                        _EMPTY_METADATA, cygrpc.StatusCode.ok, b'test details!',
                        _EMPTY_FLAGS),)
                self._call.start_server_batch(operations,