Page MenuHomePhabricator
Differential D8507

[keyserver][lib] Simplify createThread return type following deprecation of old codeVersions
ClosedPublic
Actions

Authored by ashoat on Jul 15 2023, 8:04 PM.
Tags
None
Referenced Files
Unknown Object (File)
Tue, Apr 22, 8:25 AM
Unknown Object (File)
Sun, Apr 20, 12:49 PM
Unknown Object (File)
Sun, Apr 20, 12:25 AM
Unknown Object (File)
Sat, Apr 19, 9:32 AM
Unknown Object (File)
Wed, Apr 2, 4:22 AM
Unknown Object (File)
Mar 20 2025, 3:14 AM
Unknown Object (File)
Mar 17 2025, 10:43 PM
Unknown Object (File)
Mar 17 2025, 10:43 PM
View All 99 Files
Subscribers
tomek

Details

Reviewers
michal
kamil
Commits
rCOMMbfb41af325bb: [keyserver][lib] Simplify createThread return type following deprecation of old…
Summary

See the first diff in the stack. We're no longer expecting to see these old codeVersions, so we're simplifying this code.

Depends on D8506

Test Plan

I tested this whole stack together by playing around with the web app and making sure thread mutation endpoints seemed to work correctly

Diff Detail

Repository
rCOMM Comm
Lint
No Lint Coverage
Unit
No Test Coverage

Revision Contents
Changeset List

PathSize
keyserver/
src/
creators/
25 lines
22 lines
responders/
3 lines
lib/
types/
3 lines

Diff 28702

View Options

keyserver/src/creators/account-creator.js

use derive_more::{Display, Error, From}; use derive_more::{Display, Error, From};
use futures_core::Stream; use futures_core::Stream;
use futures_util::StreamExt; use futures_util::{StreamExt, TryStreamExt};
use reqwest::Url; use reqwest::{
multipart::{Form, Part},
Body, Url,
};
use tracing::{debug, trace, warn}; use tracing::{debug, trace, warn};
// publicly re-export some reqwest types // publicly re-export some reqwest types
pub use reqwest::Error as ReqwestError; pub use reqwest::Error as ReqwestError;
pub use reqwest::StatusCode; pub use reqwest::StatusCode;
#[derive(From, Error, Debug, Display)] #[derive(From, Error, Debug, Display)]
pub enum BlobServiceError { pub enum BlobServiceError {
▲ Show 20 Lines • Show All 91 Lines • ▼ Show 20 Lines ) -> BlobResult<impl Stream<Item = BlobResult<Vec<u8>>>> {
} }
let error = handle_http_error(response.status()); let error = handle_http_error(response.status());
if let Ok(message) = response.text().await { if let Ok(message) = response.text().await {
trace!("Error response message: {}", message); trace!("Error response message: {}", message);
} }
Err(error) Err(error)
} }
/// Uploads a blob. Returns `BlobServiceError::AlreadyExists` if blob with given hash
/// already exists.
///
/// # Example
/// ```rust
/// use std::io::{Error, ErrorKind};
///
/// let client =
/// BlobServiceClient::new("http://localhost:50053".parse()?);
///
/// let stream = async_stream::stream! {
/// yield Ok(vec![1, 2, 3]);
/// yield Ok(vec![4, 5, 6]);
/// yield Err(Error::new(ErrorKind::Other, "Oops"));
/// };
/// client.upload_blob(&blob_hash, stream).await?;
/// ```
pub async fn upload_blob<H, S>(
&self,
blob_hash: H,
data_stream: S,
) -> BlobResult<()>
where
H: Into<String>,
S: futures_core::stream::TryStream + Send + Sync + 'static,
S::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
Vec<u8>: From<S::Ok>,
bartekAuthorUnsubmitted
Done
Inline Actions

Requirements of the reqwest body stream:
A stream of Result<impl Into<Vec<u8>>, impl Into<Box<dyn Error>>>

