joyent_tokio_zookeeper/
transform.rs

1use failure::{bail, format_err};
2
3use crate::error;
4use crate::error::ZkError;
5use crate::proto::request::Request;
6use crate::proto::response::Response;
7use crate::types::acl::Acl;
8use crate::types::{MultiResponse, Stat};
9
10pub(crate) fn create(
11    res: Result<Response, ZkError>,
12) -> Result<Result<String, error::Create>, failure::Error> {
13    match res {
14        Ok(Response::String(s)) => Ok(Ok(s)),
15        Ok(r) => bail!("got non-string response to create: {:?}", r),
16        Err(ZkError::NoNode) => Ok(Err(error::Create::NoNode)),
17        Err(ZkError::NodeExists) => Ok(Err(error::Create::NodeExists)),
18        Err(ZkError::InvalidACL) => Ok(Err(error::Create::InvalidAcl)),
19        Err(ZkError::NoChildrenForEphemerals) => Ok(Err(error::Create::NoChildrenForEphemerals)),
20        Err(e) => Err(format_err!("create call failed: {:?}", e)),
21    }
22}
23
24pub(crate) fn set_data(
25    version: i32,
26    res: Result<Response, ZkError>,
27) -> Result<Result<Stat, error::SetData>, failure::Error> {
28    match res {
29        Ok(Response::Stat(stat)) => Ok(Ok(stat)),
30        Ok(r) => bail!("got a non-stat response to a set_data request: {:?}", r),
31        Err(ZkError::NoNode) => Ok(Err(error::SetData::NoNode)),
32        Err(ZkError::BadVersion) => Ok(Err(error::SetData::BadVersion { expected: version })),
33        Err(ZkError::NoAuth) => Ok(Err(error::SetData::NoAuth)),
34        Err(e) => bail!("set_data call failed: {:?}", e),
35    }
36}
37
38pub(crate) fn delete(
39    version: i32,
40    res: Result<Response, ZkError>,
41) -> Result<Result<(), error::Delete>, failure::Error> {
42    match res {
43        Ok(Response::Empty) => Ok(Ok(())),
44        Ok(r) => bail!("got non-empty response to delete: {:?}", r),
45        Err(ZkError::NoNode) => Ok(Err(error::Delete::NoNode)),
46        Err(ZkError::NotEmpty) => Ok(Err(error::Delete::NotEmpty)),
47        Err(ZkError::BadVersion) => Ok(Err(error::Delete::BadVersion { expected: version })),
48        Err(e) => Err(format_err!("delete call failed: {:?}", e)),
49    }
50}
51
52pub(crate) fn get_acl(
53    res: Result<Response, ZkError>,
54) -> Result<Result<(Vec<Acl>, Stat), error::GetAcl>, failure::Error> {
55    match res {
56        Ok(Response::GetAcl { acl, stat }) => Ok(Ok((acl, stat))),
57        Ok(r) => bail!("got non-acl response to a get_acl request: {:?}", r),
58        Err(ZkError::NoNode) => Ok(Err(error::GetAcl::NoNode)),
59        Err(e) => Err(format_err!("get_acl call failed: {:?}", e)),
60    }
61}
62
63pub(crate) fn set_acl(
64    version: i32,
65    res: Result<Response, ZkError>,
66) -> Result<Result<Stat, error::SetAcl>, failure::Error> {
67    match res {
68        Ok(Response::Stat(stat)) => Ok(Ok(stat)),
69        Ok(r) => bail!("got non-stat response to a set_acl request: {:?}", r),
70        Err(ZkError::NoNode) => Ok(Err(error::SetAcl::NoNode)),
71        Err(ZkError::BadVersion) => Ok(Err(error::SetAcl::BadVersion { expected: version })),
72        Err(ZkError::InvalidACL) => Ok(Err(error::SetAcl::InvalidAcl)),
73        Err(ZkError::NoAuth) => Ok(Err(error::SetAcl::NoAuth)),
74        Err(e) => Err(format_err!("set_acl call failed: {:?}", e)),
75    }
76}
77
78pub(crate) fn exists(res: Result<Response, ZkError>) -> Result<Option<Stat>, failure::Error> {
79    match res {
80        Ok(Response::Stat(stat)) => Ok(Some(stat)),
81        Ok(r) => bail!("got a non-create response to a create request: {:?}", r),
82        Err(ZkError::NoNode) => Ok(None),
83        Err(e) => bail!("exists call failed: {:?}", e),
84    }
85}
86
87pub(crate) fn get_children(
88    res: Result<Response, ZkError>,
89) -> Result<Option<Vec<String>>, failure::Error> {
90    match res {
91        Ok(Response::Strings(children)) => Ok(Some(children)),
92        Ok(r) => bail!("got non-strings response to get-children: {:?}", r),
93        Err(ZkError::NoNode) => Ok(None),
94        Err(e) => Err(format_err!("get-children call failed: {:?}", e)),
95    }
96}
97
98pub(crate) fn get_data(
99    res: Result<Response, ZkError>,
100) -> Result<Option<(Vec<u8>, Stat)>, failure::Error> {
101    match res {
102        Ok(Response::GetData { bytes, stat }) => Ok(Some((bytes, stat))),
103        Ok(r) => bail!("got non-data response to get-data: {:?}", r),
104        Err(ZkError::NoNode) => Ok(None),
105        Err(e) => Err(format_err!("get-data call failed: {:?}", e)),
106    }
107}
108
109pub(crate) fn check(
110    version: i32,
111    res: Result<Response, ZkError>,
112) -> Result<Result<(), error::Check>, failure::Error> {
113    match res {
114        Ok(Response::Empty) => Ok(Ok(())),
115        Ok(r) => bail!("got a non-check response to a check request: {:?}", r),
116        Err(ZkError::NoNode) => Ok(Err(error::Check::NoNode)),
117        Err(ZkError::BadVersion) => Ok(Err(error::Check::BadVersion { expected: version })),
118        Err(e) => bail!("check call failed: {:?}", e),
119    }
120}
121
122/// The subset of [`proto::Request`] that a multi request needs to retain.
123///
124/// In order to properly handle errors, a multi request needs to retain the
125/// expected version for each constituent set data, delete, or check operation.
126/// Unfortunately, executing a multi request requires transferring ownership of
127/// the `proto::Request`, which contains this information, to the future. A
128/// `RequestMarker` is used to avoid cloning the whole `proto::Request`, which
129/// can be rather large, when only the version information is necessary.
130#[derive(Debug)]
131pub(crate) enum RequestMarker {
132    Create,
133    SetData { version: i32 },
134    Delete { version: i32 },
135    Check { version: i32 },
136}
137
138impl From<&Request> for RequestMarker {
139    fn from(r: &Request) -> RequestMarker {
140        match r {
141            Request::Create { .. } => RequestMarker::Create,
142            Request::SetData { version, .. } => RequestMarker::SetData { version: *version },
143            Request::Delete { version, .. } => RequestMarker::Delete { version: *version },
144            Request::Check { version, .. } => RequestMarker::Check { version: *version },
145            _ => unimplemented!(),
146        }
147    }
148}
149
150pub(crate) fn multi(
151    req: &RequestMarker,
152    res: Result<Response, ZkError>,
153) -> Result<Result<MultiResponse, error::Multi>, failure::Error> {
154    // Handle multi-specific errors.
155    match res {
156        Err(ZkError::Ok) => return Ok(Err(error::Multi::RolledBack)),
157        // Confusingly, the ZooKeeper server uses RuntimeInconsistency to
158        // indicate that a request in a multi batch was skipped because an
159        // earlier request in the batch failed.
160        // Source: https://github.com/apache/zookeeper/blob/372e713a9/zookeeper-server/src/main/java/org/apache/zookeeper/server/DataTree.java#L945-L946
161        Err(ZkError::RuntimeInconsistency) => return Ok(Err(error::Multi::Skipped)),
162        _ => (),
163    };
164
165    Ok(match req {
166        RequestMarker::Create => create(res)?
167            .map(MultiResponse::Create)
168            .map_err(|err| err.into()),
169        RequestMarker::SetData { version } => set_data(*version, res)?
170            .map(MultiResponse::SetData)
171            .map_err(|err| err.into()),
172        RequestMarker::Delete { version } => delete(*version, res)?
173            .map(|_| MultiResponse::Delete)
174            .map_err(|err| err.into()),
175        RequestMarker::Check { version } => check(*version, res)?
176            .map(|_| MultiResponse::Check)
177            .map_err(|err| err.into()),
178    })
179}
joyent_tokio_zookeeper/transform.rs

joyent_tokio_zookeeper/
transform.rs
