Python grpc 模块，insecure_channel() 实例源码

def create_grpc_channel(target, pem=None, opts=None):
    """Construct a grpc channel.

    Args:
        target: url of target include host:port
        pem: ssl/tls pem file as bytes
        opts: grpc channel options
                grpc.default_authority: default authority
                grpc.ssl_target_name_override: ssl target name override

    Returns:
        grpc channel

    """
    if pem is None:
        return grpc.insecure_channel(target, opts)
    else:
        creds = grpc.ssl_channel_credentials(pem)
        return grpc.secure_channel(target, creds, opts)

def check_payment(self):
        """
        Checks if the Lightning payment has been received for this invoice
        """
        if self.status == 'pending_invoice':
            return False

        channel = grpc.insecure_channel(settings.LND_RPCHOST)
        stub = lnrpc.LightningStub(channel)

        r_hash_base64 = self.r_hash.encode('utf-8')
        r_hash_bytes = str(codecs.decode(r_hash_base64, 'base64'))
        invoice_resp = stub.LookupInvoice(ln.PaymentHash(r_hash=r_hash_bytes))

        if invoice_resp.settled:
            # Payment complete
            self.status = 'complete'
            self.save()
            return True
        else:
            # Payment not received
            return False

def get_channel(self):
        '''
        Return a tiller channel
        '''
        tiller_ip = self._get_tiller_ip()
        tiller_port = self._get_tiller_port()
        try:
            return grpc.insecure_channel(
                '%s:%s' % (tiller_ip, tiller_port),
                options=[
                    ('grpc.max_send_message_length', MAX_MESSAGE_LENGTH),
                    ('grpc.max_receive_message_length', MAX_MESSAGE_LENGTH)
                ]
            )
        except Exception:
            raise ex.ChannelException()

def send(node,task,message = ""):
        try:
            channel = grpc.insecure_channel(node )
            taskType,task =task / SERVICE ,task % SERVICE
            if taskType == DESCOVERY:
                stub = grpc_pb2_grpc.DiscoveryStub(channel)
                if task ==EXCHANGENODE:
                    response = stub.ExchangeNode(grpc_pb2.Node(number = len(Node.__Nodes),ipport = Node.getNodesList() ))
                    for node in response.ipport :
                        Node.__Nodes.add(node)



            elif taskType == SYNCHRONIZATION:
                stub = grpc_pb2_grpc.SynchronizationStub(channel)
                synchronization.Task(stub,task,message)

        except Exception as e :
            Node.delNode(node)

        return

def run():
  protoConfig = ProtoConfig.getConfig()
  channel = grpc.insecure_channel('localhost:%s' % protoConfig.ports.frontEndPort)
  stub = sensors_pb2.FrontEndStub(channel)
  rfBlasterRequestOn = sensors_pb2.SendToRfBlasterRequest(
      button=2,
      on=True
  )
  rfBlasterRequestOff = sensors_pb2.SendToRfBlasterRequest(
      button=2,
      on=False
  )
  stub.SendToRfBlaster(rfBlasterRequestOn)
  sleep(1)
  stub.SendToRfBlaster(rfBlasterRequestOff)
  exit()

def run():
  protoConfig = ProtoConfig.getConfig()
  channel = grpc.insecure_channel('localhost:%s' % protoConfig.ports.pushFrontEndPort)
  stub = sensors_pb2.PushFrontEndStub(channel)

  subscribeRequest = sensors_pb2.SubscribeRequest(
    status=sensors_pb2.SubscribeRequest.SUBSCRIBE,
    username="peridot"
  )
  subscribeReply = stub.Subscribe(subscribeRequest)
  start_index = subscribeReply.start_index or 0
  print(subscribeReply)
  req = sensors_pb2.GetButtonPressedRequest(
    index=subscribeReply.start_index)
  while True:
    for event in stub.StreamButtonPressed(req):
      req.index += 1
      print('index: %s, event: %s' % (req.index, event))
    time.sleep(1)

