xref: /aosp_15_r20/external/grpc-grpc/examples/python/cancellation/client.py (revision cc02d7e222339f7a4f6ba5f422e6413f4bd931f2) 
1# Copyright 2019 the gRPC authors.
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7#     http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14"""An example of cancelling requests in gRPC."""
15
16from __future__ import absolute_import
17from __future__ import division
18from __future__ import print_function
19
20import argparse
21import logging
22import signal
23import sys
24
25import grpc
26
27from examples.python.cancellation import hash_name_pb2
28from examples.python.cancellation import hash_name_pb2_grpc
29
30_DESCRIPTION = "A client for finding hashes similar to names."
31_LOGGER = logging.getLogger(__name__)
32
33
34def run_unary_client(server_target, name, ideal_distance):
35    with grpc.insecure_channel(server_target) as channel:
36        stub = hash_name_pb2_grpc.HashFinderStub(channel)
37        future = stub.Find.future(
38            hash_name_pb2.HashNameRequest(
39                desired_name=name, ideal_hamming_distance=ideal_distance
40            ),
41            wait_for_ready=True,
42        )
43
44        def cancel_request(unused_signum, unused_frame):
45            future.cancel()
46            sys.exit(0)
47
48        signal.signal(signal.SIGINT, cancel_request)
49        result = future.result()
50        print(result)
51
52
53def run_streaming_client(
54    server_target, name, ideal_distance, interesting_distance
55):
56    with grpc.insecure_channel(server_target) as channel:
57        stub = hash_name_pb2_grpc.HashFinderStub(channel)
58        result_generator = stub.FindRange(
59            hash_name_pb2.HashNameRequest(
60                desired_name=name,
61                ideal_hamming_distance=ideal_distance,
62                interesting_hamming_distance=interesting_distance,
63            ),
64            wait_for_ready=True,
65        )
66
67        def cancel_request(unused_signum, unused_frame):
68            result_generator.cancel()
69            sys.exit(0)
70
71        signal.signal(signal.SIGINT, cancel_request)
72        for result in result_generator:
73            print(result)
74
75
76def main():
77    parser = argparse.ArgumentParser(description=_DESCRIPTION)
78    parser.add_argument("name", type=str, help="The desired name.")
79    parser.add_argument(
80        "--ideal-distance",
81        default=0,
82        nargs="?",
83        type=int,
84        help="The desired Hamming distance.",
85    )
86    parser.add_argument(
87        "--server",
88        default="localhost:50051",
89        type=str,
90        nargs="?",
91        help="The host-port pair at which to reach the server.",
92    )
93    parser.add_argument(
94        "--show-inferior",
95        default=None,
96        type=int,
97        nargs="?",
98        help=(
99            "Also show candidates with a Hamming distance less than this value."
100        ),
101    )
102
103    args = parser.parse_args()
104    if args.show_inferior is not None:
105        run_streaming_client(
106            args.server, args.name, args.ideal_distance, args.show_inferior
107        )
108    else:
109        run_unary_client(args.server, args.name, args.ideal_distance)
110
111
112if __name__ == "__main__":
113    logging.basicConfig()
114    main()
115