mirror of
https://github.com/php-win-ext/grpc.git
synced 2026-03-24 09:02:15 +01:00
de6ed9ba9f
- Switched from yapf to black - Reconfigure isort for black - Resolve black/pylint idiosyncrasies Note: I used `--experimental-string-processing` because black was producing "implicit string concatenation", similar to what described here: https://github.com/psf/black/issues/1837. While currently this feature is experimental, it will be enabled by default: https://github.com/psf/black/issues/2188. After running black with the new string processing so that the generated code merges these `"hello" " world"` strings concatenations, then I removed `--experimental-string-processing` for stability, and regenerated the code again. To the reviewer: don't even try to open "Files Changed" tab 😄 It's better to review commit-by-commit, and ignore `run black and isort`.
130 lines
4.6 KiB
Python
130 lines
4.6 KiB
Python
# Copyright 2020 The 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.
|
|
|
|
from concurrent.futures import ThreadPoolExecutor
|
|
import logging
|
|
import threading
|
|
from typing import Iterator
|
|
|
|
import grpc
|
|
|
|
import phone_pb2
|
|
import phone_pb2_grpc
|
|
|
|
|
|
class CallMaker:
|
|
def __init__(
|
|
self,
|
|
executor: ThreadPoolExecutor,
|
|
channel: grpc.Channel,
|
|
phone_number: str,
|
|
) -> None:
|
|
self._executor = executor
|
|
self._channel = channel
|
|
self._stub = phone_pb2_grpc.PhoneStub(self._channel)
|
|
self._phone_number = phone_number
|
|
self._session_id = None
|
|
self._audio_session_link = None
|
|
self._call_state = None
|
|
self._peer_responded = threading.Event()
|
|
self._call_finished = threading.Event()
|
|
self._consumer_future = None
|
|
|
|
def _response_watcher(
|
|
self, response_iterator: Iterator[phone_pb2.StreamCallResponse]
|
|
) -> None:
|
|
try:
|
|
for response in response_iterator:
|
|
# NOTE: All fields in Proto3 are optional. This is the recommended way
|
|
# to check if a field is present or not, or to exam which one-of field is
|
|
# fulfilled by this message.
|
|
if response.HasField("call_info"):
|
|
self._on_call_info(response.call_info)
|
|
elif response.HasField("call_state"):
|
|
self._on_call_state(response.call_state.state)
|
|
else:
|
|
raise RuntimeError(
|
|
"Received StreamCallResponse without call_info and"
|
|
" call_state"
|
|
)
|
|
except Exception as e:
|
|
self._peer_responded.set()
|
|
raise
|
|
|
|
def _on_call_info(self, call_info: phone_pb2.CallInfo) -> None:
|
|
self._session_id = call_info.session_id
|
|
self._audio_session_link = call_info.media
|
|
|
|
def _on_call_state(self, call_state: phone_pb2.CallState.State) -> None:
|
|
logging.info(
|
|
"Call toward [%s] enters [%s] state",
|
|
self._phone_number,
|
|
phone_pb2.CallState.State.Name(call_state),
|
|
)
|
|
self._call_state = call_state
|
|
if call_state == phone_pb2.CallState.State.ACTIVE:
|
|
self._peer_responded.set()
|
|
if call_state == phone_pb2.CallState.State.ENDED:
|
|
self._peer_responded.set()
|
|
self._call_finished.set()
|
|
|
|
def call(self) -> None:
|
|
request = phone_pb2.StreamCallRequest()
|
|
request.phone_number = self._phone_number
|
|
response_iterator = self._stub.StreamCall(iter((request,)))
|
|
# Instead of consuming the response on current thread, spawn a consumption thread.
|
|
self._consumer_future = self._executor.submit(
|
|
self._response_watcher, response_iterator
|
|
)
|
|
|
|
def wait_peer(self) -> bool:
|
|
logging.info("Waiting for peer to connect [%s]...", self._phone_number)
|
|
self._peer_responded.wait(timeout=None)
|
|
if self._consumer_future.done():
|
|
# If the future raises, forwards the exception here
|
|
self._consumer_future.result()
|
|
return self._call_state == phone_pb2.CallState.State.ACTIVE
|
|
|
|
def audio_session(self) -> None:
|
|
assert self._audio_session_link is not None
|
|
logging.info("Consuming audio resource [%s]", self._audio_session_link)
|
|
self._call_finished.wait(timeout=None)
|
|
logging.info("Audio session finished [%s]", self._audio_session_link)
|
|
|
|
|
|
def process_call(
|
|
executor: ThreadPoolExecutor, channel: grpc.Channel, phone_number: str
|
|
) -> None:
|
|
call_maker = CallMaker(executor, channel, phone_number)
|
|
call_maker.call()
|
|
if call_maker.wait_peer():
|
|
call_maker.audio_session()
|
|
logging.info("Call finished!")
|
|
else:
|
|
logging.info("Call failed: peer didn't answer")
|
|
|
|
|
|
def run():
|
|
executor = ThreadPoolExecutor()
|
|
with grpc.insecure_channel("localhost:50051") as channel:
|
|
future = executor.submit(
|
|
process_call, executor, channel, "555-0100-XXXX"
|
|
)
|
|
future.result()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
logging.basicConfig(level=logging.INFO)
|
|
run()
|