def test_stream_max_metrics_buffer():
    sys.stdout = ThreadPrinter()
    sys.argv = ["", '{"LogLevel": 1, "PingTimeoutDuration": 5000}']
    col = MockStreamCollector("MyStreamCollector", 99)
    col.start()
    t_end = time.time() + 5
    # wait for our collector to print its preamble
    while len(sys.stdout.lines) == 0 and time.time() < t_end:
        time.sleep(.1)
    resp = json.loads(sys.stdout.lines[0])
    client = StreamCollectorStub(
        grpc.insecure_channel(resp["ListenAddress"]))
    metric = snap.Metric(
        namespace=[snap.NamespaceElement(value="intel"),
                   snap.NamespaceElement(value="streaming"),
                   snap.NamespaceElement(value="random"),
                   snap.NamespaceElement(value="int")],
        version=1,
        config={"max-metrics-buffer": 5},
        unit="some unit",
        description="some description")
    col_arg = CollectArg(metric).pb
    mtr = iter([col_arg])
    metrics = client.StreamMetrics(mtr)
    start_waiting_for_new_metric = time.time()
    a = next(metrics)
    retrieve_metric_time = time.time()
    assert round(retrieve_metric_time - start_waiting_for_new_metric) == 5
    assert len(a.Metrics_Reply.metrics) == 5
    col.stop()

def test_get_config_policy():
    sys.stdout = ThreadPrinter()
    sys.argv = ["", '{"LogLevel": 1, "PingTimeoutDuration": 5000}']
    col = MockStreamCollector("MyStreamCollector", 99)
    col.start()
    t_end = time.time() + 5
    # wait for our collector to print its preamble
    while len(sys.stdout.lines) == 0 and time.time() < t_end:
        time.sleep(.1)
    resp = json.loads(sys.stdout.lines[0])
    client = StreamCollectorStub(
        grpc.insecure_channel(resp["ListenAddress"]))
    reply = client.GetConfigPolicy(Empty())
    assert reply.error == ""
    assert reply.string_policy["intel.streaming.random"].rules["password"].default == "pass"
    col.stop()

def processor_client():
    """Returns a client (grpc) fixture that is passed into processor
    tests """
    sys.stdout = ThreadPrinter()
    sys.argv = ["", '{}']
    proc = MockProcessor("MyProcessor", 1)
    proc.start()
    t_end = time.time() + 5
    # wait for our collector to print its preamble
    while len(sys.stdout.lines) == 0 and time.time() < t_end:
        time.sleep(.1)
    resp = json.loads(sys.stdout.lines[0])
    client = ProcessorStub(
        grpc.insecure_channel(resp["ListenAddress"]))
    yield client
    proc.stop()

def _get_values(self, pause_secs=2.0):
        channel = grpc.insecure_channel(self.hostname)
        stub = LocationServiceStub(channel)
        while not self.stopped:
            logger.info("Connecting to gRPC server at {0}...".format(self.hostname))
            try:
                client_info = ClientInfo(info="{0} client".format(socket.gethostname()))
                server_info = stub.registerClient(client_info)
            except BaseException as e:
                logger.error("Failed to connect to gRPC server at {0} [{1}]".format(self.hostname, e))
                time.sleep(pause_secs)
                continue

            logger.info("Connected to gRPC server at {0} [{1}]".format(self.hostname, server_info.info))

            try:
                for val in stub.getLocations(client_info):
                    with self.value_lock:
                        self.__currval = copy.deepcopy(val)
                    self._mark_ready()
            except BaseException as e:
                logger.info("Disconnected from gRPC server at {0} [{1}]".format(self.hostname, e))
                time.sleep(pause_secs)

    # Non-blocking

def connect(self, endpoint):
        if self.connected:
            return
        try:
            self.log.info('insecurely-connecting', endpoint=endpoint)
            self.channel = grpc.insecure_channel(endpoint)
            self.connected = True
            self.log.info('insecurely-connected', endpoint=endpoint)
            return

        except _Rendezvous, e:
            if e.code() == grpc.StatusCode.UNAVAILABLE:
                self.log.info('grpc-endpoint-not-available')
            else:
                self.log.exception(e)

        except Exception, e:
            self.log.exception('cannot-connect', endpoint=endpoint)

def run():

    channel = grpc.insecure_channel('localhost:50055')

    # Test fetch the schema
    stub = schema_pb2.SchemaServiceStub(channel)
    res = stub.GetSchema(Empty())
    print '\nSchema:\n'
    for key in res.protos:
        print '%s %s file begins %s\n' % (30 * '~', key, (35 - len(key)) * '~')
        print res.protos[key]
        print '%s %s file ends %s' % (30 * '~', key, (37 - len(key)) * '~')
    for key in res.descriptors:
        print '%s -> descriptor of %d bytes' % (key, len(res.descriptors[key]))

    # Ping health state as an example
    stub = voltha_pb2.HealthServiceStub(channel)
    res = stub.GetHealthStatus(Empty())
    print '\nHealth state:', res.state

    # Try another API
    stub = voltha_pb2.ExampleServiceStub(channel)
    res = stub.ListAddresses(Empty())
    print '\nExample objects returned:\n', res.addresses

