FIxed

azure-kusto-data/Cargo.toml

@@ -22,7 +22,6 @@ async-trait = "0.1.57"
 async-convert = "1.0.0"
 bytes = "1.2.1"
 futures = "0.3.23"
-http = "0.2.8"
 serde = { version = "1.0.143", features = ["derive"] }
 serde_json = "1.0.81"
 serde_with = { version = "2.0.0", features = ["json"] }

azure-kusto-data/src/cloud_info.rs

@@ -1,13 +1,11 @@
-use arrow::datatypes::ToByteSlice;
 use std::borrow::Cow;
 
+use azure_core::error::Error as CoreError;
 use azure_core::prelude::*;
-use azure_core::{collect_pinned_stream, Context, Pipeline, Request};
+use azure_core::{Context, Method, Pipeline, Request, StatusCode};
 use futures::lock::Mutex;
 use hashbrown::hash_map::EntryRef;
 use hashbrown::HashMap;
-use http::Method;
-use http::StatusCode;
 use once_cell::sync::Lazy;
 use serde::{Deserialize, Serialize};
 
@@ -50,24 +48,24 @@ impl CloudInfo {
 
     async fn fetch(pipeline: &Pipeline, endpoint: &str) -> Result<CloudInfo, crate::error::Error> {
         let metadata_endpoint = format!("{}/{}", endpoint, CloudInfo::METADATA_ENDPOINT);
-        let mut request = Request::new(metadata_endpoint.parse()?, Method::GET);
+        let mut request = Request::new(
+            metadata_endpoint.parse().map_err(CoreError::from)?,
+            Method::Get,
+        );
         request.insert_headers(&Accept::from("application/json"));
         request.insert_headers(&AcceptEncoding::from("gzip, deflate"));
         let response = pipeline.send(&mut Context::new(), &mut request).await?;
         let (status_code, _header_map, pinned_stream) = response.deconstruct();
         match status_code {
-            StatusCode::OK => {
-                let data = collect_pinned_stream(pinned_stream).await?;
+            StatusCode::Ok => {
+                let data = pinned_stream.collect().await?;
                 let result: AzureAd = serde_json::from_slice(&data)?;
                 Ok(result.azure_ad)
             }
-            StatusCode::NOT_FOUND => Ok(Default::default()),
+            StatusCode::NotFound => Ok(Default::default()),
             _ => Err(crate::error::Error::HttpError(
                 status_code,
-                String::from_utf8_lossy(
-                    collect_pinned_stream(pinned_stream).await?.to_byte_slice(),
-                )
-                .to_string(),
+                String::from_utf8_lossy((pinned_stream).collect().await?.as_ref()).to_string(),
             )),
         }
     }

azure-kusto-data/src/connection_string.rs

@@ -339,8 +339,14 @@ impl ConnectionStringAuth {
         match self {
             ConnectionStringAuth::Default => Arc::new(DefaultAzureCredential::default()),
             ConnectionStringAuth::UserAndPassword { .. } => unimplemented!(),
-            ConnectionStringAuth::Token { .. } => unimplemented!(),
-            ConnectionStringAuth::TokenCallback { .. } => unimplemented!(),
+            ConnectionStringAuth::Token { token } => Arc::new(ConstTokenCredential { token }),
+            ConnectionStringAuth::TokenCallback {
+                token_callback,
+                time_to_live,
+            } => Arc::new(CallbackTokenCredential {
+                token_callback,
+                time_to_live,
+            }),
             ConnectionStringAuth::Application {
                 client_id,
                 client_secret,

azure-kusto-data/src/error.rs

@@ -1,5 +1,5 @@
 //! Defines [Error] for representing failures in various operations.
-use http::StatusCode;
+use azure_core::StatusCode;
 use std::fmt::Debug;
 use std::num::TryFromIntError;
 
@@ -43,10 +43,6 @@ pub enum Error {
     /// Errors raised when the operation is not supported
     #[error("Operation not supported: {0}")]
     UnsupportedOperation(String),
-
-    /// Invalid uri error
-    #[error("Invalid uri: {0}")]
-    InvalidUri(#[from] http::uri::InvalidUri),
 }
 
 /// Errors raised when an invalid argument or option is provided.

azure-kusto-data/src/operations/query.rs

@@ -16,7 +16,6 @@ use azure_core::prelude::*;
 use azure_core::{Method, Request, Response as HttpResponse, Response, Url};
 use futures::future::BoxFuture;
 use futures::{Stream, TryFutureExt, TryStreamExt};
-use http::header::CONNECTION;
 use serde::{Deserialize, Serialize};
 use std::collections::HashMap;
 use std::io::ErrorKind;
@@ -538,7 +537,7 @@ pub fn prepare_request(url: Url, http_method: Method) -> Request {
         "Kusto.Rust.Client:{}",
         env!("CARGO_PKG_VERSION"),
     )));
-    request.insert_header(CONNECTION.as_str(), "Keep-Alive");
+    request.insert_header("Connection", "Keep-Alive");
     request
 }