async fn wait_on_operation_stream<S>(&self, mut stream: S, build_id: &str) -> StreamOutcome
  where
    S: Stream<Item = Result<bazel_protos::operations::Operation, grpcio::Error>> + Unpin,
  {
    let mut operation_name_opt: Option<String> = None;

    trace!(
      "wait_on_operation_stream (build_id={}): monitoring stream",
      build_id
    );

    loop {
      match stream.next().await {
        Some(Ok(operation)) => {
          trace!(
            "wait_on_operation_stream (build_id={}): got operation: {:?}",
            build_id,
            &operation
          );

          // Extract the operation name.
          // Note: protobuf can return empty string for an empty field so convert empty strings
          // to None.
          operation_name_opt =
            Some(operation.get_name().to_string()).filter(|s| !s.trim().is_empty());

          // Continue monitoring if the operation is not complete.
          if !operation.get_done() {
            continue;
          }

          // Otherwise, return to the main loop with the operation as the result.
          return StreamOutcome::Complete(OperationOrStatus::Operation(operation));
        }

        Some(Err(err)) => {
          debug!("wait_on_operation_stream: got error: {:?}", err);
          match rpcerror_to_status_or_string(&err) {
            Ok(status) => return StreamOutcome::Complete(OperationOrStatus::Status(status)),
            Err(message) => {
              let code = match err {
                grpcio::Error::RpcFailure(rpc_status) => rpc_status.status,
                _ => grpcio::RpcStatusCode::UNKNOWN,
              };
              let mut status = bazel_protos::status::Status::new();
              status.set_code(code.into());
              status.set_message(format!("gRPC error: {}", message));
              return StreamOutcome::Complete(OperationOrStatus::Status(status));
            }
          }
        }

        None => {
          // Stream disconnected unexpectedly.
          debug!("wait_on_operation_stream: unexpected disconnect from RE server");
          return StreamOutcome::StreamClosed(operation_name_opt);
        }
      }
    }
  }
// Monitors the operation stream returned by the REv2 Execute and WaitExecution methods.
// Outputs progress reported by the server and returns the next actionable operation
// or gRPC status back to the main loop (plus the operation name so the main loop can
// reconnect).
Monitors the operation stream returned by the REv2 Execute and WaitExecution methods.
Outputs progress reported by the server and returns the next actionable operation
or gRPC status back to the main loop (plus the operation name so the main loop can
reconnect).
Monitors the operation stream returned by the REv2 Execute and WaitExecution methods . Outputs progress reported by the server and returns the next actionable operation or gRPC status back to the main loop ( plus the operation name so the main loop can reconnect ) .
async fn wait_on_operation_stream<S>(&self, mut stream: S, build_id: &str) -> StreamOutcome
  where
    S: Stream<Item = Result<bazel_protos::operations::Operation, grpcio::Error>> + Unpin,
  {
    let mut operation_name_opt: Option<String> = None;
    trace!(
      "wait_on_operation_stream (build_id={}): monitoring stream",
      build_id
    );
    loop {
      match stream.next().await {
        Some(Ok(operation)) => {
          trace!(
            "wait_on_operation_stream (build_id={}): got operation: {:?}",
            build_id,
            &operation
          );
          // Extract the operation name.
          // Note: protobuf can return empty string for an empty field so convert empty strings
          // to None.
          operation_name_opt =
            Some(operation.get_name().to_string()).filter(|s| !s.trim().is_empty());
          // Continue monitoring if the operation is not complete.
          if !operation.get_done() {
            continue;
          }
          // Otherwise, return to the main loop with the operation as the result.
          return StreamOutcome::Complete(OperationOrStatus::Operation(operation));
        }
        Some(Err(err)) => {
          debug!("wait_on_operation_stream: got error: {:?}", err);
          match rpcerror_to_status_or_string(&err) {
            Ok(status) => return StreamOutcome::Complete(OperationOrStatus::Status(status)),
            Err(message) => {
              let code = match err {
                grpcio::Error::RpcFailure(rpc_status) => rpc_status.status,
                _ => grpcio::RpcStatusCode::UNKNOWN,
              };
              let mut status = bazel_protos::status::Status::new();
              status.set_code(code.into());
              status.set_message(format!("gRPC error: {}", message));
              return StreamOutcome::Complete(OperationOrStatus::Status(status));
            }
          }
        }
        None => {
          // Stream disconnected unexpectedly.
          debug!("wait_on_operation_stream: unexpected disconnect from RE server");
          return StreamOutcome::StreamClosed(operation_name_opt);
        }
      }
    }
  }
async fn wait_on_operation_stream < S > ( & self , mut stream : S , build_id : & str ) -> StreamOutcome where S : Stream < Item = Result < bazel_protos :: operations :: Operation , grpcio :: Error > > + Unpin , { let mut operation_name_opt : Option < String > = None ; trace ! ( " " build_id ) ; loop { match stream . next ( ) . await { Some ( Ok ( operation ) ) => { trace ! ( " " build_id operation ) ; // Extract the operation name. // Note: protobuf can return empty string for an empty field so convert empty strings // to None. operation_name_opt = Some ( operation . get_name ( ) . to_string ( ) ) . filter ( | s | ! s . trim ( ) . is_empty ( ) ) ; // Continue monitoring if the operation is not complete. if ! operation . get_done ( ) { continue ; } // Otherwise, return to the main loop with the operation as the result. return StreamOutcome :: Complete ( OperationOrStatus :: Operation ( operation ) ) ; } Some ( Err ( err ) ) => { debug ! ( " " err ) ; match rpcerror_to_status_or_string ( & err ) { Ok ( status ) => return StreamOutcome :: Complete ( OperationOrStatus :: Status ( status ) ) , Err ( message ) => { let code = match err { grpcio :: Error :: RpcFailure ( rpc_status ) => rpc_status . status , _ => grpcio :: RpcStatusCode :: UNKNOWN , } ; let mut status = bazel_protos :: status :: Status :: new ( ) ; status . set_code ( code . into ( ) ) ; status . set_message ( format ! ( " " message ) ) ; return StreamOutcome :: Complete ( OperationOrStatus :: Status ( status ) ) ; } } } None => { // Stream disconnected unexpectedly. debug ! ( " " ) ; return StreamOutcome :: StreamClosed ( operation_name_opt ) ; } } } }
Monitors the operation stream returned by the REv2 Execute and WaitExecution methods.
Monitors the operation stream returned by the REv2 Execute and WaitExecution methods .
&self
stream
S
build_id
&str
&self
stream
S
build_id
&str