def run():
    channel = grpc.insecure_channel('localhost:50051')
    try:
        grpc.channel_ready_future(channel).result(timeout=10)
    except grpc.FutureTimeoutError:
        sys.exit('Error connecting to server')
    else:
        stub = users_service.UsersStub(channel)
        metadata = [('ip', '127.0.0.1')]
        response = stub.CreateUser(
            users_messages.CreateUserRequest(username='tom'),
            metadata=metadata,
        )
        if response:
            print("User created:", response.user.username)
        request = users_messages.GetUsersRequest(
            user=[users_messages.User(username="alexa", user_id=1),
                  users_messages.User(username="christie", user_id=1)]
        )
        response = stub.GetUsers(request, timeout=0.00001)
        for resp in response:
            print(resp)

def run():
    channel = grpc.insecure_channel('localhost:50051')
    try:
        grpc.channel_ready_future(channel).result(timeout=10)
    except grpc.FutureTimeoutError:
        sys.exit('Error connecting to server')
    else:
        stub = users_service.UsersStub(channel)
        metadata = [('ip', '127.0.0.1')]
        response = stub.CreateUser(
            users_messages.CreateUserRequest(username='tom'),
            metadata=metadata,
        )
        if response:
            print("User created:", response.user.username)
        request = users_messages.GetUsersRequest(
            user=[users_messages.User(username="alexa", user_id=1),
                  users_messages.User(username="christie", user_id=1)]
        )
        response = stub.GetUsers(request, timeout=0.00001)
        for resp in response:
            print(resp)

def run():
    channel = grpc.insecure_channel('localhost:50051')
    try:
        grpc.channel_ready_future(channel).result(timeout=10)
    except grpc.FutureTimeoutError:
        sys.exit('Error connecting to server')
    else:
        stub = users_service.UsersStub(channel)
        metadata = [('ip', '127.0.0.1')]
        response = stub.CreateUser(
            users_messages.CreateUserRequest(username='tom'),
            metadata=metadata,
        )
        if response:
            print("User created:", response.user.username)
        request = users_messages.GetUsersRequest(
            user=[users_messages.User(username="alexa", user_id=1),
                  users_messages.User(username="christie", user_id=1)]
        )
        response = stub.GetUsers(request)
        for resp in response:
            print(resp)

def tx_push(ctx, txblob):
    tx = None
    try:
        txbin = bytes(hstr2bin(txblob))
        pbdata = qrl_pb2.Transaction()
        pbdata.ParseFromString(txbin)
        tx = Transaction.from_pbdata(pbdata)
    except Exception as e:
        click.echo("tx blob is not valid")
        quit(1)

    tmp_json = tx.to_json()
    # FIXME: binary fields are represented in base64. Improve output
    print(tmp_json)
    if (len(tx.signature) == 0):
        click.echo('Signature missing')
        quit(1)

    channel = grpc.insecure_channel(ctx.obj.node_public_address)
    stub = qrl_pb2_grpc.PublicAPIStub(channel)
    pushTransactionReq = qrl_pb2.PushTransactionReq(transaction_signed=tx.pbdata)
    pushTransactionResp = stub.PushTransaction(pushTransactionReq, timeout=5)
    print(pushTransactionResp.some_response)

def setUpClass(cls):
        """Instantiate a server for tests"""
        # Avoid setting up a Riot API handler and a cache system
        MatchFetcher.riot_api_handler = mock.MagicMock()
        MatchFetcher.cache_manager = mock.MagicMock()

        cls.server, cls.service = start_server("123", 50002, 10)

        # Only initialize once the stub
        cls.channel = grpc.insecure_channel("localhost:50002")
        cls.stub = service_pb2.MatchFetcherStub(cls.channel)

def generate_invoice(self, user, article):
        """
        Generates a new invoice
        """
        assert self.status == 'pending_invoice', "Already generated invoice"
        channel = grpc.insecure_channel(settings.LND_RPCHOST)
        stub = lnrpc.LightningStub(channel)

        add_invoice_resp = stub.AddInvoice(ln.Invoice(value=settings.MIN_VIEW_AMOUNT, memo="User '{}' | ArticleId {}".format(user.username, article.id)))
        r_hash_base64 = codecs.encode(add_invoice_resp.r_hash, 'base64')
        self.r_hash = r_hash_base64.decode('utf-8')
        self.payment_request = add_invoice_resp.payment_request
        self.status = 'pending_payment'
        self.save()

