mirrors/syncstorage-rs
1
0
Fork 0
mirror of https://github.com/mozilla-services/syncstorage-rs.git synced 2026-05-05 12:16:21 +02:00
Code Issues Actions 4 Packages Projects Releases Wiki Activity

chore: cargo fmt (1.31.0)

Browse Source
This commit is contained in:
Philip Jenvey 2018-12-12 13:05:16 -08:00
parent 1d3a8ae3ad
commit fe07704495
No known key found for this signature in database
GPG Key ID: 5B9F83DE4F7EB7FA
10 changed files with 154 additions and 91 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/db/mod.rs
View File

@ -145,7 +145,8 @@ pub trait Db: Send + Debug {
self.get_collection_timestamp(params::GetCollectionTimestamp {
user_id,
collection,
}).or_else(|e| {
})
.or_else(|e| {
if e.is_colllection_not_found() {
Ok(SyncTimestamp::from_seconds(0f64))
} else {
@ -160,7 +161,8 @@ pub trait Db: Send + Debug {
user_id,
collection,
id: bso,
}).or_else(|e| {
})
.or_else(|e| {
if e.is_colllection_not_found() {
Ok(SyncTimestamp::from_seconds(0f64))
} else {

9
src/db/mysql/batch.rs
View File

@ -27,7 +27,8 @@ pub fn create(db: &MysqlDb, params: params::CreateBatch) -> Result<results::Crea
batches::id.eq(&timestamp),
batches::bsos.eq(&bsos),
batches::expiry.eq(timestamp + BATCH_LIFETIME),
)).execute(&db.conn)?;
))
.execute(&db.conn)?;
Ok(timestamp)
}
@ -114,7 +115,8 @@ fn batch_string_to_bsos(bsos: &str) -> Result<Vec<params::PostCollectionBso>> {
serde_json::from_str(line).map_err(|e| {
DbError::internal(&format!("Couldn't deserialize batch::load_bsos bso: {}", e))
})
}).collect()
})
.collect()
}
/// Serialize bsos into strings separated by newlines
@ -125,7 +127,8 @@ fn bsos_to_batch_string(bsos: &[params::PostCollectionBso]) -> Result<String> {
serde_json::to_string(bso).map_err(|e| {
DbError::internal(&format!("Couldn't serialize batch::create bso: {}", e))
})
}).collect();
})
.collect();
batch_strings.map(|bs| {
format!(
"{}{}",

15
src/db/mysql/models.rs
View File

@ -362,7 +362,8 @@ impl MysqlDb {
bso::payload.eq(payload),
bso::modified.eq(timestamp),
bso::expiry.eq(timestamp + (ttl as i64 * 1000)),
)).execute(&self.conn)?;
))
.execute(&self.conn)?;
}
self.touch_collection(user_id as u32, collection_id)
})
@ -388,7 +389,8 @@ impl MysqlDb {
bso::payload,
bso::sortindex,
bso::expiry,
)).filter(bso::user_id.eq(user_id))
))
.filter(bso::user_id.eq(user_id))
.filter(bso::collection_id.eq(collection_id as i32)) // XXX:
.filter(bso::expiry.gt(self.timestamp().as_i64()))
.into_boxed();
@ -461,7 +463,8 @@ impl MysqlDb {
bso::payload,
bso::sortindex,
bso::expiry,
)).filter(bso::user_id.eq(user_id as i32))
))
.filter(bso::user_id.eq(user_id as i32))
.filter(bso::collection_id.eq(&collection_id))
.filter(bso::id.eq(&params.id))
.filter(bso::expiry.ge(self.timestamp().as_i64()))
@ -584,7 +587,8 @@ impl MysqlDb {
.into_iter()
.map(|cr| {
SyncTimestamp::from_i64(cr.modified).and_then(|ts| Ok((cr.collection_id, ts)))
}).collect::<Result<HashMap<_, _>>>()?;
})
.collect::<Result<HashMap<_, _>>>()?;
self.map_collection_names(modifieds)
}
@ -597,7 +601,8 @@ impl MysqlDb {
.remove(&id)
.map(|name| (name, value))
.ok_or_else(|| DbError::internal("load_collection_names get"))
}).collect()
})
.collect()
}
fn load_collection_names<'a>(

