/
githubmirror
/
ydb-go-sdk
Обзор
Документация
Войти
/
githubmirror
/
ydb-go-sdk
Код
Запросы
0
Пакеты
0
Релизы
0
Аналитика
Безопасность
master
topic/example_test.go
246 строк
6 KB
Konstantin Prokopenko
Topic consumer availability period (#1943)
25 ноя 2025, 13:43
Не верифицирован
25 ноя 2025, 13:43
995c4d2
Код
Авторство
О чём код?
package topic_test import ( "context" "fmt" "io" "log" "os" "time" ydb "github.com/ydb-platform/ydb-go-sdk/v3" "github.com/ydb-platform/ydb-go-sdk/v3/topic/topicoptions" "github.com/ydb-platform/ydb-go-sdk/v3/topic/topictypes" ) func Example_createTopic() { ctx := context.TODO() connectionString := os.Getenv("YDB_CONNECTION_STRING") if connectionString == "" { connectionString = "grpc://localhost:2136/local" } db, err := ydb.Open(ctx, connectionString) if err != nil { log.Printf("failed connect: %v", err) return } defer db.Close(ctx) // cleanup resources err = db.Topic().Create(ctx, "topic-path", // optional topicoptions.CreateWithSupportedCodecs(topictypes.CodecRaw, topictypes.CodecGzip), // optional topicoptions.CreateWithMinActivePartitions(3), ) if err != nil { log.Printf("failed create topic: %v", err) return } } func Example_alterTopic() { ctx := context.TODO() connectionString := os.Getenv("YDB_CONNECTION_STRING") if connectionString == "" { connectionString = "grpc://localhost:2136/local" } db, err := ydb.Open(ctx, connectionString) if err != nil { log.Printf("failed connect: %v", err) return } defer db.Close(ctx) // cleanup resources err = db.Topic().Alter(ctx, "topic-path", topicoptions.AlterWithAddConsumers(topictypes.Consumer{ Name: "new-consumer", SupportedCodecs: []topictypes.Codec{topictypes.CodecRaw, topictypes.CodecGzip}, // optional }), ) if err != nil { log.Printf("failed alter topic: %v", err) return } } func Example_createTopicWithConsumerAvailabilityPeriod() { ctx := context.TODO() connectionString := os.Getenv("YDB_CONNECTION_STRING") if connectionString == "" { connectionString = "grpc://localhost:2136/local" } db, err := ydb.Open(ctx, connectionString) if err != nil { log.Printf("failed connect: %v", err) return } defer db.Close(ctx) // cleanup resources // Create topic with consumer that has 24-hour availability period // Messages for this consumer won't expire for at least 24 hours even if not committed availabilityPeriod := 24 * time.Hour err = db.Topic().Create(ctx, "topic-path", topicoptions.CreateWithConsumer(topictypes.Consumer{ Name: "my-consumer", Important: true, AvailabilityPeriod: &availabilityPeriod, // Messages available for at least 24 hours SupportedCodecs: []topictypes.Codec{topictypes.CodecRaw, topictypes.CodecGzip}, }), ) if err != nil { log.Printf("failed create topic: %v", err) return } } func Example_alterConsumerAvailabilityPeriod() { ctx := context.TODO() connectionString := os.Getenv("YDB_CONNECTION_STRING") if connectionString == "" { connectionString = "grpc://localhost:2136/local" } db, err := ydb.Open(ctx, connectionString) if err != nil { log.Printf("failed connect: %v", err) return } defer db.Close(ctx) // cleanup resources // Set availability period to 48 hours for existing consumer err = db.Topic().Alter(ctx, "topic-path", topicoptions.AlterConsumerWithAvailabilityPeriod("my-consumer", 48*time.Hour), ) if err != nil { log.Printf("failed alter consumer availability period: %v", err) return } // Reset availability period to default value err = db.Topic().Alter(ctx, "topic-path", topicoptions.AlterConsumerResetAvailabilityPeriod("my-consumer"), ) if err != nil { log.Printf("failed reset consumer availability period: %v", err) return } } func Example_describeTopic() { ctx := context.TODO() connectionString := os.Getenv("YDB_CONNECTION_STRING") if connectionString == "" { connectionString = "grpc://localhost:2136/local" } db, err := ydb.Open(ctx, connectionString) if err != nil { log.Printf("failed connect: %v", err) return } defer db.Close(ctx) // cleanup resources descResult, err := db.Topic().Describe(ctx, "topic-path") if err != nil { log.Printf("failed describe topic: %v", err) return } fmt.Printf("describe: %#v\n", descResult) } func Example_desrcibeTopicConsumer() { ctx := context.TODO() connectionString := os.Getenv("YDB_CONNECTION_STRING") if connectionString == "" { connectionString = "grpc://localhost:2136/local" } db, err := ydb.Open( ctx, connectionString, ) if err != nil { log.Printf("failed connect: %v", err) return } defer db.Close(ctx) // cleanup resources descResult, err := db.Topic().DescribeTopicConsumer(ctx, "topic-path", "new-consumer") if err != nil { log.Printf("failed describe topic consumer: %v", err) return } fmt.Printf("describe consumer: %#v\n", descResult) } func Example_dropTopic() { ctx := context.TODO() connectionString := os.Getenv("YDB_CONNECTION_STRING") if connectionString == "" { connectionString = "grpc://localhost:2136/local" } db, err := ydb.Open(ctx, connectionString) if err != nil { log.Printf("failed connect: %v", err) return } defer db.Close(ctx) // cleanup resources err = db.Topic().Drop(ctx, "topic-path") if err != nil { log.Printf("failed drop topic: %v", err) return } } func Example_readMessage() { ctx := context.TODO() connectionString := os.Getenv("YDB_CONNECTION_STRING") if connectionString == "" { connectionString = "grpc://localhost:2136/local" } db, err := ydb.Open(ctx, connectionString) if err != nil { log.Printf("failed connect: %v", err) return } defer db.Close(ctx) // cleanup resources reader, err := db.Topic().StartReader("consumer", topicoptions.ReadTopic("/topic/path")) if err != nil { fmt.Printf("failed start reader: %v", err) return } for { mess, err := reader.ReadMessage(ctx) if err != nil { fmt.Printf("failed start reader: %v", err) return } content, err := io.ReadAll(mess) if err != nil { fmt.Printf("failed start reader: %v", err) return } fmt.Println(string(content)) } }