def authenticate(self, request, signature, csrf_token, username=None):
        channel = grpc.insecure_channel(settings.LND_RPCHOST)
        stub = lnrpc.LightningStub(channel)

        verifymessage_resp = stub.VerifyMessage(ln.VerifyMessageRequest(msg=csrf_token, signature=signature))

        if not verifymessage_resp.valid:
            print "Invalid signature"
            return None

        pubkey = verifymessage_resp.pubkey
        # Try fetching an existing profile
        try:
            profile = Profile.objects.get(identity_pubkey=pubkey)
            return profile.user
        except Profile.DoesNotExist:
            # Create a new profile if they provided a username
            if len(username) > 3 and len(username) < 36:
                user = User(username=username)
                user.save()
                profile, created = Profile.objects.get_or_create(
                    user=user,
                    identity_pubkey=pubkey)
                assert created is True
                # TODO Auth them in
            else:
                raise Exception("No username provided")
        return user

def run():
    df = pd.DataFrame(columns=list('abc'), data=pd.np.random.rand(10, 3))
    channel = grpc.insecure_channel('localhost:50051')
    stub = skl_pb2.PredictionServiceStub(channel)
    print("-------------- Predict --------------")
    model_spec = model_pb2.ModelSpec(model_id='123abc')
    req = predict_pb2.PredictionRequest(model_spec=model_spec, input=pandas_to_proto(df))
    pred = stub.Predict(req)
    print(pred)

def __init__(self, endpoint):
        """
        Initializes a new instance of BblfshClient.

        :param endpoint: The address of the Babelfish server, \
                         for example "0.0.0.0:9432"
        :type endpoint: str
        """
        self._channel = grpc.insecure_channel(endpoint)
        self._stub = ProtocolServiceStub(self._channel)

def setUp(self):
        self._server = greeter_server._GreeterServer(greeter_server._GreeterService(), TEST_PORT)
        self._server.start()
        channel = grpc.insecure_channel('localhost:{port}'.format(port=TEST_PORT))
        self._client = helloworld_pb2_grpc.GreeterStub(channel)

def run():
    channel = grpc.insecure_channel('localhost:50051')
    stub = helloworld_pb2_grpc.GreeterStub(channel)
    response = stub.SayHello(helloworld_pb2.HelloRequest(name='you'))
    print("Greeter client received: " + response.message)

def build_transformer(self):
        logger.debug("client: build_transformer, server address: %s", self.server_address)
        if self.is_trans_built:
            logger.debug("client: build_transformer: transformer is already built")
            return
        options = [('grpc.max_send_message_length', -1), ('grpc.max_receive_message_length', -1)]
        channel = grpc.insecure_channel(self.server_address, options=options)
        if not is_channel_ready(channel):
            raise RuntimeError("gRPC channel is not ready...")
        self.RPC = hetr_pb2_grpc.HetrStub(channel)

        if self.close_transformer_response_future is not None:
            response = self.close_transformer_response_future.result()
            if not response.status:
                raise RuntimeError("RPC close_transformer request failed: {}"
                                   .format(response.message))
            self.is_trans_built = False
            self.close_transformer_response_future = None

        response = self.RPC.BuildTransformer(
            hetr_pb2.BuildTransformerRequest(transformer_type=self.transformer_type),
            _TIMEOUT_SECONDS)
        if response.status:
            self.is_trans_built = True
        else:
            self.is_trans_built = False
            raise RuntimeError("RPC build_transformer request failed: {}".format(response.message))

def run():
  protoConfig = ProtoConfig.getConfig()
  channel = grpc.insecure_channel('localhost:%s' % protoConfig.ports.frontEndPort)
  stub = sensors_pb2.FrontEndStub(channel)

  dbchannel = grpc.insecure_channel('localhost:%s' % protoConfig.ports.daoPort)
  dbstub = dao_pb2.DaoStub(dbchannel)

  lux = getLux(stub)
  print('lux: ', lux)

  temperature = getTemperature(stub)
  print('temperature: ', temperature)

  irButton = getIrButtonPressed(stub)
  print('irButton: ', irButton)

  loudness = getSound(stub)
  print('loudness: ', loudness)

  req = sensors_pb2.GetButtonPressedRequest()
  buttonPressed = stub.GetButtonPressed(req).pressed
  print('buttonPressed', buttonPressed)
  req = sensors_pb2.SetLedStripRequest(
    length=30,
    brightness=100 if buttonPressed else 0,
    speed=5)
  response = stub.SetLedStrip(req)

  req = dao_pb2.SelectRequest(
    table='lux',
    limit=10,
    cols=[
      dao_pb2.RequestCol(name='lux'),
      dao_pb2.RequestCol(name='date')
      ],
    )
  columns = dbstub.Select(req).columns
  print('result %s' % columns)