bartek: Requirements of the reqwest body stream: A stream of `Result<impl Into<Vec<u8>>, impl…
{
debug!("Revoke holder request");
michalUnsubmitted
Not Done
Inline Actions

Wrong message?

michal: Wrong message?
let url = self.get_blob_url(None)?;
let stream = data_stream.map_ok(Vec::from);
let streaming_body = Body::wrap_stream(stream);
let form = Form::new()
.text("blob_hash", blob_hash.into())
.part("blob_data", Part::stream(streaming_body));
let response = self.http_client.put(url).multipart(form).send().await?;
debug!("Response status: {}", response.status());
if response.status().is_success() {
trace!("Blob upload successful");
return Ok(());
}
let error = handle_http_error(response.status());
if let Ok(message) = response.text().await {
trace!("Error response message: {}", message);
}
Err(error)
}
} }
// private helper methods // private helper methods
impl BlobServiceClient { impl BlobServiceClient {
fn get_blob_url( fn get_blob_url(
&self, &self,
blob_hash: Option<&str>, blob_hash: Option<&str>,
) -> Result<Url, BlobServiceError> { ) -> Result<Url, BlobServiceError> {
Show All 24 Lines
View Options

keyserver/src/creators/thread-creator.js

use derive_more::{Display, Error, From}; use derive_more::{Display, Error, From};
use futures_core::Stream; use futures_core::Stream;
use futures_util::StreamExt; use futures_util::{StreamExt, TryStreamExt};
use reqwest::Url; use reqwest::{
multipart::{Form, Part},
Body, Url,
};
use tracing::{debug, trace, warn}; use tracing::{debug, trace, warn};
// publicly re-export some reqwest types // publicly re-export some reqwest types
pub use reqwest::Error as ReqwestError; pub use reqwest::Error as ReqwestError;
pub use reqwest::StatusCode; pub use reqwest::StatusCode;
#[derive(From, Error, Debug, Display)] #[derive(From, Error, Debug, Display)]
pub enum BlobServiceError { pub enum BlobServiceError {
▲ Show 20 Lines • Show All 91 Lines • ▼ Show 20 Lines ) -> BlobResult<impl Stream<Item = BlobResult<Vec<u8>>>> {
} }
let error = handle_http_error(response.status()); let error = handle_http_error(response.status());
if let Ok(message) = response.text().await { if let Ok(message) = response.text().await {
trace!("Error response message: {}", message); trace!("Error response message: {}", message);
} }
Err(error) Err(error)
} }
/// Uploads a blob. Returns `BlobServiceError::AlreadyExists` if blob with given hash
/// already exists.
///
/// # Example
/// ```rust
/// use std::io::{Error, ErrorKind};
///
/// let client =
/// BlobServiceClient::new("http://localhost:50053".parse()?);
///
/// let stream = async_stream::stream! {
/// yield Ok(vec![1, 2, 3]);
/// yield Ok(vec![4, 5, 6]);
/// yield Err(Error::new(ErrorKind::Other, "Oops"));
/// };
/// client.upload_blob(&blob_hash, stream).await?;
/// ```
pub async fn upload_blob<H, S>(
&self,
blob_hash: H,
data_stream: S,
) -> BlobResult<()>
where
H: Into<String>,
S: futures_core::stream::TryStream + Send + Sync + 'static,
S::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
Vec<u8>: From<S::Ok>,
bartekAuthorUnsubmitted
Done
Inline Actions

Requirements of the reqwest body stream:
A stream of Result<impl Into<Vec<u8>>, impl Into<Box<dyn Error>>>