9
src/db/util.rs
View File

@ -21,11 +21,7 @@ pub fn ms_since_epoch() -> i64 {
/// Internally represents a Sync timestamp as a u64 representing milliseconds since the epoch.
#[derive(Copy, Clone, Debug, PartialEq, PartialOrd, Deserialize, Serialize, FromSqlRow)]
pub struct SyncTimestamp(
#[serde(
deserialize_with = "deserialize_ts",
serialize_with = "serialize_ts"
)]
u64,
#[serde(deserialize_with = "deserialize_ts", serialize_with = "serialize_ts")] u64,
);
impl SyncTimestamp {
@ -48,7 +44,8 @@ impl SyncTimestamp {
} else {
Ok(v)
}
}).map(|v: f64| (v * 1_000f64) as u64)
})
.map(|v: f64| (v * 1_000f64) as u64)
.map(SyncTimestamp::from_milliseconds)
}

3
src/error.rs
View File

@ -148,7 +148,8 @@ impl ResponseError for ApiError {
HttpResponse::build(self.status)
.if_true(self.is_conflict(), |resp| {
resp.header("Retry-After", RETRY_AFTER.to_string());
}).json(self.weave_error_code() as i32)
})
.json(self.weave_error_code() as i32)
}
}

27
src/server/mod.rs
View File

@ -15,27 +15,35 @@ macro_rules! init_routes {
($app:expr) => {
$app.resource("/1.5/{uid}/info/collections", |r| {
r.method(http::Method::GET).with(handlers::get_collections);
}).resource("/1.5/{uid}/info/collection_counts", |r| {
})
.resource("/1.5/{uid}/info/collection_counts", |r| {
r.method(http::Method::GET)
.with(handlers::get_collection_counts);
}).resource("/1.5/{uid}/info/collection_usage", |r| {
})
.resource("/1.5/{uid}/info/collection_usage", |r| {
r.method(http::Method::GET)
.with(handlers::get_collection_usage);
}).resource("/1.5/{uid}/info/configuration", |r| {
})
.resource("/1.5/{uid}/info/configuration", |r| {
r.method(http::Method::GET)
.with(handlers::get_configuration);
}).resource("/1.5/{uid}/info/quota", |r| {
})
.resource("/1.5/{uid}/info/quota", |r| {
r.method(http::Method::GET).with(handlers::get_quota);
}).resource("/1.5/{uid}", |r| {
})
.resource("/1.5/{uid}", |r| {
r.method(http::Method::DELETE).with(handlers::delete_all);
}).resource("/1.5/{uid}/storage", |r| {
})
.resource("/1.5/{uid}/storage", |r| {
r.method(http::Method::DELETE).with(handlers::delete_all);
}).resource("/1.5/{uid}/storage/{collection}", |r| {
})
.resource("/1.5/{uid}/storage/{collection}", |r| {
r.method(http::Method::DELETE)
.with(handlers::delete_collection);
r.method(http::Method::GET).with(handlers::get_collection);
r.method(http::Method::POST).with(handlers::post_collection);
}).resource("/1.5/{uid}/storage/{collection}/{bso}", |r| {
})
.resource("/1.5/{uid}/storage/{collection}/{bso}", |r| {
r.method(http::Method::DELETE).with(handlers::delete_bso);
r.method(http::Method::GET).with(handlers::get_bso);
r.method(http::Method::PUT).with(handlers::put_bso);
@ -89,7 +97,8 @@ impl Server {
};
build_app(state)
}).bind(format!("127.0.0.1:{}", settings.port))
})
.bind(format!("127.0.0.1:{}", settings.port))
.unwrap()
.start();
Ok(sys)

15
src/server/test.rs
View File