def __init__(self, protoConfig):
    super().__init__()
    wioChannel = grpc.insecure_channel('localhost:%s' % protoConfig.ports.wioPort)
    self.wioStub = sensors_pb2.WioLinkStub(wioChannel)
    arduinoChannel = grpc.insecure_channel('localhost:%s' % protoConfig.ports.arduinoPort)
    self.arduinoStub = sensors_pb2.ArduinoStub(arduinoChannel)

def __init__(self, protoConfig):
    super().__init__()
    pushchannel = grpc.insecure_channel('localhost:%s' % protoConfig.ports.pushPort)
    self.pushStub = sensors_pb2.PushStub(pushchannel)

def stub(self):
        if self._stub is None:
            self._stub = bridge_pb2.BridgeStub(grpc.insecure_channel('%s:%d' % self._server))
            self._init_ten_times()
        return self._stub

def send(self, method_name, message_factory):
        if self._stub is None:
            self._stub = bridge_pb2.BridgeStub(grpc.insecure_channel('%s:%d' % self._server))
            for _ in range(9):
                method = getattr(self._stub, method_name)
                message = message_factory()
                try:
                    return method(message)
                except grpc.RpcError as e:
                    pass
        return getattr(self._stub, method_name)(message_factory())

def test_stream():
    sys.stdout = ThreadPrinter()
    sys.argv = ["", '{"LogLevel": 1, "PingTimeoutDuration": 5000}']
    col = MockStreamCollector("MyStreamCollector", 99)
    col.start()
    t_end = time.time() + 5
    # wait for our collector to print its preamble
    while len(sys.stdout.lines) == 0 and time.time() < t_end:
        time.sleep(.1)
    resp = json.loads(sys.stdout.lines[0])
    client = StreamCollectorStub(
        grpc.insecure_channel(resp["ListenAddress"]))
    metric = snap.Metric(
        namespace=[snap.NamespaceElement(value="intel"),
                   snap.NamespaceElement(value="streaming"),
                   snap.NamespaceElement(value="random"),
                   snap.NamespaceElement(value="int")],
        version=1,
        unit="some unit",
        description="some description")
    mtr = iter([CollectArg(metric).pb])
    metrics = client.StreamMetrics(mtr)
    assert next(metrics).Metrics_Reply.metrics[0].int64_data == 200
    start_waiting_for_new_metric = time.time()
    assert next(metrics).Metrics_Reply.metrics[0].int64_data == 200
    retrieve_metric_time = time.time()
    assert round(retrieve_metric_time - start_waiting_for_new_metric) == 1
    col.stop()

def test_multiple_stream():
    sys.stdout = ThreadPrinter()
    sys.argv = ["", '{"LogLevel": 1, "PingTimeoutDuration": 5000}']
    col = MockStreamCollector("MyStreamCollector", 99)
    col.start()
    t_end = time.time() + 5
    # wait for our collector to print its preamble
    while len(sys.stdout.lines) == 0 and time.time() < t_end:
        time.sleep(.1)
    resp = json.loads(sys.stdout.lines[0])
    client = StreamCollectorStub(
        grpc.insecure_channel(resp["ListenAddress"]))
    metric = snap.Metric(
        namespace=[snap.NamespaceElement(value="intel"),
                   snap.NamespaceElement(value="streaming"),
                   snap.NamespaceElement(value="random"),
                   snap.NamespaceElement(value="int")],
        version=1,
        unit="some unit",
        description="some description",
        config={"send_multiple": True}
    )
    mtr = iter([CollectArg(metric).pb])
    metrics = client.StreamMetrics(mtr)
    a = next(metrics)
    assert len(a.Metrics_Reply.metrics) == 3
    col.stop()

