ydb 0.17.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

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, YdbResult};

#[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 message_data = b"hello from tx writer".to_vec();

    let topic_path_clone = topic_path.clone();
    let message_clone = message_data.clone();
    let topic_client_clone = topic_client.clone();

    client
        .query_client()
        .retry_tx(async |tx| {
            let mut tc = topic_client_clone.clone();
            let mut writer = tc.create_writer_tx(topic_path_clone.clone(), tx).await?;
            writer
                .write(
                    TopicWriterMessage::builder()
                        .data(message_clone.clone())
                        .build(),
                )
                .await?;
            writer.stop().await?;
            Ok(true)
        })
        .await?;

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

    let batch = timeout(Duration::from_secs(10), reader.read_batch())
        .await
        .map_err(|_| crate::YdbError::custom("timed out waiting for committed tx message"))??;

    let mut msg = batch.messages.into_iter().next().unwrap();
    let body = msg.read_and_take().await?.unwrap();
    assert_eq!(body, message_data);

    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?;

    let topic_path_clone = topic_path.clone();
    let topic_client_clone = topic_client.clone();

    client
        .query_client()
        .retry_tx(async |tx| {
            let mut tc = topic_client_clone.clone();
            let mut writer = tc.create_writer_tx(topic_path_clone.clone(), tx).await?;
            writer
                .write(
                    TopicWriterMessage::builder()
                        .data(b"should be discarded".to_vec())
                        .build(),
                )
                .await?;
            writer.stop().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(())
}