Perform a `handshake` with the server, passing credentials and establishing a session. If the server returns an "authorization" header, it is automatically parsed and set as a token for future requests. Any other data returned by the server in the handshake response is returned as a binary blob.
(&mut self, username: &str, password: &str)
| 161 | /// a token for future requests. Any other data returned by the server in the handshake |
| 162 | /// response is returned as a binary blob. |
| 163 | pub async fn handshake(&mut self, username: &str, password: &str) -> Result<Bytes> { |
| 164 | let cmd = HandshakeRequest { |
| 165 | protocol_version: 0, |
| 166 | payload: Default::default(), |
| 167 | }; |
| 168 | let mut req = tonic::Request::new(stream::iter(vec![cmd])); |
| 169 | let val = BASE64_STANDARD.encode(format!("{username}:{password}")); |
| 170 | let val = format!("Basic {val}") |
| 171 | .parse() |
| 172 | .map_err(|_| ArrowError::ParseError("Cannot parse header".to_string()))?; |
| 173 | req.metadata_mut().insert("authorization", val); |
| 174 | let req = self.set_request_headers(req)?; |
| 175 | let resp = self |
| 176 | .flight_client |
| 177 | .handshake(req) |
| 178 | .await |
| 179 | .map_err(|e| ArrowError::IpcError(format!("Can't handshake {e}")))?; |
| 180 | if let Some(auth) = resp.metadata().get("authorization") { |
| 181 | let auth = auth |
| 182 | .to_str() |
| 183 | .map_err(|_| ArrowError::ParseError("Can't read auth header".to_string()))?; |
| 184 | let bearer = "Bearer "; |
| 185 | if !auth.starts_with(bearer) { |
| 186 | return Err(ArrowError::ParseError("Invalid auth header!".to_string()))?; |
| 187 | } |
| 188 | let auth = auth[bearer.len()..].to_string(); |
| 189 | self.token = Some(auth); |
| 190 | } |
| 191 | let responses: Vec<HandshakeResponse> = resp |
| 192 | .into_inner() |
| 193 | .try_collect() |
| 194 | .await |
| 195 | .map_err(|_| ArrowError::ParseError("Can't collect responses".to_string()))?; |
| 196 | let resp = match responses.as_slice() { |
| 197 | [resp] => resp.payload.clone(), |
| 198 | [] => Bytes::new(), |
| 199 | _ => Err(ArrowError::ParseError( |
| 200 | "Multiple handshake responses".to_string(), |
| 201 | ))?, |
| 202 | }; |
| 203 | Ok(resp) |
| 204 | } |
| 205 | |
| 206 | /// Execute a update query on the server, and return the number of records affected |
| 207 | pub async fn execute_update( |
no test coverage detected