ydb 0.17.1

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

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

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

    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(())
}