bartek: Requirements of the reqwest body stream: A stream of `Result<impl Into<Vec<u8>>, impl…
{
debug!("Revoke holder request");
michalUnsubmitted
Not Done
Inline Actions

Wrong message?

michal: Wrong message?
let url = self.get_blob_url(None)?;
let stream = data_stream.map_ok(Vec::from);
let streaming_body = Body::wrap_stream(stream);
let form = Form::new()
.text("blob_hash", blob_hash.into())
.part("blob_data", Part::stream(streaming_body));
let response = self.http_client.put(url).multipart(form).send().await?;
debug!("Response status: {}", response.status());
if response.status().is_success() {
trace!("Blob upload successful");
return Ok(());
}
let error = handle_http_error(response.status());
if let Ok(message) = response.text().await {
trace!("Error response message: {}", message);
}
Err(error)
}
} }
// private helper methods // private helper methods
impl BlobServiceClient { impl BlobServiceClient {
fn get_blob_url( fn get_blob_url(
&self, &self,
blob_hash: Option<&str>, blob_hash: Option<&str>,
) -> Result<Url, BlobServiceError> { ) -> Result<Url, BlobServiceError> {
Show All 24 Lines
View Options

keyserver/src/responders/thread-responders.js

use derive_more::{Display, Error, From}; use derive_more::{Display, Error, From};
use futures_core::Stream; use futures_core::Stream;
use futures_util::StreamExt; use futures_util::{StreamExt, TryStreamExt};
use reqwest::Url; use reqwest::{
multipart::{Form, Part},
Body, Url,
};
use tracing::{debug, trace, warn}; use tracing::{debug, trace, warn};
// publicly re-export some reqwest types // publicly re-export some reqwest types
pub use reqwest::Error as ReqwestError; pub use reqwest::Error as ReqwestError;
pub use reqwest::StatusCode; pub use reqwest::StatusCode;
#[derive(From, Error, Debug, Display)] #[derive(From, Error, Debug, Display)]
pub enum BlobServiceError { pub enum BlobServiceError {
▲ Show 20 Lines • Show All 91 Lines • ▼ Show 20 Lines ) -> BlobResult<impl Stream<Item = BlobResult<Vec<u8>>>> {
} }
let error = handle_http_error(response.status()); let error = handle_http_error(response.status());
if let Ok(message) = response.text().await { if let Ok(message) = response.text().await {
trace!("Error response message: {}", message); trace!("Error response message: {}", message);
} }
Err(error) Err(error)
} }
/// Uploads a blob. Returns `BlobServiceError::AlreadyExists` if blob with given hash
/// already exists.
///
/// # Example
/// ```rust
/// use std::io::{Error, ErrorKind};
///
/// let client =
/// BlobServiceClient::new("http://localhost:50053".parse()?);
///
/// let stream = async_stream::stream! {
/// yield Ok(vec![1, 2, 3]);
/// yield Ok(vec![4, 5, 6]);
/// yield Err(Error::new(ErrorKind::Other, "Oops"));
/// };
/// client.upload_blob(&blob_hash, stream).await?;
/// ```
pub async fn upload_blob<H, S>(
&self,
blob_hash: H,
data_stream: S,
) -> BlobResult<()>
where
H: Into<String>,
S: futures_core::stream::TryStream + Send + Sync + 'static,
S::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
Vec<u8>: From<S::Ok>,
bartekAuthorUnsubmitted
Done
Inline Actions

Requirements of the reqwest body stream:
A stream of Result<impl Into<Vec<u8>>, impl Into<Box<dyn Error>>>

