taba_client/client/
mod.rs

1//! The client: the configuration, the call with the common contract's retries (doc 19, section 18.4) and the
2//! nine operations of the surface (section 18.3).
3
4use std::fmt;
5use std::sync::Arc;
6use std::time::Duration;
7
8use crate::base64;
9use crate::http::{self, Request, Response, Url};
10use crate::json::{self, Fields};
11use crate::transport::{TcpTransport, Timeouts, Transport, TransportError};
12
13/// The version of the API contract this connector implements (UC-CNT-028).
14pub const CONTRACT_VERSION: &str = "1";
15
16/// Above this, `consume` waits no more than the `waitTime` plus the slack below.
17const LONG_POLL_SLACK: Duration = Duration::from_secs(10);
18
19#[derive(Debug, Clone)]
20pub struct Config {
21    /// Address of the client API (`http://host:port` or `https://...`).
22    pub base_url: String,
23    pub token: Option<String>,
24    pub connect_timeout: Duration,
25    pub request_timeout: Duration,
26    /// Attempts, counting the first.
27    pub max_attempts: u32,
28    pub user_agent: String,
29    /// CA certificate (PEM) to trust, besides the system roots.
30    pub ca_pem: Option<Vec<u8>>,
31}
32
33impl Config {
34    pub fn new(base_url: impl Into<String>) -> Config {
35        Config {
36            base_url: base_url.into(),
37            token: None,
38            connect_timeout: Duration::from_secs(5),
39            request_timeout: Duration::from_secs(30),
40            max_attempts: 3,
41            user_agent: format!("taba-rust/{}", env!("CARGO_PKG_VERSION")),
42            ca_pem: None,
43        }
44    }
45}
46
47/// The only kind of error of the connector: the API `ErrorResponse`, or a failure of the client itself
48/// (`status` zero and `code` `client.network`, `client.timeout` or `client.invalid-config`).
49#[derive(Debug, Clone, PartialEq, Eq)]
50pub struct Error {
51    pub status: u16,
52    pub code: String,
53    pub message: String,
54    pub request_id: Option<String>,
55    /// Seconds of the `Retry-After`, if it came.
56    pub retry_after: Option<u64>,
57    /// The `details` list of the `ErrorResponse`, as JSON text.
58    pub details: Option<String>,
59    /// The response body, as it came (empty on a client failure).
60    pub body: Vec<u8>,
61}
62
63impl Error {
64    /// An error of the client itself (`status` zero), for whoever uses the connector to build their own.
65    pub fn from_message(code: &str, message: impl Into<String>) -> Error {
66        Error::client(code, message)
67    }
68
69    fn client(code: &str, message: impl Into<String>) -> Error {
70        Error {
71            status: 0,
72            code: code.to_owned(),
73            message: message.into(),
74            request_id: None,
75            retry_after: None,
76            details: None,
77            body: Vec::new(),
78        }
79    }
80
81    /// A failure of the client itself, not a server response.
82    pub fn is_client_side(&self) -> bool {
83        self.status == 0
84    }
85}
86
87impl fmt::Display for Error {
88    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
89        write!(f, "{}: {}", self.code, self.message)?;
90        if let Some(id) = &self.request_id {
91            write!(f, " (requestId {id})")?;
92        }
93        Ok(())
94    }
95}
96
97impl std::error::Error for Error {}
98
99/// A call to the API, before it becomes HTTP.
100#[derive(Debug, Clone)]
101pub struct Call {
102    pub method: &'static str,
103    /// Path with the segments already encoded (`/topics/orders`).
104    pub path: String,
105    pub query: Vec<(String, String)>,
106    pub body: Vec<u8>,
107    /// Only applies with a body; default `application/json`.
108    /// Extra headers (an `Accept` of its own replaces the default `application/json`).
109    pub headers: Vec<(String, String)>,
110    pub content_type: Option<String>,
111    pub request_id: Option<String>,
112    pub traceparent: Option<String>,
113    /// Overrides the total time of the call (`consume` uses `waitTime + 10 s`).
114    pub timeout: Option<Duration>,
115    /// Retries `502` and `504`: only publishing to a FIFO queue, where the primary deduplicates.
116    pub retry_gateway: bool,
117    /// After a `503`, these responses count as success (the operation had already been applied):
118    /// creating a topic (`409 topic.already-exists`) and deleting (`404`).
119    pub recover: Vec<(u16, &'static str)>,
120}
121
122impl Call {
123    pub fn new(method: &'static str, path: impl Into<String>) -> Call {
124        Call {
125            method,
126            path: path.into(),
127            query: Vec::new(),
128            body: Vec::new(),
129            headers: Vec::new(),
130            content_type: None,
131            request_id: None,
132            traceparent: None,
133            timeout: None,
134            retry_gateway: false,
135            recover: Vec::new(),
136        }
137    }
138
139    pub fn query(mut self, key: &str, value: impl ToString) -> Call {
140        self.query.push((key.to_owned(), value.to_string()));
141        self
142    }
143
144    pub fn json_body(mut self, body: String) -> Call {
145        self.body = body.into_bytes();
146        self
147    }
148}
149
150/// The response of a call that succeeded.
151#[derive(Debug, Clone)]
152pub struct Reply {
153    pub status: u16,
154    pub headers: Vec<(String, String)>,
155    pub body: Vec<u8>,
156    pub request_id: Option<String>,
157    /// `true` when the response is a `409`/`404` accepted as success after a `503` (the operation
158    /// had been applied on the previous attempt).
159    pub recovered: bool,
160}
161
162impl Reply {
163    pub fn text(&self) -> String {
164        String::from_utf8_lossy(&self.body).into_owned()
165    }
166}
167
168type Sleeper = Arc<dyn Fn(Duration) + Send + Sync>;
169
170pub struct Client {
171    config: Config,
172    url: Url,
173    transport: Box<dyn Transport>,
174    sleeper: Sleeper,
175}
176
177impl Client {
178    pub fn new(config: Config) -> Result<Client, Error> {
179        let transport = TcpTransport::new(config.ca_pem.clone());
180        Client::with_transport(config, Box::new(transport), Arc::new(std::thread::sleep))
181    }
182
183    /// To swap the network stack (and, in tests, the wait between attempts).
184    pub fn with_transport(
185        config: Config,
186        transport: Box<dyn Transport>,
187        sleeper: Sleeper,
188    ) -> Result<Client, Error> {
189        let url =
190            Url::parse(&config.base_url).map_err(|m| Error::client("client.invalid-config", m))?;
191        Ok(Client {
192            config,
193            url,
194            transport,
195            sleeper,
196        })
197    }
198
199    pub fn config(&self) -> &Config {
200        &self.config
201    }
202
203    fn wire_request(&self, call: &Call) -> Request {
204        let mut headers = Vec::new();
205        if let Some(token) = &self.config.token {
206            headers.push(("Authorization".to_owned(), format!("Bearer {token}")));
207        }
208        if !call
209            .headers
210            .iter()
211            .any(|(k, _)| k.eq_ignore_ascii_case("accept"))
212        {
213            headers.push(("Accept".to_owned(), "application/json".to_owned()));
214        }
215        headers.extend(call.headers.iter().cloned());
216        headers.push(("User-Agent".to_owned(), self.config.user_agent.clone()));
217        if !call.body.is_empty() {
218            headers.push((
219                "Content-Type".to_owned(),
220                call.content_type
221                    .clone()
222                    .unwrap_or_else(|| "application/json".to_owned()),
223            ));
224        }
225        if let Some(id) = &call.request_id {
226            headers.push(("X-Request-Id".to_owned(), id.clone()));
227        }
228        if let Some(tp) = &call.traceparent {
229            headers.push(("traceparent".to_owned(), tp.clone()));
230        }
231        Request {
232            method: call.method.to_owned(),
233            target: http::build_target(&self.url.base_path, &call.path, &call.query),
234            headers,
235            body: call.body.clone(),
236        }
237    }
238
239    /// Makes the call, retrying only what the contract says to retry (section 18.4):
240    /// `429` and `503` (waiting for the `Retry-After`, or 1 s, 2 s, 4 s), `502` and `504` only with
241    /// `retry_gateway`, and a network failure **before** sending. Never a `4xx` (except `429`), nor
242    /// a failure after sending the body, nor a timeout.
243    pub fn call(&self, call: &Call) -> Result<Reply, Error> {
244        let request = self.wire_request(call);
245        let timeouts = Timeouts {
246            connect: self.config.connect_timeout,
247            total: call.timeout.unwrap_or(self.config.request_timeout),
248        };
249        let max = self.config.max_attempts.max(1);
250        let mut attempt = 1;
251        let mut after_503 = false;
252        loop {
253            let backoff = Duration::from_secs(1 << (attempt - 1).min(6));
254            match self.transport.send(&self.url, &request, &timeouts) {
255                Ok(response) => {
256                    if response.status < 400 {
257                        return Ok(reply(response, false));
258                    }
259                    if after_503
260                        && call.recover.iter().any(|(status, code)| {
261                            *status == response.status
262                                && error_code(&response).as_deref() == Some(*code)
263                        })
264                    {
265                        return Ok(reply(response, true));
266                    }
267                    let error = error_from(&response);
268                    let retry = match response.status {
269                        429 | 503 => true,
270                        502 | 504 => call.retry_gateway,
271                        _ => false,
272                    };
273                    if !retry || attempt >= max {
274                        return Err(error);
275                    }
276                    after_503 |= response.status == 503;
277                    (self.sleeper)(
278                        error
279                            .retry_after
280                            .map(Duration::from_secs)
281                            .unwrap_or(backoff),
282                    );
283                }
284                Err(TransportError::Connect(message)) => {
285                    if attempt >= max {
286                        return Err(Error::client("client.network", message));
287                    }
288                    after_503 = true;
289                    (self.sleeper)(backoff);
290                }
291                Err(TransportError::Timeout { sent: false }) => {
292                    if attempt >= max {
293                        return Err(Error::client(
294                            "client.timeout",
295                            "timed out opening the connection",
296                        ));
297                    }
298                    after_503 = true;
299                    (self.sleeper)(backoff);
300                }
301                Err(TransportError::Timeout { sent: true }) => {
302                    return Err(Error::client(
303                        "client.timeout",
304                        "timed out waiting for the response",
305                    ));
306                }
307                Err(TransportError::Io(message)) => {
308                    return Err(Error::client("client.network", message))
309                }
310                Err(TransportError::Protocol(message)) => {
311                    return Err(Error::client(
312                        "client.network",
313                        format!("invalid response: {message}"),
314                    ))
315                }
316            }
317            attempt += 1;
318        }
319    }
320}
321
322fn reply(response: Response, recovered: bool) -> Reply {
323    Reply {
324        request_id: response.header("x-request-id").map(str::to_owned),
325        status: response.status,
326        headers: response.headers,
327        body: response.body,
328        recovered,
329    }
330}
331
332fn error_code(response: &Response) -> Option<String> {
333    let text = std::str::from_utf8(&response.body).ok()?;
334    Fields::parse(text).ok()?.string("code")
335}
336
337/// The response's `ErrorResponse`; if the body is not one, the error carries the code `http.<status>`.
338fn error_from(response: &Response) -> Error {
339    let text = String::from_utf8_lossy(&response.body);
340    let fields = Fields::parse(&text).ok();
341    let header_request_id = response.header("x-request-id").map(str::to_owned);
342    let (code, message, request_id, details) = match &fields {
343        Some(f) if f.string("code").is_some() => (
344            f.string("code").unwrap_or_default(),
345            f.string("message").unwrap_or_default(),
346            f.string("requestId").or(header_request_id),
347            f.raw("details").map(str::to_owned),
348        ),
349        _ => (
350            format!("http.{}", response.status),
351            if text.trim().is_empty() {
352                format!("response {} with no body", response.status)
353            } else {
354                text.trim().chars().take(300).collect()
355            },
356            header_request_id,
357            None,
358        ),
359    };
360    Error {
361        status: response.status,
362        code,
363        message,
364        request_id,
365        retry_after: response
366            .header("retry-after")
367            .and_then(|v| v.trim().parse().ok()),
368        details,
369        body: response.body.clone(),
370    }
371}
372
373// ---------------------------------------------------------------------------------------------
374// Modelos
375// ---------------------------------------------------------------------------------------------
376
377fn bad_response(what: &str, e: json::JsonError) -> Error {
378    Error::client("client.invalid-response", format!("{what}: {e}"))
379}
380
381#[derive(Debug, Clone, PartialEq, Eq)]
382pub struct Topic {
383    pub id: String,
384    pub name: String,
385    /// `queue` or `distributor`.
386    pub topic_type: String,
387    pub created_at: String,
388    pub fifo: Option<bool>,
389    pub volatile: Option<bool>,
390    /// The topic JSON, as the server returned it.
391    pub raw: String,
392}
393
394impl Topic {
395    pub fn parse(raw: &str) -> Result<Topic, Error> {
396        let f = Fields::parse(raw).map_err(|e| bad_response("topic", e))?;
397        Ok(Topic {
398            id: f.string("id").unwrap_or_default(),
399            name: f.string("name").unwrap_or_default(),
400            topic_type: f.string("type").unwrap_or_default(),
401            created_at: f.string("createdAt").unwrap_or_default(),
402            fifo: f.boolean("fifo"),
403            volatile: f.boolean("volatile"),
404            raw: raw.to_owned(),
405        })
406    }
407}
408
409/// A page (`Page<T>`): the items and the cursor of the next one, if any.
410#[derive(Debug, Clone, PartialEq, Eq)]
411pub struct Page<T> {
412    pub items: Vec<T>,
413    pub next_cursor: Option<String>,
414    pub raw: String,
415}
416
417fn parse_page<T>(raw: &str, item: impl Fn(&str) -> Result<T, Error>) -> Result<Page<T>, Error> {
418    let f = Fields::parse(raw).map_err(|e| bad_response("page", e))?;
419    let items_raw = f.raw("items").unwrap_or("[]");
420    let items = json::array_items(items_raw)
421        .map_err(|e| bad_response("page items", e))?
422        .into_iter()
423        .map(item)
424        .collect::<Result<Vec<_>, _>>()?;
425    Ok(Page {
426        items,
427        next_cursor: f.string("nextCursor"),
428        raw: raw.to_owned(),
429    })
430}
431
432/// A delivered message. The `body` always comes out as bytes (section 18.4).
433#[derive(Debug, Clone, PartialEq, Eq)]
434pub struct Message {
435    pub id: String,
436    pub body: Vec<u8>,
437    pub content_type: String,
438    pub receipt_token: String,
439    pub delivery_count: u64,
440    pub published_at: String,
441    pub invisible_until: String,
442    pub group_id: Option<String>,
443    pub deduplication_id: Option<String>,
444    pub dead_lettered_at: Option<String>,
445    pub dead_letter_reason: Option<String>,
446    /// The message JSON, as the server returned it.
447    pub raw: String,
448}
449
450fn is_json_type(content_type: &str) -> bool {
451    let media = content_type
452        .split(';')
453        .next()
454        .unwrap_or_default()
455        .trim()
456        .to_ascii_lowercase();
457    media == "application/json" || media.ends_with("+json")
458}
459
460impl Message {
461    pub fn parse(raw: &str) -> Result<Message, Error> {
462        let f = Fields::parse(raw).map_err(|e| bad_response("message", e))?;
463        let content_type = f
464            .string("contentType")
465            .unwrap_or_else(|| "application/json".to_owned());
466        let body = match f.raw("body") {
467            None => Vec::new(),
468            Some(body) if f.string("bodyEncoding").as_deref() == Some("base64") => {
469                let text =
470                    json::decode_string(body).map_err(|e| bad_response("message body", e))?;
471                base64::decode(&text).ok_or_else(|| {
472                    Error::client("client.invalid-response", "invalid base64 body")
473                })?
474            }
475            Some(body) if is_json_type(&content_type) => body.as_bytes().to_vec(),
476            Some(body) => json::decode_string(body)
477                .map_err(|e| bad_response("message body", e))?
478                .into_bytes(),
479        };
480        Ok(Message {
481            id: f.string("id").unwrap_or_default(),
482            body,
483            content_type,
484            receipt_token: f.string("receiptToken").unwrap_or_default(),
485            delivery_count: f.unsigned("deliveryCount").unwrap_or(0),
486            published_at: f.string("publishedAt").unwrap_or_default(),
487            invisible_until: f.string("invisibleUntil").unwrap_or_default(),
488            group_id: f.string("groupId"),
489            deduplication_id: f.string("deduplicationId"),
490            dead_lettered_at: f.string("deadLetteredAt"),
491            dead_letter_reason: f.string("deadLetterReason"),
492            raw: raw.to_owned(),
493        })
494    }
495}
496
497#[derive(Debug, Clone, PartialEq, Eq)]
498pub struct PublishResponse {
499    /// The message `id` (queue). Absent from a distributor's response.
500    pub id: Option<String>,
501    /// FIFO queue only.
502    pub duplicate: Option<bool>,
503    /// Distributor only: how many queues matched.
504    pub matched: Option<u64>,
505    pub raw: String,
506}
507
508#[derive(Debug, Clone, PartialEq, Eq)]
509pub struct VisibilityResponse {
510    pub id: String,
511    pub invisible_until: String,
512    pub raw: String,
513}
514
515// ---------------------------------------------------------------------------------------------
516// Requests
517// ---------------------------------------------------------------------------------------------
518
519fn json_map(pairs: &[(String, String)]) -> String {
520    let items: Vec<String> = pairs
521        .iter()
522        .map(|(k, v)| format!("{}:{}", json::quote(k), json::quote(v)))
523        .collect();
524    format!("{{{}}}", items.join(","))
525}
526
527fn json_list(items: &[String]) -> String {
528    let items: Vec<String> = items.iter().map(|i| json::quote(i)).collect();
529    format!("[{}]", items.join(","))
530}
531
532/// `POST /topics`. The connector sends only what was filled in; whatever does not apply to the type, the server
533/// refuses.
534#[derive(Debug, Clone, Default)]
535pub struct TopicRequest {
536    pub name: String,
537    /// `queue` or `distributor`.
538    pub topic_type: String,
539    pub tags: Vec<(String, String)>,
540    pub prefixes: Vec<String>,
541    pub volatile: Option<bool>,
542    pub idle_timeout: Option<u64>,
543    pub fifo: Option<bool>,
544    pub max_message_bytes: Option<u64>,
545    pub ttl: Option<u64>,
546    pub max_deliveries: Option<u64>,
547}
548
549impl TopicRequest {
550    pub fn to_json(&self) -> String {
551        let mut fields = vec![
552            format!("\"name\":{}", json::quote(&self.name)),
553            format!("\"type\":{}", json::quote(&self.topic_type)),
554        ];
555        if !self.tags.is_empty() {
556            fields.push(format!("\"tags\":{}", json_map(&self.tags)));
557        }
558        if !self.prefixes.is_empty() {
559            fields.push(format!("\"prefixes\":{}", json_list(&self.prefixes)));
560        }
561        if let Some(v) = self.volatile {
562            fields.push(format!("\"volatile\":{v}"));
563        }
564        if let Some(v) = self.idle_timeout {
565            fields.push(format!("\"idleTimeout\":{v}"));
566        }
567        if let Some(v) = self.fifo {
568            fields.push(format!("\"fifo\":{v}"));
569        }
570        if let Some(v) = self.max_message_bytes {
571            fields.push(format!("\"maxMessageBytes\":{v}"));
572        }
573        if let Some(v) = self.ttl {
574            fields.push(format!("\"ttl\":{v}"));
575        }
576        if let Some(v) = self.max_deliveries {
577            fields.push(format!("\"maxDeliveries\":{v}"));
578        }
579        format!("{{{}}}", fields.join(","))
580    }
581}
582
583/// A `PATCH` field that accepts going back to the server default (`null`).
584#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
585pub enum Nullable<T> {
586    /// Does not send the field.
587    #[default]
588    Keep,
589    /// Sends `null`: goes back to the server default.
590    Default,
591    Set(T),
592}
593
594impl<T: fmt::Display> Nullable<T> {
595    fn push(&self, name: &str, fields: &mut Vec<String>) {
596        match self {
597            Nullable::Keep => {}
598            Nullable::Default => fields.push(format!("\"{name}\":null")),
599            Nullable::Set(v) => fields.push(format!("\"{name}\":{v}")),
600        }
601    }
602}
603
604/// `PATCH /topics/{topic}`: only what was filled in is sent.
605#[derive(Debug, Clone, Default)]
606pub struct TopicUpdate {
607    /// Replaces the whole map (`Some(empty)` clears the tags).
608    pub tags: Option<Vec<(String, String)>>,
609    pub prefixes: Option<Vec<String>>,
610    pub idle_timeout: Option<u64>,
611    pub max_message_bytes: Nullable<u64>,
612    pub ttl: Nullable<u64>,
613    pub max_deliveries: Nullable<u64>,
614}
615
616impl TopicUpdate {
617    pub fn to_json(&self) -> String {
618        let mut fields = Vec::new();
619        if let Some(tags) = &self.tags {
620            fields.push(format!("\"tags\":{}", json_map(tags)));
621        }
622        if let Some(prefixes) = &self.prefixes {
623            fields.push(format!("\"prefixes\":{}", json_list(prefixes)));
624        }
625        if let Some(v) = self.idle_timeout {
626            fields.push(format!("\"idleTimeout\":{v}"));
627        }
628        self.max_message_bytes.push("maxMessageBytes", &mut fields);
629        self.ttl.push("ttl", &mut fields);
630        self.max_deliveries.push("maxDeliveries", &mut fields);
631        format!("{{{}}}", fields.join(","))
632    }
633}
634
635#[derive(Debug, Clone, Default)]
636pub struct PublishOptions {
637    /// Default `application/json`.
638    pub content_type: Option<String>,
639    pub group_id: Option<String>,
640    pub deduplication_id: Option<String>,
641    pub request_id: Option<String>,
642    /// Set it when the topic is a FIFO queue: only there can `502`/`504` be retried.
643    pub fifo: bool,
644}
645
646#[derive(Debug, Clone, Default)]
647pub struct ConsumeOptions {
648    pub max_messages: Option<u32>,
649    pub visibility_timeout: Option<u32>,
650    pub wait_time: Option<u32>,
651    pub auto_create: bool,
652    /// The next four only apply with `auto_create`.
653    pub volatile: Option<bool>,
654    pub idle_timeout: Option<u32>,
655    pub fifo: Option<bool>,
656    pub tags: Vec<(String, String)>,
657    pub request_id: Option<String>,
658}
659
660fn segment(s: &str) -> String {
661    http::percent_encode(s)
662}
663
664// ---------------------------------------------------------------------------------------------
665// The nine operations
666// ---------------------------------------------------------------------------------------------
667
668impl Client {
669    /// `POST /topics`.
670    pub fn create_topic(&self, request: &TopicRequest) -> Result<Topic, Error> {
671        let mut call = Call::new("POST", "/topics").json_body(request.to_json());
672        call.recover = vec![(409, "topic.already-exists")];
673        let reply = self.call(&call)?;
674        if reply.recovered {
675            return self.get_topic(&request.name);
676        }
677        Topic::parse(&reply.text())
678    }
679
680    /// `GET /topics/{topic}`.
681    pub fn get_topic(&self, name: &str) -> Result<Topic, Error> {
682        let reply = self.call(&Call::new("GET", format!("/topics/{}", segment(name))))?;
683        Topic::parse(&reply.text())
684    }
685
686    /// `GET /topics`. Parameters that are `None` do not go in the query.
687    pub fn list_topics(
688        &self,
689        limit: Option<u32>,
690        cursor: Option<&str>,
691        name: Option<&str>,
692    ) -> Result<Page<Topic>, Error> {
693        let mut call = Call::new("GET", "/topics");
694        if let Some(limit) = limit {
695            call = call.query("limit", limit);
696        }
697        if let Some(cursor) = cursor {
698            call = call.query("cursor", cursor);
699        }
700        if let Some(name) = name {
701            call = call.query("name", name);
702        }
703        parse_page(&self.call(&call)?.text(), Topic::parse)
704    }
705
706    /// All topics, page by page.
707    pub fn list_all_topics(&self, name: Option<&str>) -> Result<Vec<Topic>, Error> {
708        let mut all = Vec::new();
709        let mut cursor: Option<String> = None;
710        loop {
711            let page = self.list_topics(None, cursor.as_deref(), name)?;
712            all.extend(page.items);
713            match page.next_cursor {
714                Some(next) => cursor = Some(next),
715                None => return Ok(all),
716            }
717        }
718    }
719
720    /// `PATCH /topics/{topic}`.
721    pub fn update_topic(&self, name: &str, update: &TopicUpdate) -> Result<Topic, Error> {
722        let call =
723            Call::new("PATCH", format!("/topics/{}", segment(name))).json_body(update.to_json());
724        Topic::parse(&self.call(&call)?.text())
725    }
726
727    /// `DELETE /topics/{topic}`.
728    pub fn delete_topic(&self, name: &str) -> Result<(), Error> {
729        let mut call = Call::new("DELETE", format!("/topics/{}", segment(name)));
730        call.recover = vec![(404, "topic.not-found")];
731        self.call(&call).map(|_| ())
732    }
733
734    /// `POST /topics/{topic}/messages`. The body goes as bytes, without validation.
735    pub fn publish(
736        &self,
737        topic: &str,
738        body: &[u8],
739        options: &PublishOptions,
740    ) -> Result<PublishResponse, Error> {
741        let mut call = Call::new("POST", format!("/topics/{}/messages", segment(topic)));
742        call.body = body.to_vec();
743        call.content_type = Some(
744            options
745                .content_type
746                .clone()
747                .unwrap_or_else(|| "application/json".to_owned()),
748        );
749        if let Some(group) = &options.group_id {
750            call = call.query("groupId", group);
751        }
752        if let Some(dedup) = &options.deduplication_id {
753            call = call.query("deduplicationId", dedup);
754        }
755        call.request_id = options.request_id.clone();
756        call.retry_gateway = options.fifo;
757        // A publication with an empty body still needs a `Content-Type`: the body goes as it came.
758        let raw = self.call(&call)?.text();
759        let f = Fields::parse(&raw).map_err(|e| bad_response("publication", e))?;
760        Ok(PublishResponse {
761            id: f.string("id"),
762            duplicate: f.boolean("duplicate"),
763            matched: f.unsigned("matched"),
764            raw,
765        })
766    }
767
768    pub fn publish_json(&self, topic: &str, json_text: &str) -> Result<PublishResponse, Error> {
769        self.publish(topic, json_text.as_bytes(), &PublishOptions::default())
770    }
771
772    pub fn publish_text(&self, topic: &str, text: &str) -> Result<PublishResponse, Error> {
773        let options = PublishOptions {
774            content_type: Some("text/plain; charset=utf-8".to_owned()),
775            ..PublishOptions::default()
776        };
777        self.publish(topic, text.as_bytes(), &options)
778    }
779
780    /// `GET /topics/{topic}/messages`. The read time is `waitTime + 10 s`, not the
781    /// `requestTimeout`.
782    pub fn consume(&self, topic: &str, options: &ConsumeOptions) -> Result<Vec<Message>, Error> {
783        let call = self.consume_call(topic, options);
784        let raw = self.call(&call)?.text();
785        let f = Fields::parse(&raw).map_err(|e| bad_response("consumo", e))?;
786        json::array_items(f.raw("messages").unwrap_or("[]"))
787            .map_err(|e| bad_response("messages", e))?
788            .into_iter()
789            .map(Message::parse)
790            .collect()
791    }
792
793    pub fn consume_call(&self, topic: &str, options: &ConsumeOptions) -> Call {
794        let mut call = Call::new("GET", format!("/topics/{}/messages", segment(topic)));
795        if let Some(v) = options.max_messages {
796            call = call.query("maxMessages", v);
797        }
798        if let Some(v) = options.visibility_timeout {
799            call = call.query("visibilityTimeout", v);
800        }
801        if let Some(v) = options.wait_time {
802            call = call.query("waitTime", v);
803        }
804        if options.auto_create {
805            call = call.query("autoCreate", true);
806            if let Some(v) = options.volatile {
807                call = call.query("volatile", v);
808            }
809            if let Some(v) = options.idle_timeout {
810                call = call.query("idleTimeout", v);
811            }
812            if let Some(v) = options.fifo {
813                call = call.query("fifo", v);
814            }
815            for (k, v) in &options.tags {
816                call = call.query(&format!("tag.{k}"), v);
817            }
818        }
819        call.request_id = options.request_id.clone();
820        call.timeout =
821            Some(Duration::from_secs(u64::from(options.wait_time.unwrap_or(0))) + LONG_POLL_SLACK);
822        call
823    }
824
825    /// `DELETE /topics/{topic}/messages/{id}?receiptToken=`.
826    pub fn ack(&self, topic: &str, message_id: &str, receipt_token: &str) -> Result<(), Error> {
827        let call = Call::new(
828            "DELETE",
829            format!(
830                "/topics/{}/messages/{}",
831                segment(topic),
832                segment(message_id)
833            ),
834        )
835        .query("receiptToken", receipt_token);
836        self.call(&call).map(|_| ())
837    }
838
839    /// `PUT /topics/{topic}/messages/{id}/visibility?receiptToken=`.
840    pub fn change_visibility(
841        &self,
842        topic: &str,
843        message_id: &str,
844        receipt_token: &str,
845        seconds: u32,
846    ) -> Result<VisibilityResponse, Error> {
847        let call = Call::new(
848            "PUT",
849            format!(
850                "/topics/{}/messages/{}/visibility",
851                segment(topic),
852                segment(message_id)
853            ),
854        )
855        .query("receiptToken", receipt_token)
856        .json_body(format!("{{\"visibilityTimeout\":{seconds}}}"));
857        let raw = self.call(&call)?.text();
858        let f = Fields::parse(&raw).map_err(|e| bad_response("invisibilidade", e))?;
859        Ok(VisibilityResponse {
860            id: f.string("id").unwrap_or_default(),
861            invisible_until: f.string("invisibleUntil").unwrap_or_default(),
862            raw,
863        })
864    }
865
866    /// The consume loop (D71.1): long polling (`waitTime` 20 and one message at a time, if not
867    /// set), calls the `handler` and acknowledges by itself **only if it returns `Ok`**. `stop` is
868    /// checked before each consume and after each message; returning `true`, the loop ends.
869    /// A server error while consuming or acknowledging propagates.
870    pub fn subscribe<H, S>(
871        &self,
872        topic: &str,
873        options: &ConsumeOptions,
874        mut stop: S,
875        mut handler: H,
876    ) -> Result<(), Error>
877    where
878        H: FnMut(&Message) -> Result<(), Box<dyn std::error::Error>>,
879        S: FnMut() -> bool,
880    {
881        let mut options = options.clone();
882        options.wait_time.get_or_insert(20);
883        options.max_messages.get_or_insert(1);
884        while !stop() {
885            for message in self.consume(topic, &options)? {
886                if handler(&message).is_ok() {
887                    match self.ack(topic, &message.id, &message.receipt_token) {
888                        // The message already went back to the queue (deadline expired, `409`) or is already gone (`404`):
889                        // there is nothing to acknowledge, and the loop goes on.
890                        Err(e) if e.status == 409 || e.status == 404 => {}
891                        other => other?,
892                    }
893                }
894                if stop() {
895                    return Ok(());
896                }
897            }
898        }
899        Ok(())
900    }
901}
902
903mod handlers;
904pub use handlers::{
905    ConsumeError, Delivery, HandlerError, DEFAULT_MAX_MESSAGES, DEFAULT_VISIBILITY_TIMEOUT,
906    DEFAULT_WAIT_TIME,
907};
908
909#[cfg(test)]
910mod tests;
taba_client/client/mod.rs

taba_client/client/
mod.rs
