internal message to public message - because the message is part of p…

…ublic interface
ydb-platform · Dec 14, 2023 · 9e58d03 · 9e58d03
1 parent f1d4f44
commit 9e58d03
Show file tree

Hide file tree

Showing 8 changed files with 28 additions and 28 deletions.
diff --git a/internal/topic/topicwriterinternal/encoders_test.go b/internal/topic/topicwriterinternal/encoders_test.go
@@ -51,13 +51,13 @@ func TestEncoderSelector_CodecMeasure(t *testing.T) {
 			var messages []messageWithDataContent
 			for i := 0; i < smallCount; i++ {
 				data := make([]byte, smallSize)
-				message := newMessageDataWithContent(Message{Data: bytes.NewReader(data)}, testCommonEncoders)
+				message := newMessageDataWithContent(PublicMessage{Data: bytes.NewReader(data)}, testCommonEncoders)
 				messages = append(messages, message)
 			}
 
 			for i := 0; i < largeCount; i++ {
 				data := make([]byte, largeSize)
-				message := newMessageDataWithContent(Message{Data: bytes.NewReader(data)}, testCommonEncoders)
+				message := newMessageDataWithContent(PublicMessage{Data: bytes.NewReader(data)}, testCommonEncoders)
 				messages = append(messages, message)
 			}
 
@@ -160,7 +160,7 @@ func TestCompressMessages(t *testing.T) {
 		require.NoError(t, readInParallelWithCodec(messages, rawtopiccommon.CodecRaw, 1))
 	})
 	t.Run("RawError", func(t *testing.T) {
-		mess := newMessageDataWithContent(Message{}, testCommonEncoders)
+		mess := newMessageDataWithContent(PublicMessage{}, testCommonEncoders)
 		_, err := mess.GetEncodedBytes(rawtopiccommon.CodecGzip)
 		require.NoError(t, err)
 		messages := []messageWithDataContent{mess}
@@ -171,7 +171,7 @@ func TestCompressMessages(t *testing.T) {
 	t.Run("GzipOneThread", func(t *testing.T) {
 		var messages []messageWithDataContent
 		for i := 0; i < messageCount; i++ {
-			mess := newMessageDataWithContent(Message{Data: strings.NewReader("asdf")}, testCommonEncoders)
+			mess := newMessageDataWithContent(PublicMessage{Data: strings.NewReader("asdf")}, testCommonEncoders)
 			messages = append(messages, mess)
 		}
 
@@ -185,7 +185,7 @@ func TestCompressMessages(t *testing.T) {
 	t.Run("GzipOk", func(t *testing.T) {
 		var messages []messageWithDataContent
 		for i := 0; i < messageCount; i++ {
-			mess := newMessageDataWithContent(Message{Data: strings.NewReader("asdf")}, testCommonEncoders)
+			mess := newMessageDataWithContent(PublicMessage{Data: strings.NewReader("asdf")}, testCommonEncoders)
 			messages = append(messages, mess)
 		}
 
@@ -198,7 +198,7 @@ func TestCompressMessages(t *testing.T) {
 	t.Run("GzipErr", func(t *testing.T) {
 		var messages []messageWithDataContent
 		for i := 0; i < messageCount; i++ {
-			mess := newMessageDataWithContent(Message{Data: strings.NewReader("asdf")}, testCommonEncoders)
+			mess := newMessageDataWithContent(PublicMessage{Data: strings.NewReader("asdf")}, testCommonEncoders)
 			messages = append(messages, mess)
 		}
 		messages[0].dataWasRead = true

diff --git a/internal/topic/topicwriterinternal/message.go b/internal/topic/topicwriterinternal/message.go
@@ -14,7 +14,7 @@ import (
 
 var errNoRawContent = xerrors.Wrap(errors.New("ydb: internal state error - no raw message content"))
 
-type Message struct {
+type PublicMessage struct {
 	SeqNo     int64
 	CreatedAt time.Time
 	Data      io.Reader
@@ -53,7 +53,7 @@ func NewPartitioningWithPartitionID(id int64) PublicFuturePartitioning {
 }
 
 type messageWithDataContent struct {
-	Message
+	PublicMessage
 
 	dataWasRead         bool
 	encoders            *EncoderMap
@@ -182,11 +182,11 @@ func (m *messageWithDataContent) getEncodedBytes(codec rawtopiccommon.Codec) ([]
 }
 
 func newMessageDataWithContent(
-	message Message, //nolint:gocritic
+	message PublicMessage, //nolint:gocritic
 	encoders *EncoderMap,
 ) messageWithDataContent {
 	return messageWithDataContent{
-		Message:  message,
-		encoders: encoders,
+		PublicMessage: message,
+		encoders:      encoders,
 	}
 }
diff --git a/internal/topic/topicwriterinternal/writer.go b/internal/topic/topicwriterinternal/writer.go
@@ -40,7 +40,7 @@ func NewWriter(cred credentials.Credentials, options []PublicWriterOption) (*Wri
 	}, nil
 }
 
-func (w *Writer) Write(ctx context.Context, messages ...Message) error {
+func (w *Writer) Write(ctx context.Context, messages ...PublicMessage) error {
 	if err := ctx.Err(); err != nil {
 		return err
 	}

diff --git a/internal/topic/topicwriterinternal/writer_reconnector.go b/internal/topic/topicwriterinternal/writer_reconnector.go
@@ -207,7 +207,7 @@ func (w *WriterReconnector) start() {
 	w.background.Start(name+", sendloop", w.connectionLoop)
 }
 
-func (w *WriterReconnector) Write(ctx context.Context, messages []Message) error {
+func (w *WriterReconnector) Write(ctx context.Context, messages []PublicMessage) error {
 	if err := w.background.CloseReason(); err != nil {
 		return xerrors.WithStackTrace(fmt.Errorf("ydb: writer is closed: %w", err))
 	}
@@ -291,7 +291,7 @@ func (w *WriterReconnector) checkMessages(messages []messageWithDataContent) err
 	return nil
 }
 
-func (w *WriterReconnector) createMessagesWithContent(messages []Message) ([]messageWithDataContent, error) {
+func (w *WriterReconnector) createMessagesWithContent(messages []PublicMessage) ([]messageWithDataContent, error) {
 	res := make([]messageWithDataContent, 0, len(messages))
 	for i := range messages {
 		mess := newMessageDataWithContent(messages[i], w.encodersMap)

diff --git a/internal/topic/topicwriterinternal/writer_reconnector_test.go b/internal/topic/topicwriterinternal/writer_reconnector_test.go
@@ -77,10 +77,10 @@ func TestWriterImpl_CheckMessages(t *testing.T) {
 		maxSize := 5
 		w.cfg.MaxMessageSize = maxSize
 
-		err := w.Write(ctx, []Message{{Data: bytes.NewReader(make([]byte, maxSize))}})
+		err := w.Write(ctx, []PublicMessage{{Data: bytes.NewReader(make([]byte, maxSize))}})
 		require.NoError(t, err)
 
-		err = w.Write(ctx, []Message{{Data: bytes.NewReader(make([]byte, maxSize+1))}})
+		err = w.Write(ctx, []PublicMessage{{Data: bytes.NewReader(make([]byte, maxSize+1))}})
 		require.Error(t, err)
 	})
 }
@@ -141,7 +141,7 @@ func TestWriterImpl_Write(t *testing.T) {
 
 			writeCompleted := make(empty.Chan)
 			go func() {
-				err := e.writer.Write(e.ctx, []Message{{SeqNo: seqNo, CreatedAt: messageTime, Data: bytes.NewReader(messageData)}})
+				err := e.writer.Write(e.ctx, []PublicMessage{{SeqNo: seqNo, CreatedAt: messageTime, Data: bytes.NewReader(messageData)}})
 				require.NoError(t, err)
 				close(writeCompleted)
 			}()
@@ -187,7 +187,7 @@ func TestWriterImpl_WriteCodecs(t *testing.T) {
 		})
 
 		require.NoError(t, err)
-		require.NoError(t, e.writer.Write(e.ctx, []Message{{
+		require.NoError(t, e.writer.Write(e.ctx, []PublicMessage{{
 			Data: bytes.NewReader(messContent),
 		}}))
 
@@ -215,7 +215,7 @@ func TestWriterImpl_WriteCodecs(t *testing.T) {
 		})
 
 		require.NoError(t, err)
-		require.NoError(t, e.writer.Write(e.ctx, []Message{{
+		require.NoError(t, e.writer.Write(e.ctx, []PublicMessage{{
 			Data: bytes.NewReader(messContent),
 		}}))
 
@@ -242,7 +242,7 @@ func TestWriterImpl_WriteCodecs(t *testing.T) {
 		codecs := make(map[rawtopiccommon.Codec]empty.Struct)
 
 		for i := 0; i < codecMeasureIntervalBatches; i++ {
-			require.NoError(t, e.writer.Write(e.ctx, []Message{{
+			require.NoError(t, e.writer.Write(e.ctx, []PublicMessage{{
 				Data: bytes.NewReader(messContentShort),
 			}}))
 			// wait send
@@ -251,7 +251,7 @@ func TestWriterImpl_WriteCodecs(t *testing.T) {
 		}
 
 		for i := 0; i < codecMeasureIntervalBatches; i++ {
-			require.NoError(t, e.writer.Write(e.ctx, []Message{{
+			require.NoError(t, e.writer.Write(e.ctx, []PublicMessage{{
 				Data: bytes.NewReader(messContentLong),
 			}}))
 			// wait send
@@ -711,12 +711,12 @@ func TestCalculateAllowedCodecs(t *testing.T) {
 }
 
 func newTestMessageWithDataContent(num int) messageWithDataContent {
-	res := newMessageDataWithContent(Message{SeqNo: int64(num)}, testCommonEncoders)
+	res := newMessageDataWithContent(PublicMessage{SeqNo: int64(num)}, testCommonEncoders)
 	return res
 }
 
-func newTestMessages(numbers ...int) []Message {
-	messages := make([]Message, len(numbers))
+func newTestMessages(numbers ...int) []PublicMessage {
+	messages := make([]PublicMessage, len(numbers))
 	for i, num := range numbers {
 		messages[i].SeqNo = int64(num)
 	}

diff --git a/internal/topic/topicwriterinternal/writer_stream_interface.go b/internal/topic/topicwriterinternal/writer_stream_interface.go
@@ -6,7 +6,7 @@ import (
 
 //go:generate mockgen -source writer_stream_interface.go -destination writer_stream_interface_mock_test.go -package topicwriterinternal -write_package_comment=false
 type StreamWriter interface {
-	Write(ctx context.Context, messages []Message) error
+	Write(ctx context.Context, messages []PublicMessage) error
 	WaitInit(ctx context.Context) (info InitialInfo, err error)
 	Close(ctx context.Context) error
 }

diff --git a/internal/topic/topicwriterinternal/writer_stream_interface_mock_test.go b/internal/topic/topicwriterinternal/writer_stream_interface_mock_test.go
diff --git a/internal/topic/topicwriterinternal/writer_test.go b/internal/topic/topicwriterinternal/writer_test.go
@@ -41,7 +41,7 @@ func TestWriterWrite(t *testing.T) {
 			w := Writer{
 				streamWriter: strm,
 			}
-			require.NoError(t, w.Write(ctx, Message{SeqNo: 1}))
+			require.NoError(t, w.Write(ctx, PublicMessage{SeqNo: 1}))
 		})
 	})
 }
@@ -56,6 +56,6 @@ func TestWriterWriteMessage(t *testing.T) {
 		w := Writer{
 			streamWriter: strm,
 		}
-		require.NoError(t, w.Write(ctx, Message{SeqNo: 1}, Message{SeqNo: 3}))
+		require.NoError(t, w.Write(ctx, PublicMessage{SeqNo: 1}, PublicMessage{SeqNo: 3}))
 	})
 }