topic

README

ydb-go-sdk - pure Go native and database/sql driver for YDB

topic - pure Go native client for YDB Topic

Supports topic clients for YDB, see ydb-go-sdk for describe driver all features YDB is an open-source Distributed SQL Database that combines high availability and scalability with strict consistency and ACID transactions with CDC and Data stream features.

Installation

go get -u github.com/ydb-platform/ydb-go-sdk/v3

Example Usage

• connect to YDB

db, err := ydb.Open(ctx, "grpcs://localhost:2135/local")
if err != nil {
    log.Fatal(err)
}

• send messages

producerAndGroupID := "group-id"
writer, err := db.Topic().StartWriter(producerAndGroupID, "topicName",
    topicoptions.WithMessageGroupID(producerAndGroupID),
    topicoptions.WithCodec(topictypes.CodecGzip),
)
if err != nil {
    log.Fatal(err)
}

data1 := []byte{1, 2, 3}
data2 := []byte{4, 5, 6}
mess1 := topicwriter.Message{Data: bytes.NewReader(data1)}
mess2 := topicwriter.Message{Data: bytes.NewReader(data2)}

err = w.Write(ctx, mess1, mess2)
if err != nil {
	log.Fatal(err)
}

• read messages

for {
    msg, err := reader.ReadMessage(ctx)
    if err != nil {
        log.Fatal(err)
    }
    content, err := io.ReadAll(msg)
    if err != nil {
        log.Fatal(err)
    }
    fmt.Println(string(content))
    err = reader.Commit(msg.Context(), msg)
    if err != nil {
        log.Fatal(err)
    }
}
 

Documentation

Overview

Example (Topic_read_message)

package main

import (
	"context"
	"fmt"
	"io/ioutil"

	ydb "github.com/ydb-platform/ydb-go-sdk/v3"
	"github.com/ydb-platform/ydb-go-sdk/v3/topic/topicoptions"
)

func main() {
	ctx := context.TODO()
	db, err := ydb.Open(ctx, "grpcs://localhost:2135/local")
	if err != nil {
		fmt.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 := ioutil.ReadAll(mess)
		if err != nil {
			fmt.Printf("failed start reader: %v", err)
			return
		}
		fmt.Println(string(content))
	}
}

Index

• type Client

Examples

• Package (Topic_read_message)

Types

type Client

type Client interface {
	// Alter change topic options
	//
	// Experimental
	//
	// Notice: This API is EXPERIMENTAL and may be changed or removed in a
	// later release.
	Alter(ctx context.Context, path string, opts ...topicoptions.AlterOption) error

	// Create create topic
	//
	// Experimental
	//
	// Notice: This API is EXPERIMENTAL and may be changed or removed in a
	// later release.
	Create(ctx context.Context, path string, opts ...topicoptions.CreateOption) error

	// Describe describe topic
	//
	// Experimental
	//
	// Notice: This API is EXPERIMENTAL and may be changed or removed in a
	// later release.
	Describe(ctx context.Context, path string, opts ...topicoptions.DescribeOption) (topictypes.TopicDescription, error)

	// Drop drop topic
	//
	// Experimental
	//
	// Notice: This API is EXPERIMENTAL and may be changed or removed in a
	// later release.
	Drop(ctx context.Context, path string, opts ...topicoptions.DropOption) error

	// StartReader start read messages from topic
	// it is fast non block call, connection starts in background
	//
	// Experimental
	//
	// Notice: This API is EXPERIMENTAL and may be changed or removed in a
	// later release.
	StartReader(
		consumer string,
		readSelectors topicoptions.ReadSelectors,
		opts ...topicoptions.ReaderOption,
	) (*topicreader.Reader, error)

	// StartWriter start write session to topic
	// it is fast non block call, connection starts in background
	//
	// Experimental
	//
	// Notice: This API is EXPERIMENTAL and may be changed or removed in a
	// later release.
	StartWriter(producerID, path string, opts ...topicoptions.WriterOption) (*topicwriter.Writer, error)
}

Client is interface for topic client

Experimental

Notice: This API is EXPERIMENTAL and may be changed or removed in a later release.