| 698 | } |
| 699 | |
| 700 | async fn put_with_ttl( |
| 701 | &self, |
| 702 | key: impl AsRef<[u8]> + Send, |
| 703 | value: impl AsRef<[u8]> + Send, |
| 704 | ttl_secs: u64, |
| 705 | ) -> ClientApiResult<()> { |
| 706 | let request = ClientWriteRequest { |
| 707 | client_id: self.client_id, |
| 708 | command: Some(WriteOperation::Insert { |
| 709 | key: Bytes::copy_from_slice(key.as_ref()), |
| 710 | value: Bytes::copy_from_slice(value.as_ref()), |
| 711 | ttl_secs: Some(ttl_secs), |
| 712 | }), |
| 713 | }; |
| 714 | |
| 715 | let (resp_tx, resp_rx) = MaybeCloneOneshot::new(); |
| 716 | |
| 717 | self.cmd_tx |
| 718 | .send(d_engine_core::ClientCmd::Propose(request, resp_tx)) |
| 719 | .await |
| 720 | .map_err(|_| channel_closed_error())?; |
| 721 | |
| 722 | let result = tokio::time::timeout(self.timeout, resp_rx) |
| 723 | .await |
| 724 | .map_err(|_| timeout_error(self.timeout))? |
| 725 | .map_err(|_| channel_closed_error())?; |
| 726 | |
| 727 | let response = |
| 728 | result.map_err(|status| server_error(format!("RPC error: {}", status.message())))?; |
| 729 | |
| 730 | if response.error != ErrorCode::Success { |
| 731 | return Err(Self::map_error_response( |
| 732 | response.error, |
| 733 | response.leader_hint, |
| 734 | response.retry_after_ms, |
| 735 | )); |
| 736 | } |
| 737 | |
| 738 | Ok(()) |
| 739 | } |
| 740 | |
| 741 | async fn get( |
| 742 | &self, |