Wait for last writtent message instead

Author: andrey-zvez

internal/topic/topicwriterinternal/queue.go

@@ -309,33 +309,13 @@ func (q *messageQueue) Wait(ctx context.Context, waiter MessageQueueAckWaiter) e
 	}
 }
 
-func (q *messageQueue) WaitEmpty(ctx context.Context) error {
-	if err := ctx.Err(); err != nil {
-		return err
-	}
-
-	ctxDone := ctx.Done()
-	for {
-		ackReceived := q.acksReceivedEvent.Waiter()
-
-		keepWaiting := false
-		q.m.WithRLock(func() {
-			keepWaiting = len(q.messagesByOrder) > 0
-		})
-
-		if !keepWaiting {
-			return nil
-		}
-
-		select {
-		case <-ctxDone:
-			return ctx.Err()
-		case <-q.closedChan:
-			return q.closedErr
-		case <-ackReceived.Done():
-			// pass next iteration
-		}
-	}
+// WaitLastWritten waits for last written message gets ack.
+func (q *messageQueue) WaitLastWritten(ctx context.Context) error {
+	var lastIndex int
+	q.m.WithRLock(func() {
+		lastIndex = q.lastWrittenIndex
+	})
+	return q.Wait(ctx, MessageQueueAckWaiter{sequenseNumbers: []int{lastIndex}})
 }
 
 type MessageQueueAckWaiter struct {

internal/topic/topicwriterinternal/writer_reconnector.go

@@ -325,7 +325,7 @@ func (w *WriterReconnector) createMessagesWithContent(messages []Message) ([]mes
 }
 
 func (w *WriterReconnector) Flush(ctx context.Context) error {
-	return w.queue.WaitEmpty(ctx)
+	return w.queue.WaitLastWritten(ctx)
 }
 
 func (w *WriterReconnector) Close(ctx context.Context) error {