@ -64,7 +64,8 @@ fn create_request(server: &TestServer, method: http::Method, path: &str) -> Clie
.set_header(
"Authorization",
create_hawk_header(method.as_str(), server.addr().port(), path),
).finish()
)
.finish()
.unwrap()
}
@ -291,13 +292,15 @@ fn invalid_content_type() {
server.addr().port(),
"/1.5/42/storage/bookmarks/wibble",
),
).set_header("Content-Type", "application/javascript")
)
.set_header("Content-Type", "application/javascript")
.json(BsoBody {
id: Some("wibble".to_string()),
sortindex: Some(0),
payload: Some("wibble".to_string()),
ttl: Some(31536000),
}).unwrap();
})
.unwrap();
let response = server.execute(request.send()).unwrap();
assert_eq!(response.status(), StatusCode::UNSUPPORTED_MEDIA_TYPE);
@ -307,13 +310,15 @@ fn invalid_content_type() {
.set_header(
"Authorization",
create_hawk_header("POST", server.addr().port(), "/1.5/42/storage/bookmarks"),
).set_header("Content-Type", "application/javascript")
)
.set_header("Content-Type", "application/javascript")
.json(json!([BsoBody {
id: Some("wibble".to_string()),
sortindex: Some(0),
payload: Some("wibble".to_string()),
ttl: Some(31536000),
}])).unwrap();
}]))
.unwrap();
let response = server.execute(request.send()).unwrap();
assert_eq!(response.status(), StatusCode::UNSUPPORTED_MEDIA_TYPE);

90
src/web/extractors.rs
View File

