init write impl

Author: rekby

examples/topic/topicwriter/topic_writer_transaction.go

@@ -3,17 +3,17 @@ package topicwriter
 import (
 	"context"
 	"fmt"
-	"io"
 	"strings"
 
 	"github.com/ydb-platform/ydb-go-sdk/v3"
 	"github.com/ydb-platform/ydb-go-sdk/v3/query"
+	"github.com/ydb-platform/ydb-go-sdk/v3/topic/topicreader"
 	"github.com/ydb-platform/ydb-go-sdk/v3/topic/topicwriter"
 )
 
-func SendMessageWithinTransaction(ctx context.Context, db *ydb.Driver, writer *topicwriter.Writer, id int64) error {
+func TableAndTopicWithinTransaction(ctx context.Context, db *ydb.Driver, writer *topicwriter.Writer, id int64) error {
 	return db.Query().DoTx(ctx, func(ctx context.Context, t query.TxActor) error {
-		row, err := t.ReadRow(ctx, "SELECT val FROM table WHERE id=$id", query.WithParameters(
+		row, err := t.QueryRow(ctx, "SELECT val FROM table WHERE id=$id", query.WithParameters(
 			ydb.ParamsBuilder().
 				Param("$id").Int64(id).
 				Build()))
@@ -36,68 +36,18 @@ func SendMessageWithinTransaction(ctx context.Context, db *ydb.Driver, writer *t
 	})
 }
 
-func SendWithRecreateWriter(
-	ctx context.Context,
-	db *ydb.Driver,
-	writerFabric func(ctx context.Context, db *ydb.Driver) (*topicwriter.Writer, error),
-	ids <-chan int64,
-) error {
-	// second loop
-	var lastData int64
-	processed := true
-	return db.Query().Do(ctx, func(ctx context.Context, s query.Session) error {
-		writer, err := writerFabric(ctx, db)
+func CopyMessagesBetweenTopics(ctx context.Context, db *ydb.Driver, reader *topicreader.Reader, writer *topicwriter.Writer) error {
+	return db.Query().DoTx(ctx, func(ctx context.Context, tx query.TxActor) error {
+		batch, err := reader.PopMessagesBatchTx(ctx, tx)
 		if err != nil {
 			return err
 		}
-		defer func() { _ = writer.Close(ctx) }()
 
-		for {
-			var id int64
-			if processed {
-				select {
-				case <-ctx.Done():
-					return ctx.Err()
-				case val, ok := <-ids:
-					if !ok {
-						return io.EOF
-					}
-					lastData = val
-					processed = false
-				}
-			}
-
-			id = lastData
-			t, err := s.Begin(ctx, nil)
-			if err != nil {
-				return err
-			}
-
-			row, err := t.ReadRow(ctx, "SELECT val FROM table WHERE id=$id", query.WithParameters(
-				ydb.ParamsBuilder().
-					Param("$id").Int64(id).
-					Build()))
-			if err != nil {
-				return err
-			}
-
-			var val int64
-			if err = row.Scan(&val); err != nil {
-				return err
-			}
-
-			err = writer.WriteWithTx(ctx, t, topicwriter.Message{
-				Data: strings.NewReader(fmt.Sprintf("val: %v processed", val)),
-			})
-			if err != nil {
-				return err
-			}
-
-			err = t.CommitTx(ctx)
-			if err != nil {
-				return err
-			}
-			processed = true
+		sendMessages := make([]topicwriter.Message, len(batch.Messages))
+		for i, mess := range batch.Messages {
+			sendMessages[i] = topicwriter.Message{Data: mess}
 		}
-	})
+
+		return writer.WriteWithTx(ctx, tx, sendMessages...)
+	}, query.WithIdempotent())
 }

internal/topic/topicwriterinternal/writer.go

@@ -7,6 +7,7 @@ import (
 
 	"github.com/ydb-platform/ydb-go-sdk/v3/credentials"
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/grpcwrapper/rawtopic/rawtopicwriter"
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/tx"
 )
 
 //go:generate mockgen -destination raw_topic_writer_stream_mock_test.go --typed -package topicwriterinternal -write_package_comment=false github.com/ydb-platform/ydb-go-sdk/v3/internal/topic/topicwriterinternal RawTopicWriterStream
@@ -48,6 +49,11 @@ func (w *Writer) Write(ctx context.Context, messages ...PublicMessage) error {
 	return w.streamWriter.Write(ctx, messages)
 }
 
+func (w *Writer) WriteWithTx(ctx context.Context, tx tx.Transaction, messages ...PublicMessage) error {
+	//TODO implement me
+	panic("implement me")
+}
+
 func (w *Writer) WaitInit(ctx context.Context) (info InitialInfo, err error) {
 	return w.streamWriter.WaitInit(ctx)
 }

topic/topicwriter/topicwriter.go

@@ -2,7 +2,10 @@ package topicwriter
 
 import (
 	"context"
+	"errors"
+
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/tx"
+	"github.com/ydb-platform/ydb-go-sdk/v3/internal/xerrors"
 
 	"github.com/ydb-platform/ydb-go-sdk/v3/internal/topic/topicwriterinternal"
 )
@@ -11,7 +14,10 @@ type (
 	Message = topicwriterinternal.PublicMessage
 )
 
-var ErrQueueLimitExceed = topicwriterinternal.PublicErrQueueIsFull
+var (
+	ErrQueueLimitExceed = topicwriterinternal.PublicErrQueueIsFull
+	errBadTxId          = xerrors.Wrap(errors.New("ydb: bad transaction type, supported transactions from query service only"))
+)
 
 // Writer represent write session to topic
 // It handles connection problems, reconnect to server when need and resend buffered messages
@@ -45,11 +51,12 @@ func (w *Writer) Write(ctx context.Context, messages ...Message) error {
 }
 
 func (w *Writer) WriteWithTx(ctx context.Context, transaction tx.Identifier, messages ...Message) error {
-	panic("not impl")
-}
+	internalTx, ok := transaction.(tx.Transaction)
+	if !ok {
+		return xerrors.WithStackTrace(errBadTxId)
+	}
 
-func (w *Writer) WriteOpts(ctx context.Context, messages []Message, opts ...WriteOption) error {
-	panic("not impl")
+	return w.inner.WriteWithTx(ctx, internalTx, messages...)
 }
 
 // WaitInit waits until the reader is initialized