ydb 0.18.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
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136

use std::time::Duration;
use tokio::time::timeout;
use tracing_test::traced_test;

use crate::client_topic::client::CreateTopicOptionsBuilder;
use crate::client_topic::list_types::ConsumerBuilder;
use crate::test_integration_helper::create_client;
use crate::{TopicWriterMessage, Transaction, YdbResult, closure};

#[tokio::test]
#[traced_test]
#[ignore] // need YDB access
async fn topic_writer_tx_write_and_commit() -> YdbResult<()> {
    let client = create_client().await?;
    let database = client.database();
    let topic_path = format!("{database}/writer_tx_commit_test");
    let consumer_name = "writer-tx-consumer";

    let mut topic_client = client.topic_client();
    let _ = topic_client.drop_topic(topic_path.clone()).await;
    topic_client
        .create_topic(
            topic_path.clone(),
            CreateTopicOptionsBuilder::default()
                .consumers(vec![
                    ConsumerBuilder::default()
                        .name(consumer_name.to_string())
                        .build()?,
                ])
                .build()?,
        )
        .await?;

    let messages = [b"first tx writer".to_vec(), b"second tx writer".to_vec()];

    for message_data in &messages {
        client
            .query_client()
            .retry_tx(closure!(
                [&mut topic_client, &topic_path, message_data],
                async |tx: &mut Transaction| {
                    let mut writer = topic_client.create_writer_tx(topic_path, tx).await?;
                    writer
                        .write(
                            TopicWriterMessage::builder()
                                .data(message_data.clone())
                                .build(),
                        )
                        .await?;
                    Ok(true)
                }
            ))
            .await?;
    }

    let mut reader = topic_client
        .create_reader(consumer_name, topic_path.clone())
        .await?;

    let received = timeout(Duration::from_secs(10), async {
        let mut received = Vec::with_capacity(messages.len());
        while received.len() < messages.len() {
            let batch = reader.read_batch().await?;
            for mut message in batch.messages {
                if let Some(body) = message.read_and_take().await? {
                    received.push(body);
                }
            }
        }
        YdbResult::Ok(received)
    })
    .await
    .map_err(|_| crate::YdbError::custom("timed out waiting for committed tx messages"))??;

    assert_eq!(received, messages);

    let _ = topic_client.drop_topic(topic_path).await;
    Ok(())
}

#[tokio::test]
#[traced_test]
#[ignore] // need YDB access
async fn topic_writer_tx_rollback_discards_message() -> YdbResult<()> {
    let client = create_client().await?;
    let database = client.database();
    let topic_path = format!("{database}/writer_tx_rollback_test");
    let consumer_name = "writer-tx-rb-consumer";

    let mut topic_client = client.topic_client();
    let _ = topic_client.drop_topic(topic_path.clone()).await;
    topic_client
        .create_topic(
            topic_path.clone(),
            CreateTopicOptionsBuilder::default()
                .consumers(vec![
                    ConsumerBuilder::default()
                        .name(consumer_name.to_string())
                        .build()?,
                ])
                .build()?,
        )
        .await?;

    client
        .query_client()
        .retry_tx(closure!(
            [&mut topic_client, &topic_path],
            async |tx: &mut Transaction| {
                let mut writer = topic_client.create_writer_tx(topic_path, tx).await?;
                writer
                    .write(
                        TopicWriterMessage::builder()
                            .data(b"should be discarded".to_vec())
                            .build(),
                    )
                    .await?;
                tx.rollback().await?;
                Ok(())
            }
        ))
        .await?;

    let mut reader = topic_client
        .create_reader(consumer_name, topic_path.clone())
        .await?;

    match timeout(Duration::from_millis(500), reader.read_batch()).await {
        Err(_) => {}
        Ok(Ok(_)) => panic!("rolled-back tx message must not be visible to reader"),
        Ok(Err(e)) => return Err(e),
    }

    let _ = topic_client.drop_topic(topic_path).await;
    Ok(())
}