@ -106,7 +106,8 @@ impl FromRequest<ServerState> for BsoBodies {
"Invalid Content-Type".to_owned(),
RequestErrorLocation::Header,
Some("Content-Type".to_owned()),
).into(),
)
.into(),
));
}
}
@ -125,7 +126,8 @@ impl FromRequest<ServerState> for BsoBodies {
"Mimetype/encoding/content-length error".to_owned(),
RequestErrorLocation::Header,
None,
).into(),
)
.into(),
));
};
@ -135,7 +137,8 @@ impl FromRequest<ServerState> for BsoBodies {
"Invalid JSON in request body".to_owned(),
RequestErrorLocation::Body,
Some("bsos".to_owned()),
).into()
)
.into()
}
// Define a new bool to check from a static closure to release the reference on the
@ -201,7 +204,8 @@ impl FromRequest<ServerState> for BsoBodies {
"Input BSO has duplicate ID".to_owned(),
RequestErrorLocation::Body,
Some("bsos".to_owned()),
).into(),
)
.into(),
);
} else {
bso_ids.push(id.clone());
@ -213,7 +217,8 @@ impl FromRequest<ServerState> for BsoBodies {
"Input BSO has no ID".to_owned(),
RequestErrorLocation::Body,
Some("bsos".to_owned()),
).into(),
)
.into(),
);
};
match BatchBsoBody::from_raw_bso(&bso) {
@ -267,7 +272,8 @@ impl FromRequest<ServerState> for BsoBody {
"Invalid Content-Type".to_owned(),
RequestErrorLocation::Header,
Some("Content-Type".to_owned()),
).into(),
)
.into(),
));
}
}
@ -282,16 +288,19 @@ impl FromRequest<ServerState> for BsoBody {
e.to_string(),
RequestErrorLocation::Body,
Some("bso".to_owned()),
).into();
)
.into();
err.into()
}).and_then(move |bso: Json<BsoBody>| {
})
.and_then(move |bso: Json<BsoBody>| {
// Check the max payload size manually with our desired limit
if bso.payload.as_ref().map(|s| s.len()).unwrap_or_default() > max_payload_size {
let err: ApiError = ValidationErrorKind::FromDetails(
"payload too large".to_owned(),
RequestErrorLocation::Body,
Some("bso".to_owned()),
).into();
)
.into();
return future::err(err.into());
}
if let Err(e) = bso.validate() {
@ -330,7 +339,8 @@ impl FromRequest<ServerState> for BsoParam {
RequestErrorLocation::Path,
Some("bso".to_owned()),
)
})?.into_inner();
})?
.into_inner();
bso.validate().map_err(|e| {
ValidationErrorKind::FromValidationErrors(e, RequestErrorLocation::Path)
})?;
@ -362,7 +372,8 @@ impl FromRequest<ServerState> for CollectionParam {
RequestErrorLocation::Path,
Some("collection".to_owned()),
)
})?.into_inner();
})?
.into_inner();
collection.validate().map_err(|e| {
ValidationErrorKind::FromValidationErrors(e, RequestErrorLocation::Path)
})?;
@ -426,7 +437,8 @@ impl FromRequest<ServerState> for CollectionRequest {
"Invalid accept".to_string(),
RequestErrorLocation::Header,
Some("accept".to_string()),
).into());
)
.into());
}
None => ReplyFormat::Json,
};
@ -472,7 +484,8 @@ impl FromRequest<ServerState> for CollectionPostRequest {
CollectionParam,
BsoQueryParams,
BsoBodies,
)>::extract(&req).and_then(move |(user_id, db, collection, query, mut bsos)| {
)>::extract(&req)
.and_then(move |(user_id, db, collection, query, mut bsos)| {
let collection = collection.collection.clone();
if collection == "crypto" {
// Verify the client didn't mess up the crypto if we have a payload
@ -484,7 +497,8 @@ impl FromRequest<ServerState> for CollectionPostRequest {
"Known-bad BSO payload".to_owned(),
RequestErrorLocation::Body,
Some("bsos".to_owned()),
).into(),
)
.into(),
);
}
}
@ -579,7 +593,8 @@ impl FromRequest<ServerState> for BsoPutRequest {
BsoQueryParams,
BsoParam,
BsoBody,
)>::extract(req).and_then(|(user_id, db, collection, query, bso, body)| {
)>::extract(req)
.and_then(|(user_id, db, collection, query, bso, body)| {
let collection = collection.collection.clone();
if collection == "crypto" {
// Verify the client didn't mess up the crypto if we have a payload
@ -590,7 +605,8 @@ impl FromRequest<ServerState> for BsoPutRequest {
"Known-bad BSO payload".to_owned(),
RequestErrorLocation::Body,
Some("bsos".to_owned()),
).into(),
)
.into(),
);
}
}
@ -696,11 +712,11 @@ impl FromRequest<ServerState> for Box<dyn Db> {
#[serde(default)]
pub struct BsoQueryParams {
/// lower-bound on last-modified time
#[serde(deserialize_with = "deserialize_sync_timestamp",)]
#[serde(deserialize_with = "deserialize_sync_timestamp")]
pub newer: Option<SyncTimestamp>,
/// upper-bound on last-modified time
#[serde(deserialize_with = "deserialize_sync_timestamp",)]
#[serde(deserialize_with = "deserialize_sync_timestamp")]
pub older: Option<SyncTimestamp>,
/// order in which to return results (string)
@ -719,7 +735,7 @@ pub struct BsoQueryParams {
pub ids: Vec<String>,
// flag, whether to include full bodies (bool)
#[serde(deserialize_with = "deserialize_present_value",)]
#[serde(deserialize_with = "deserialize_present_value")]
pub full: bool,
}
@ -737,7 +753,8 @@ impl FromRequest<ServerState> for BsoQueryParams {
RequestErrorLocation::QueryString,
None,
)
})?.into_inner();
})?
.into_inner();
params.validate().map_err(|e| {
ValidationErrorKind::FromValidationErrors(e, RequestErrorLocation::QueryString)
})?;
@ -788,25 +805,28 @@ impl FromRequest<ServerState> for Option<BatchRequest> {
e.to_string(),
RequestErrorLocation::Header,
Some((*header).to_owned()),
).into();
)
.into();
err
})?,
None => continue,
};
let count = value.parse::<(u32)>().map_err(|_| {
let err: ApiError = ValidationErrorKind::FromDetails(
format!("Invalid integer value: {}", value),
RequestErrorLocation::Header,
Some((*header).to_owned()),
).into();
err
})?;
let err: ApiError = ValidationErrorKind::FromDetails(
format!("Invalid integer value: {}", value),
RequestErrorLocation::Header,
Some((*header).to_owned()),
)
.into();
err
})?;
if count > *limit {
return Err(ValidationErrorKind::FromDetails(
"size-limit-exceeded".to_owned(),
RequestErrorLocation::Header,
None,
).into())
)
.into());
}
}
@ -892,16 +912,20 @@ impl FromRequest<ServerState> for Option<PreConditionHeader> {
e.to_string(),
RequestErrorLocation::Header,
Some(field_name.to_owned()),
).into()
}).and_then(|v| {
)
.into()
})
.and_then(|v| {
SyncTimestamp::from_header(v).map_err(|e| {
ValidationErrorKind::FromDetails(
e.to_string(),
RequestErrorLocation::Header,
Some(field_name.to_owned()),
).into()
)
.into()
})
}).map(|v| {
})
.map(|v| {
let header = if field_name == "X-If-Modified-Since" {
PreConditionHeader::IfModifiedSince(v)
} else {

47
src/web/handlers.rs
View File

@ -98,12 +98,14 @@ pub fn delete_collection(coll: CollectionRequest) -> FutureResponse<HttpResponse
} else {
Box::new(future::err(e))
}
}).map_err(From::from)
})
.map_err(From::from)
.map(move |result| {
HttpResponse::Ok()
.if_true(delete_bsos, |resp| {
resp.header("X-Last-Modified", result.as_header());
}).json(result)
})
.json(result)
}),
)
}
@ -138,13 +140,15 @@ where
} else {
Err(e)
}
}).map_err(From::from)
})
.map_err(From::from)
.and_then(|result| {
coll.db
.extract_resource(coll.user_id, Some(coll.collection), None)
.map_err(From::from)
.map(move |ts| (result, ts))
}).map(move |(result, ts)| {
})
.map(move |(result, ts)| {
let mut builder = HttpResponse::build(StatusCode::OK);
let resp = builder
.header("X-Last-Modified", ts.as_header())
@ -182,7 +186,8 @@ pub fn post_collection(coll: CollectionPostRequest) -> FutureResponse<HttpRespon
collection: coll.collection,
bsos: coll.bsos.valid.into_iter().map(From::from).collect(),
failed: coll.bsos.invalid,
}).map_err(From::from)
})
.map_err(From::from)
.map(|result| {
HttpResponse::build(StatusCode::OK)
.header("X-Last-Modified", result.modified.as_header())
@ -198,8 +203,8 @@ pub fn post_collection_batch(coll: CollectionPostRequest) -> FutureResponse<Http
None => {
let err: DbError = DbErrorKind::BatchNotFound.into();
let err: ApiError = err.into();
return Box::new(future::err(err.into()))
},
return Box::new(future::err(err.into()));
}
};
let fut = if let Some(id) = breq.id {
@ -210,7 +215,8 @@ pub fn post_collection_batch(coll: CollectionPostRequest) -> FutureResponse<Http
user_id: coll.user_id.clone(),
collection: coll.collection.clone(),
id,
}).and_then(move |is_valid| {
})
.and_then(move |is_valid| {
if is_valid {
Box::new(future::ok(id))
} else {
@ -224,8 +230,7 @@ pub fn post_collection_batch(coll: CollectionPostRequest) -> FutureResponse<Http
user_id: coll.user_id.clone(),
collection: coll.collection.clone(),
bsos: vec![],
})
)
}))
};
let db = coll.db.clone();
@ -254,14 +259,13 @@ pub fn post_collection_batch(coll: CollectionPostRequest) -> FutureResponse<Http
if e.is_conflict() {
return future::err(e);
}
failed.extend(
bso_ids.into_iter().map(|id| (id, "db error".to_owned())),
)
failed.extend(bso_ids.into_iter().map(|id| (id, "db error".to_owned())))
}
};
future::ok((id, success, failed))
})
}).map_err(From::from);
})
.map_err(From::from);
Box::new(fut.and_then(move |(id, success, failed)| {
let mut resp = json!({
@ -279,7 +283,8 @@ pub fn post_collection_batch(coll: CollectionPostRequest) -> FutureResponse<Http
user_id: user_id.clone(),
collection: collection.clone(),
id,
}).and_then(move |batch| {
})
.and_then(move |batch| {
// TODO: validate *actual* sizes of the batch items
// (max_total_records, max_total_bytes)
if let Some(batch) = batch {
@ -292,7 +297,8 @@ pub fn post_collection_batch(coll: CollectionPostRequest) -> FutureResponse<Http
let err: DbError = DbErrorKind::BatchNotFound.into();
Box::new(future::err(err.into()))
}
}).map_err(From::from)
})
.map_err(From::from)
.map(|result| {
resp["modified"] = json!(result.modified);
HttpResponse::build(StatusCode::OK)
@ -311,7 +317,8 @@ pub fn delete_bso(bso_req: BsoRequest) -> FutureResponse<HttpResponse> {
user_id: bso_req.user_id,
collection: bso_req.collection,
id: bso_req.bso,
}).map_err(From::from)
})
.map_err(From::from)
.map(|result| HttpResponse::Ok().json(json!({ "modified": result }))),
)
}
@ -324,7 +331,8 @@ pub fn get_bso(bso_req: BsoRequest) -> FutureResponse<HttpResponse> {
user_id: bso_req.user_id,
collection: bso_req.collection,
id: bso_req.bso,
}).map_err(From::from)
})
.map_err(From::from)
.map(|result| {
result.map_or_else(
|| HttpResponse::NotFound().finish(),
@ -345,7 +353,8 @@ pub fn put_bso(bso_req: BsoPutRequest) -> FutureResponse<HttpResponse> {
sortindex: bso_req.body.sortindex,
payload: bso_req.body.payload,
ttl: bso_req.body.ttl,
}).map_err(From::from)
})
.map_err(From::from)
.map(|result| {
HttpResponse::build(StatusCode::OK)
.header("X-Last-Modified", result.as_header())

24
src/web/middleware.rs
View File

@ -46,14 +46,17 @@ impl<S> Middleware<S> for WeaveTimestamp {
let error: ApiError = ApiErrorKind::Internal(format!(
"Invalid X-Last-Modified response header: {}",
e
)).into();
))
.into();
error
})?.parse::<f64>()
})?
.parse::<f64>()
.map_err(|e| {
let error: ApiError = ApiErrorKind::Internal(format!(
"Invalid X-Last-Modified response header: {}",
e
)).into();
))
.into();
error
})?;
if resp_ts > ts {
@ -70,7 +73,8 @@ impl<S> Middleware<S> for WeaveTimestamp {
let error: ApiError = ApiErrorKind::Internal(format!(
"Invalid X-Weave-Timestamp response header: {}",
e
)).into();
))
.into();
error
})?,
);
@ -108,7 +112,8 @@ impl Middleware<ServerState> for DbTransaction {
match *req.method() {
Method::GET | Method::HEAD => db.lock_for_read(lc),
_ => db.lock_for_write(lc),
}.or_else(move |e| {
}
.or_else(move |e| {
// Middleware::response won't be called: rollback immediately
db2.rollback().and_then(|_| future::err(e))
}),
@ -123,7 +128,8 @@ impl Middleware<ServerState> for DbTransaction {
req.extensions_mut().insert((db, in_transaction));
future::ok(None)
})
}).map_err(Into::into);
})
.map_err(Into::into);
Ok(Started::Future(Box::new(fut)))
}
@ -185,7 +191,8 @@ impl Middleware<ServerState> for PreConditionCheck {
.header("X-Last-Modified", resource_ts.as_header())
.body(""); // 304 can't return any content
future::ok(Some(resp))
}).map_err(Into::into);
})
.map_err(Into::into);
Ok(Started::Future(Box::new(fut)))
}
@ -218,7 +225,8 @@ impl Middleware<ServerState> for PreConditionCheck {
resp.headers_mut().insert("X-Last-Modified", ts_header);
}
future::ok(resp)
}).map_err(Into::into);
})
.map_err(Into::into);
Ok(Response::Future(Box::new(fut)))
}
}