example of transactional writer

Author: rekby

examples/topic/topicwriter/topic_writer_transaction.go

@@ -53,3 +53,26 @@ func CopyMessagesBetweenTopics(ctx context.Context, db *ydb.Driver, reader *topi
 		return nil
 	}, query.WithIdempotent())
 }
+
+func CopyMessagesBetweenTopicsTxWriter(ctx context.Context, db *ydb.Driver, reader *topicreader.Reader, topic string) error {
+	return db.Query().DoTx(ctx, func(ctx context.Context, tx query.TxActor) error {
+		writer, err := db.Topic().StartWransactionalWriter(ctx, tx, topic)
+		if err != nil {
+			return err
+		}
+
+		batch, err := reader.PopMessagesBatchTx(ctx, tx)
+		if err != nil {
+			return err
+		}
+
+		for _, mess := range batch.Messages {
+
+			if err = writer.Write(ctx, topicwriter.Message{Data: mess}); err != nil {
+				return err
+			}
+		}
+
+		return nil
+	}, query.WithIdempotent())
+}

topic/client.go

@@ -3,6 +3,7 @@ package topic
 import (
 	"context"
 
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/tx"
 	"github.com/ydb-platform/ydb-go-sdk/v3/topic/topiclistener"
 	"github.com/ydb-platform/ydb-go-sdk/v3/topic/topicoptions"
 	"github.com/ydb-platform/ydb-go-sdk/v3/topic/topicreader"
@@ -47,4 +48,7 @@ type Client interface {
 	// StartWriter start write session to topic
 	// it is fast non block call, connection starts in background
 	StartWriter(topicPath string, opts ...topicoptions.WriterOption) (*topicwriter.Writer, error)
+
+	// StartTransactionalWriter
+	StartWransactionalWriter(ctx context.Context, tx tx.Identifier, topicpath string, opts ...topicoptions.WriterOption) (*topicwriter.TxWriter, error)
 }

topic/topicwriter/topicwriter.go

@@ -92,3 +92,10 @@ func (w *Writer) Close(ctx context.Context) error {
 func (w *Writer) Flush(ctx context.Context) error {
 	return w.inner.Flush(ctx)
 }
+
+type TxWriter struct{}
+
+func (w *TxWriter) Write(ctx context.Context, messages ...Message) error {
+	//TODO implement me
+	panic("implement me")
+}