def test_stream_max_collect_duration():
    sys.stdout = ThreadPrinter()
    sys.argv = ["", '{"LogLevel": 1, "PingTimeoutDuration": 5000}']
    col = MockStreamCollector("MyStreamCollector", 99)
    col.start()
    t_end = time.time() + 5
    # wait for our collector to print its preamble
    while len(sys.stdout.lines) == 0 and time.time() < t_end:
        time.sleep(.1)
    resp = json.loads(sys.stdout.lines[0])
    client = StreamCollectorStub(
        grpc.insecure_channel(resp["ListenAddress"]))
    metric = snap.Metric(
        namespace=[snap.NamespaceElement(value="intel"),
                   snap.NamespaceElement(value="streaming"),
                   snap.NamespaceElement(value="random"),
                   snap.NamespaceElement(value="int")],
        version=1,
        config={
            "max-collect-duration": 2,
            "stream_delay": 3
        },
        unit="some unit",
        description="some description")
    col_arg = CollectArg(metric).pb
    mtr = iter([col_arg])
    metrics = client.StreamMetrics(mtr)
    start_waiting_for_new_metric = time.time()
    a = next(metrics)
    retrieve_metric_time = time.time()
    assert round(retrieve_metric_time - start_waiting_for_new_metric) == 2
    assert len(a.Metrics_Reply.metrics) == 0
    start_waiting_for_new_metric = time.time()
    a = next(metrics)
    retrieve_metric_time = time.time()
    assert round(retrieve_metric_time - start_waiting_for_new_metric) == 1
    assert len(a.Metrics_Reply.metrics) == 1
    col.stop()

def test_stream_max_metrics_buffer_with_max_collect_duration():
    sys.stdout = ThreadPrinter()
    sys.argv = ["", '{"LogLevel": 1, "PingTimeoutDuration": 5000}']
    col = MockStreamCollector("MyStreamCollector", 99)
    col.start()
    t_end = time.time() + 5
    # wait for our collector to print its preamble
    while len(sys.stdout.lines) == 0 and time.time() < t_end:
        time.sleep(.1)
    resp = json.loads(sys.stdout.lines[0])
    client = StreamCollectorStub(
        grpc.insecure_channel(resp["ListenAddress"]))
    metric = snap.Metric(
        namespace=[snap.NamespaceElement(value="intel"),
                   snap.NamespaceElement(value="streaming"),
                   snap.NamespaceElement(value="random"),
                   snap.NamespaceElement(value="int")],
        version=1,
        config={
            "stream_delay": 3,
            "max-collect-duration": 4,
            "max-metrics-buffer": 3
        },
        unit="some unit",
        description="some description")
    col_arg = CollectArg(metric).pb
    mtr = iter([col_arg])
    metrics = client.StreamMetrics(mtr)
    start_waiting_for_new_metric = time.time()
    a = next(metrics)
    retrieve_metric_time = time.time()
    assert round(retrieve_metric_time - start_waiting_for_new_metric) == 4
    assert len(a.Metrics_Reply.metrics) == 1
    start_waiting_for_new_metric = time.time()
    a = next(metrics)
    retrieve_metric_time = time.time()
    assert round(retrieve_metric_time - start_waiting_for_new_metric) == 4
    assert len(a.Metrics_Reply.metrics) == 1
    col.stop()

def publisher_client():
    """Returns a client (grpc) fixture that is passed into publisher
    tests """
    sys.stdout = ThreadPrinter()
    pub = MockPublisher("MyPublisher", 1)
    pub.start()
    t_end = time.time() + 5
    # wait for our collector to print its preamble
    while len(sys.stdout.lines) == 0 and time.time() < t_end:
        time.sleep(.1)
    resp = json.loads(sys.stdout.lines[0])
    client = PublisherStub(
        grpc.insecure_channel(resp["ListenAddress"]))
    yield client
    pub.stop()

def collector_client():
    """Returns a client (grpc) fixture that is passed into collector tests"""
    sys.stdout = ThreadPrinter()
    sys.argv = ["", '{"LogLevel": 1, "PingTimeoutDuration": 5000}']
    col = MockCollector("MyCollector", 99)
    col.start()
    t_end = time.time() + 5
    # wait for our collector to print its preamble
    while len(sys.stdout.lines) == 0 and time.time() < t_end:
        time.sleep(.1)
    resp = json.loads(sys.stdout.lines[0])
    client = CollectorStub(
        grpc.insecure_channel(resp["ListenAddress"]))
    yield client
    col.stop()

def _get_prediction_service_stub():
    channel = grpc.insecure_channel('{}:{}'.format('127.0.0.1', 9000))
    return prediction_service.PredictionServiceStub(channel)

def run_peer_server_as_process_and_stub(port, radiostation_port=conf.PORT_RADIOSTATION, group_id=None, score=None):
    process = run_peer_server_as_process(port, radiostation_port, group_id, score)
    channel = grpc.insecure_channel('localhost:' + str(port))
    stub = loopchain_pb2_grpc.PeerServiceStub(channel)
    util.request_server_in_time(stub.GetStatus, loopchain_pb2.StatusRequest(request=""))
    return process, stub

