topicwriter

Documentation

Index

• Variables
• type Message
• type Writer
  • func NewWriter(writer *topicwriterinternal.Writer) *Writer
  • func (w *Writer) Close(ctx context.Context) error
  • func (w *Writer) Write(ctx context.Context, messages ...Message) error

Examples

• Writer.Write

Variables

var ErrQueueLimitExceed = topicwriterinternal.PublicErrQueueIsFull

Types

type Message

type Message = topicwriterinternal.Message

type Writer

type Writer struct {
	// contains filtered or unexported fields
}

Writer represent write session to topic It handles connection problems, reconnect to server when need and resend buffered messages

func NewWriter

func NewWriter(writer *topicwriterinternal.Writer) *Writer

func (*Writer) Close

func (w *Writer) Close(ctx context.Context) error

func (*Writer) Write

func (w *Writer) Write(ctx context.Context, messages ...Message) error

Write send messages to topic return after save messages into buffer in async mode (default) and after ack from server in sync mode. see topicoptions.WithSyncWrite

The method will wait first initial connection even for async mode, that mean first write may be slower. especially when connection has problems.

It returns ErrQueueLimitExceed (must be checked by errors.Is) if ctx cancelled before messages put to internal buffer or try to add more messages, that can be put to queue

Experimental

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

Example

ctx := context.Background()
db, err := ydb.Open(ctx, os.Getenv("YDB_CONNECTION_STRING"))
if err != nil {
	log.Fatalf("failed ydb connection: %v", err)
}

producerAndGroupID := "group-id"
writer, err := db.Topic().StartWriter(producerAndGroupID, "topicName",
	topicoptions.WithMessageGroupID(producerAndGroupID),
)
if err != nil {
	log.Fatalf("failed to create topic writer: %v", err)
}

err = writer.Write(ctx,
	topicwriter.Message{Data: strings.NewReader("1")},
	topicwriter.Message{Data: strings.NewReader("2")},
	topicwriter.Message{Data: strings.NewReader("3")},
)
if err == nil {
	fmt.Println("OK")
} else {
	log.Fatalf("failed write to stream")
}