Skip to content

Navigation Menu

Sign in
Appearance settings

Search code, repositories, users, issues, pull requests...

Provide feedback

We read every piece of feedback, and take your input very seriously.

Saved searches

Use saved searches to filter your results more quickly
Sign up
Appearance settings

Commit 38fdfc7

Browse files
fix(client): HTTP sender retries send empty request body (#38)
1 parent d282cfc commit 38fdfc7

File tree

6 files changed

+71
-30
lines changed

6 files changed

+71
-30
lines changed

‎buffer.go‎

Lines changed: 11 additions & 15 deletions
Original file line numberDiff line numberDiff line change
@@ -28,7 +28,6 @@ import (
2828
"bytes"
2929
"errors"
3030
"fmt"
31-
"io"
3231
"math"
3332
"math/big"
3433
"strconv"
@@ -122,20 +121,6 @@ func (b *buffer) writeBigInt(i *big.Int) {
122121
b.Write(s)
123122
}
124123

125-
// WriteTo wraps the built-in bytes.Buffer.WriteTo method
126-
// and writes the contents of the buffer to the provided
127-
// io.Writer
128-
func (b *buffer) WriteTo(w io.Writer) (int64, error) {
129-
n, err := b.Buffer.WriteTo(w)
130-
if err != nil {
131-
b.lastMsgPos -= int(n)
132-
return n, err
133-
}
134-
b.lastMsgPos = 0
135-
b.msgCount = 0
136-
return n, nil
137-
}
138-
139124
func (b *buffer) writeTableName(str string) error {
140125
if str == "" {
141126
return fmt.Errorf("table name cannot be empty: %w", errInvalidMsg)
@@ -386,6 +371,17 @@ func (b *buffer) prepareForField() bool {
386371
return true
387372
}
388373

374+
func (b *buffer) Bytes() []byte {
375+
return b.Buffer.Bytes()
376+
}
377+
378+
func (b *buffer) Reset() {
379+
b.Buffer.Reset()
380+
b.lastMsgPos = 0
381+
b.msgCount = 0
382+
b.resetMsgFlags()
383+
}
384+
389385
func (b *buffer) DiscardPendingMsg() {
390386
b.Truncate(b.lastMsgPos)
391387
b.resetMsgFlags()

‎http_sender.go‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@
2525
package questdb
2626

2727
import (
28+
"bytes"
2829
"context"
2930
"crypto/tls"
3031
"encoding/json"
@@ -194,17 +195,20 @@ func (s *httpLineSender) flush0(ctx context.Context, closing bool) error {
194195
if s.buf.msgCount == 0 {
195196
return nil
196197
}
198+
// Always reset the buffer at the end of flush.
199+
defer s.buf.Reset()
197200

198201
// We rely on the following HTTP client implicit behavior:
199202
// s.buf implements WriteTo method which is used by the client.
200203
req, err = http.NewRequest(
201204
http.MethodPost,
202205
s.uri,
203-
&s.buf,
206+
bytes.NewReader(s.buf.Bytes()),
204207
)
205208
if err != nil {
206209
return err
207210
}
211+
req.ContentLength = int64(s.BufLen())
208212

209213
if s.user != "" && s.pass != "" {
210214
req.SetBasicAuth(s.user, s.pass)

‎http_sender_test.go‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -573,6 +573,27 @@ func TestNoFlushWhenSenderIsClosedAndAutoFlushIsDisabled(t *testing.T) {
573573
assert.Empty(t, qdb.Messages(sender))
574574
}
575575

576+
func TestSuccessAfterRetries(t *testing.T) {
577+
ctx := context.Background()
578+
579+
srv, err := newTestHttpServer(failFirstThenSendToBackChannel)
580+
assert.NoError(t, err)
581+
defer srv.Close()
582+
583+
sender, err := qdb.NewLineSender(ctx, qdb.WithHttp(), qdb.WithAddress(srv.Addr()), qdb.WithRetryTimeout(time.Minute))
584+
assert.NoError(t, err)
585+
defer sender.Close(ctx)
586+
587+
err = sender.Table(testTable).Symbol("abc", "def").AtNow(ctx)
588+
assert.NoError(t, err)
589+
590+
err = sender.Flush(ctx)
591+
assert.NoError(t, err)
592+
593+
expectLines(t, srv.BackCh, []string{fmt.Sprintf("%s,abc=def", testTable)})
594+
assert.Zero(t, qdb.BufLen(sender))
595+
}
596+
576597
func TestBufferClearAfterFlush(t *testing.T) {
577598
ctx := context.Background()
578599

‎integration_test.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -132,7 +132,7 @@ func setupQuestDB0(ctx context.Context, auth ilpAuthType, setupProxy bool) (*que
132132
return nil, err
133133
}
134134
req := testcontainers.ContainerRequest{
135-
Image: "questdb/questdb:7.3.10",
135+
Image: "questdb/questdb:7.4.2",
136136
ExposedPorts: []string{"9000/tcp", "9009/tcp"},
137137
WaitingFor: wait.ForHTTP("/").WithPort("9000"),
138138
Networks: []string{networkName},

‎tcp_sender.go‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -205,6 +205,9 @@ func (s *tcpLineSender) Flush(ctx context.Context) error {
205205
s.conn.SetWriteDeadline(time.Time{})
206206
}
207207

208+
// Always reset the buffer at the end of flush.
209+
defer s.buf.Reset()
210+
208211
if _, err := s.buf.WriteTo(s.conn); err != nil {
209212
return err
210213
}

‎utils_test.go‎

Lines changed: 30 additions & 13 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@ import (
3434
"net/http"
3535
"reflect"
3636
"sync"
37+
"sync/atomic"
3738
"testing"
3839
"time"
3940

@@ -43,11 +44,12 @@ import (
4344
type serverType int64
4445

4546
const (
46-
sendToBackChannel serverType = 0
47-
readAndDiscard serverType = 1
48-
returning500 serverType = 2
49-
returning403 serverType = 3
50-
returning404 serverType = 4
47+
sendToBackChannel serverType = 0
48+
readAndDiscard serverType = 1
49+
returning500 serverType = 2
50+
returning403 serverType = 3
51+
returning404 serverType = 4
52+
failFirstThenSendToBackChannel serverType = 5
5153
)
5254

5355
type testServer struct {
@@ -186,21 +188,21 @@ func (s *testServer) serveHttp() {
186188
}
187189
}()
188190

191+
var reqs int64
189192
http.Serve(s.tcpListener, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
190193
var (
191194
err error
192195
)
193196

194197
switch s.serverType {
195-
case sendToBackChannel:
196-
r := bufio.NewReader(r.Body)
197-
var l string
198-
for err == nil {
199-
l, err = r.ReadString('\n')
200-
if err == nil && len(l) > 0 {
201-
lineFeed <- l[0 : len(l)-1]
202-
}
198+
case failFirstThenSendToBackChannel:
199+
if atomic.AddInt64(&reqs, 1) == 1 {
200+
w.WriteHeader(http.StatusInternalServerError)
201+
} else {
202+
err = readAndSendToBackChannel(r, lineFeed)
203203
}
204+
case sendToBackChannel:
205+
err = readAndSendToBackChannel(r, lineFeed)
204206
case readAndDiscard:
205207
_, err = io.Copy(io.Discard, r.Body)
206208
case returning500:
@@ -232,6 +234,21 @@ func (s *testServer) serveHttp() {
232234
}))
233235
}
234236

237+
func readAndSendToBackChannel(r *http.Request, lineFeed chan string) error {
238+
read := bufio.NewReader(r.Body)
239+
var (
240+
l string
241+
err error
242+
)
243+
for err == nil {
244+
l, err = read.ReadString('\n')
245+
if err == nil && len(l) > 0 {
246+
lineFeed <- l[0 : len(l)-1]
247+
}
248+
}
249+
return err
250+
}
251+
235252
func (s *testServer) Close() {
236253
close(s.closeCh)
237254
s.tcpListener.Close()

0 commit comments

Comments
(0)

AltStyle によって変換されたページ (->オリジナル) /