def run_black_peer_server_as_process_and_stub(port, radiostation_port=conf.PORT_RADIOSTATION, group_id=None):
    process = run_black_peer_server_as_process(port, radiostation_port, group_id)
    channel = grpc.insecure_channel('localhost:' + str(port))
    stub = loopchain_pb2_grpc.PeerServiceStub(channel)
    return process, stub

def run_radio_station_as_process_and_stub(port):
    process = run_radio_station_as_process(port)
    channel = grpc.insecure_channel('localhost:' + str(port))
    stub = loopchain_pb2_grpc.RadioStationStub(channel)
    util.request_server_in_time(stub.GetStatus, loopchain_pb2.StatusRequest(request=""))
    return process, stub

def menu4_1(params=None):
    admin_manager = AdminManager("demotool")

    print("\nInput Peer Target [IP]:[port] (default '' -> 127.0.0.1:7100, [port] -> 127.0.0.1:[port])")
    choice = input(" >>  ")
    if choice == "":
        choice = "127.0.0.1:7100"
    elif choice.find(':') == -1:
        choice = "127.0.0.1:" + choice

    select_channel_index = 0
    select_channel_string = ""
    for channel in admin_manager.get_channel_list():
        if select_channel_index != 0:
            select_channel_string += ", "
        select_channel_string += f"{select_channel_index}: {admin_manager.get_channel_list()[select_channel_index]}"
        select_channel_index += 1

    print(f"Select Channel ({select_channel_string})")
    channel_choice = input(" >>  ")
    try:
        test_globals["channel_name"] = admin_manager.get_channel_list()[int(channel_choice)]
    except Exception as e:
        print(f"wrong channel number! Now use default channel({admin_manager.get_channel_list()[0]})\n")
        test_globals["channel_name"] = admin_manager.get_channel_list()[0]

    print("your input: " + choice)
    channel = grpc.insecure_channel(choice)
    peer_stub = loopchain_pb2_grpc.PeerServiceStub(channel)
    response = peer_stub.GetStatus(loopchain_pb2.StatusRequest(request="hello"), conf.GRPC_TIMEOUT)
    print("Peer Status: " + str(response))
    menu4(peer_stub)

def __make_stub(self, is_stub_reuse=True):
        if util.datetime_diff_in_mins(self.__stub_update_time) >= conf.STUB_REUSE_TIMEOUT or \
                not is_stub_reuse or self.__stub is None:
            util.logger.spam(f"StubManager:__make_stub is_stub_reuse({is_stub_reuse}) self.__stub({self.__stub})")
            self.__stub = util.get_stub_to_server(self.__target, self.__stub_type, is_check_status=False)
            # if self.__is_secure:
            #     # TODO need treat to secure channel but not yet
            #     channel = grpc.insecure_channel(self.__target)
            # else:
            #     channel = grpc.insecure_channel(self.__target)
            #
            # self.__stub = self.__stub_type(channel)
            self.__stub_update_time = datetime.datetime.now()
        else:
            pass

def set_stub_port(self, port, IP_address):
        self.__stub_to_peer_service = loopchain_pb2_grpc.PeerServiceStub(
            grpc.insecure_channel(IP_address + ':' + str(port)))

def get_stub_to_server(target, stub_class, time_out_seconds=None, is_check_status=True):
    """gRPC connection to server

    :return: stub to server
    """
    if time_out_seconds is None:
        time_out_seconds = conf.CONNECTION_RETRY_TIMEOUT
    stub = None
    start_time = timeit.default_timer()
    duration = timeit.default_timer() - start_time

    while stub is None and duration < time_out_seconds:
        try:
            logging.debug("(util) get stub to server target: " + str(target))
            channel = grpc.insecure_channel(target)
            stub = stub_class(channel)
            if is_check_status:
                stub.Request(loopchain_pb2.Message(code=message_code.Request.status), conf.GRPC_TIMEOUT)
        except Exception as e:
            logging.warning("Connect to Server Error(get_stub_to_server): " + str(e))
            logging.debug("duration(" + str(duration)
                          + ") interval(" + str(conf.CONNECTION_RETRY_INTERVAL)
                          + ") timeout(" + str(time_out_seconds) + ")")
            # RETRY_INTERVAL ?? ??? TIMEOUT ??? ?? ??
            time.sleep(conf.CONNECTION_RETRY_INTERVAL)
            duration = timeit.default_timer() - start_time
            stub = None

    return stub

