ydb 0.16.0

Crate contains generated low-level grpc code from YDB API protobuf, used as base for ydb crate
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77

use std::convert::Infallible;
use std::time::Duration;

use secrecy::ExposeSecret;
use tokio::select;
use tokio::sync::mpsc;
use tokio::time;
use tokio_util::sync::CancellationToken;
use tracing::debug;

use crate::client_common::TokenCache;
use crate::grpc_wrapper::raw_topic_service::common::update_token::RawUpdateTokenRequest;
use crate::grpc_wrapper::raw_topic_service::stream_read::messages::RawFromClientOneOf;
use crate::{YdbError, YdbResult};

use super::reconnector;

const UPDATE_TOKEN_INTERVAL: Duration = Duration::from_secs(3600);

pub(super) struct AuthTokenSender {
    token_cache: TokenCache,
    outgoing_tx: mpsc::UnboundedSender<RawFromClientOneOf>,
    cancellation: CancellationToken,
}

impl AuthTokenSender {
    pub(super) fn new(
        attempt: &reconnector::ConnectionAttempt,
        outgoing_tx: mpsc::UnboundedSender<RawFromClientOneOf>,
    ) -> Self {
        Self {
            token_cache: attempt.token_cache.clone(),
            outgoing_tx,
            cancellation: attempt.cancellation_token.clone(),
        }
    }

    pub(super) async fn run(self) -> YdbResult<()> {
        let Self {
            token_cache,
            outgoing_tx,
            cancellation,
        } = self;

        select! {
            _ = cancellation.cancelled() => {
                debug!("topic reader tokenizer cancelled, stopping");
                Ok(())
            }
            result = update_tokens(token_cache, outgoing_tx) => {
                let Err(e) = result;
                Err(e)
            }
        }
    }
}

async fn update_tokens(
    token_cache: TokenCache,
    outgoing_tx: mpsc::UnboundedSender<RawFromClientOneOf>,
) -> YdbResult<Infallible> {
    let mut interval = time::interval(UPDATE_TOKEN_INTERVAL);
    interval.tick().await;

    loop {
        interval.tick().await;
        let token = token_cache.token();
        let request = RawFromClientOneOf::UpdateTokenRequest(RawUpdateTokenRequest {
            token: token.expose_secret().to_string(),
        });
        outgoing_tx.send(request).map_err(|e| {
            YdbError::Transport(format!(
                "topic reader tokenizer outgoing channel closed: {e}"
            ))
        })?;
    }
}