bartek: Requirements of the reqwest body stream: A stream of `Result<impl Into<Vec<u8>>, impl…
{
debug!("Revoke holder request");
michalUnsubmitted
Not Done
Inline Actions

Wrong message?

michal: Wrong message?
let url = self.get_blob_url(None)?;
let stream = data_stream.map_ok(Vec::from);
let streaming_body = Body::wrap_stream(stream);
let form = Form::new()
.text("blob_hash", blob_hash.into())
.part("blob_data", Part::stream(streaming_body));
let response = self.http_client.put(url).multipart(form).send().await?;
debug!("Response status: {}", response.status());
if response.status().is_success() {
trace!("Blob upload successful");
return Ok(());
}
let error = handle_http_error(response.status());
if let Ok(message) = response.text().await {
trace!("Error response message: {}", message);
}
Err(error)
}
} }
// private helper methods // private helper methods
impl BlobServiceClient { impl BlobServiceClient {
fn get_blob_url( fn get_blob_url(
&self, &self,
blob_hash: Option<&str>, blob_hash: Option<&str>,
) -> Result<Url, BlobServiceError> { ) -> Result<Url, BlobServiceError> {
Show All 24 Lines
View Options

lib/types/thread-types.js

use derive_more::{Display, Error, From}; use derive_more::{Display, Error, From};
use futures_core::Stream; use futures_core::Stream;
use futures_util::StreamExt; use futures_util::{StreamExt, TryStreamExt};
use reqwest::Url; use reqwest::{
multipart::{Form, Part},
Body, Url,
};
use tracing::{debug, trace, warn}; use tracing::{debug, trace, warn};
// publicly re-export some reqwest types // publicly re-export some reqwest types
pub use reqwest::Error as ReqwestError; pub use reqwest::Error as ReqwestError;
pub use reqwest::StatusCode; pub use reqwest::StatusCode;
#[derive(From, Error, Debug, Display)] #[derive(From, Error, Debug, Display)]
pub enum BlobServiceError { pub enum BlobServiceError {
▲ Show 20 Lines • Show All 91 Lines • ▼ Show 20 Lines ) -> BlobResult<impl Stream<Item = BlobResult<Vec<u8>>>> {
} }
let error = handle_http_error(response.status()); let error = handle_http_error(response.status());
if let Ok(message) = response.text().await { if let Ok(message) = response.text().await {
trace!("Error response message: {}", message); trace!("Error response message: {}", message);
} }
Err(error) Err(error)
} }
/// Uploads a blob. Returns `BlobServiceError::AlreadyExists` if blob with given hash
/// already exists.
///
/// # Example
/// ```rust
/// use std::io::{Error, ErrorKind};
///
/// let client =
/// BlobServiceClient::new("http://localhost:50053".parse()?);
///
/// let stream = async_stream::stream! {
/// yield Ok(vec![1, 2, 3]);
/// yield Ok(vec![4, 5, 6]);
/// yield Err(Error::new(ErrorKind::Other, "Oops"));
/// };
/// client.upload_blob(&blob_hash, stream).await?;
/// ```
pub async fn upload_blob<H, S>(
&self,
blob_hash: H,
data_stream: S,
) -> BlobResult<()>
where
H: Into<String>,
S: futures_core::stream::TryStream + Send + Sync + 'static,
S::Error: Into<Box<dyn std::error::Error + Send + Sync>>,
Vec<u8>: From<S::Ok>,
bartekAuthorUnsubmitted
Done
Inline Actions

Requirements of the reqwest body stream:
A stream of Result<impl Into<Vec<u8>>, impl Into<Box<dyn Error>>>

bartek: Requirements of the reqwest body stream: A stream of `Result<impl Into<Vec<u8>>, impl…
{
debug!("Revoke holder request");
michalUnsubmitted
Not Done
Inline Actions

Wrong message?

michal: Wrong message?
let url = self.get_blob_url(None)?;
let stream = data_stream.map_ok(Vec::from);
let streaming_body = Body::wrap_stream(stream);
let form = Form::new()
.text("blob_hash", blob_hash.into())
.part("blob_data", Part::stream(streaming_body));
let response = self.http_client.put(url).multipart(form).send().await?;
debug!("Response status: {}", response.status());
if response.status().is_success() {
trace!("Blob upload successful");
return Ok(());
}
let error = handle_http_error(response.status());
if let Ok(message) = response.text().await {
trace!("Error response message: {}", message);
}
Err(error)
}
} }
// private helper methods // private helper methods
impl BlobServiceClient { impl BlobServiceClient {
fn get_blob_url( fn get_blob_url(
&self, &self,
blob_hash: Option<&str>, blob_hash: Option<&str>,
) -> Result<Url, BlobServiceError> { ) -> Result<Url, BlobServiceError> {
Show All 24 Lines