def __handler_connect(self, request, context):
        """make stub to peer service

        :param request: message=target of peer_service
        :param context: 
        :return: 
        """
        logging.debug("__handler_connect %s", request.message)
        self.__stub_to_peer_service = loopchain_pb2_grpc.PeerServiceStub(grpc.insecure_channel(request.message))
        return_code = (message_code.Response.success, message_code.Response.fail)[self.__stub_to_peer_service is None]
        return loopchain_pb2.Message(code=return_code)

def run(self):
        remote = self.remote
        channel = grpc.insecure_channel(remote)
        stub = thegame_pb2_grpc.TheGameStub(channel)
        self._queue = queue.Queue()
        try:
            request_iterator = self._gen()
            response_iterator = stub.Game(request_iterator)
            for response in response_iterator:
                self._game_state = response
                self._queue.put(self._response_to_controls(response))
        finally:
            self._queue.put(Stop)

def run(self):
        self._parse()
        remote = self.remote
        channel = grpc.insecure_channel(remote)
        stub = thegame_pb2_grpc.TheGameStub(channel)
        for response in stub.View(thegame_pb2.ViewRequest(token=self.token)):
            self._game_state = response
            self._response_to_controls(response)

def get_stub_from_manager_endpoint(manager_endpoint):
    endpoint = manager_endpoint.endpoint
    # logger.debug("looking for endpoint %s" % endpoint)
    channel = grpc.insecure_channel(endpoint)
    return messages_pb2_grpc.ManagerAgentStub(channel)

def run():
    channel = grpc.insecure_channel('localhost:50051')
    stub = messages_pb2_grpc.RegistrationServiceStub(channel)
    response = stub.register(
        messages_pb2.RegisterMessage(name='manager_name', endpoint='localhost', description='This is a very long '
                                                                                            'description, bla bla bla'
                                                                                            ' bla bla bla bla bla bla '
                                                                                            'bla bla bla'))
    print("Greeter client received: %s" % response.result)

def predict(self, request_data, request_timeout=10):

        logger.info('Sending request to tfserving model')
        logger.info('Model name: ' + str(self.model_name))
        logger.info('Model version: ' + str(self.model_version))
        logger.info('Host: ' + str(self.host))

        tensor_shape = request_data.shape

        if self.model_name == 'incv4' or self.model_name == 'res152':
            features_tensor_proto = tf.contrib.util.make_tensor_proto(request_data, shape=tensor_shape)
        else:
            features_tensor_proto = tf.contrib.util.make_tensor_proto(request_data,
                                                                      dtype=tf.float32, shape=tensor_shape)

        # Create gRPC client and request
        channel = grpc.insecure_channel(self.host)
        stub = PredictionServiceStub(channel)
        request = PredictRequest()

        request.model_spec.name = self.model_name

        if self.model_version > 0:
            request.model_spec.version.value = self.model_version

        request.inputs['inputs'].CopyFrom(features_tensor_proto)

        try:
            result = stub.Predict(request, timeout=request_timeout)
            logger.info('Got scores with len: ' + str(len(list(result.outputs['scores'].float_val))))
            return list(result.outputs['scores'].float_val)
        except RpcError as e:
            logger.error(e)
            logger.error('Prediction failed!')

def predict(self, request_data, request_timeout=10):

        logger.info('Sending request to tfserving model')
        logger.info('Model name: ' + str(self.model_name))
        logger.info('Model version: ' + str(self.model_version))
        logger.info('Host: ' + str(self.host))

        tensor_shape = request_data.shape

        if self.model_name == 'incv4' or self.model_name == 'res152':
            features_tensor_proto = tf.contrib.util.make_tensor_proto(request_data, shape=tensor_shape)
        else:
            features_tensor_proto = tf.contrib.util.make_tensor_proto(request_data,
                                                                      dtype=tf.float32, shape=tensor_shape)

        # Create gRPC client and request
        channel = grpc.insecure_channel(self.host)
        stub = PredictionServiceStub(channel)
        request = PredictRequest()

        request.model_spec.name = self.model_name

        if self.model_version > 0:
            request.model_spec.version.value = self.model_version

        request.inputs['inputs'].CopyFrom(features_tensor_proto)

        try:
            result = stub.Predict(request, timeout=request_timeout)
            logger.debug('Predicted scores with len: ' + str(len(list(result.outputs['scores'].float_val))))
            return list(result.outputs['scores'].float_val)
        except RpcError as e:
            logger.warning(e)
            logger.warning('Prediction failed. Mock client will return empty prediction of length: '
                           + str(self.num_scores))
            return [0] * self.num_scores