Merge branch 'main' into steve/add-ping

Author: sklarsa

README.md

@@ -9,7 +9,7 @@ Golang client for QuestDB's [Influx Line Protocol](https://questdb.io/docs/refer
 The library requires Go 1.19 or newer.
 
 Features:
-* Context-aware API.
+* [Context](https://www.digitalocean.com/community/tutorials/how-to-use-contexts-in-go)-aware API.
 * Optimized for batch writes.
 * Supports TLS encryption and ILP authentication.
 * Automatic write retries and connection reuse for ILP over HTTP.
@@ -42,24 +42,41 @@ func main() {
 		log.Fatal(err)
 	}
 	// Make sure to close the sender on exit to release resources.
-	defer sender.Close()
+	defer sender.Close(ctx)
+
 	// Send a few ILP messages.
+	tradedTs, err := time.Parse(time.RFC3339, "2022-08-06T15:04:05.123456Z")
+	if err != nil {
+		log.Fatal(err)
+	}
 	err = sender.
-		Table("trades").
-		Symbol("name", "test_ilp1").
-		Float64Column("value", 12.4).
-		AtNow(ctx)
+		Table("trades_go").
+		Symbol("pair", "USDGBP").
+		Symbol("type", "buy").
+		Float64Column("traded_price", 0.83).
+		Float64Column("limit_price", 0.84).
+		Int64Column("qty", 100).
+		At(ctx, tradedTs)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	tradedTs, err = time.Parse(time.RFC3339, "2022-08-06T15:04:06.987654Z")
 	if err != nil {
 		log.Fatal(err)
 	}
 	err = sender.
-		Table("trades").
-		Symbol("name", "test_ilp2").
-		Float64Column("value", 11.4).
-		At(ctx, time.Now().UnixNano())
+		Table("trades_go").
+		Symbol("pair", "GBPJPY").
+		Symbol("type", "sell").
+		Float64Column("traded_price", 135.97).
+		Float64Column("limit_price", 0.84).
+		Int64Column("qty", 400).
+		At(ctx, tradedTs)
 	if err != nil {
 		log.Fatal(err)
 	}
+
 	// Make sure that the messages are sent over the network.
 	err = sender.Flush(ctx)
 	if err != nil {
@@ -75,6 +92,60 @@ To connect via TCP, set the configuration string to:
 	// ...
 ```
 
+## Pooled Line Senders
+
+**Warning: Experimental feature designed for use with HTTP senders ONLY**
+
+Version 3 of the client introduces a `LineSenderPool`, which provides a mechanism
+to pool previously-used `LineSender`s so they can be reused without having
+to allocate and instantiate new senders.
+
+A LineSenderPool is thread-safe and can be used to concurrently obtain senders
+across multiple goroutines.
+
+Since `LineSender`s must be used in a single-threaded context, a typical pattern is to Acquire
+a sender from a `LineSenderPool` at the beginning of a goroutine and use a deferred
+execution block to Close the sender at the end of the goroutine.
+
+Here is an example of the `LineSenderPool` Acquire, Release, and Close semantics:
+
+```go
+package main
+
+import (
+	"context"
+
+	qdb "github.com/questdb/go-questdb-client/v3"
+)
+
+func main() {
+	ctx := context.TODO()
+
+	pool := qdb.PoolFromConf("http::addr=localhost:9000")
+	defer func() {
+		err := pool.Close(ctx)
+		if err != nil {
+			panic(err)
+		}
+	}()
+
+	sender, err := pool.Sender(ctx)
+	if err != nil {
+		panic(err)
+	}
+
+	sender.Table("prices").
+		Symbol("ticker", "AAPL").
+		Float64Column("price", 123.45).
+		AtNow(ctx)
+
+	// Close call returns the sender back to the pool
+	if err := sender.Close(ctx); err != nil {
+		panic(err)
+	}
+}
+```
+
 ## Migration from v2
 
 v2 code example:

conf_parse.go

@@ -37,7 +37,7 @@ type configData struct {
 }
 
 func confFromStr(conf string) (*lineSenderConfig, error) {
-	senderConf := &lineSenderConfig{}
+	var senderConf *lineSenderConfig
 
 	data, err := parseConfigStr(conf)
 	if err != nil {
@@ -46,14 +46,14 @@ func confFromStr(conf string) (*lineSenderConfig, error) {
 
 	switch data.Schema {
 	case "http":
-		senderConf.senderType = httpSenderType
+		senderConf = newLineSenderConfig(httpSenderType)
 	case "https":
-		senderConf.senderType = httpSenderType
+		senderConf = newLineSenderConfig(httpSenderType)
 		senderConf.tlsMode = tlsEnabled
 	case "tcp":
-		senderConf.senderType = tcpSenderType
+		senderConf = newLineSenderConfig(tcpSenderType)
 	case "tcps":
-		senderConf.senderType = tcpSenderType
+		senderConf = newLineSenderConfig(tcpSenderType)
 		senderConf.tlsMode = tlsEnabled
 	default:
 		return nil, fmt.Errorf("invalid schema: %s", data.Schema)
@@ -90,6 +90,7 @@ func confFromStr(conf string) (*lineSenderConfig, error) {
 		case "token_y":
 			// Some clients require public key.
 			// But since Go sender doesn't need it, we ignore the values.
+			continue
 		case "auto_flush":
 			if v == "off" {
 				senderConf.autoFlushRows = 0

conf_test.go

@@ -432,7 +432,15 @@ func TestHappyCasesFromConf(t *testing.T) {
 			actual, err := qdb.ConfFromStr(tc.config)
 			assert.NoError(t, err)
 
-			expected := &qdb.LineSenderConfig{}
+			var expected *qdb.LineSenderConfig
+			switch tc.config[0] {
+			case 'h':
+				expected = qdb.NewLineSenderConfig(qdb.HttpSenderType)
+			case 't':
+				expected = qdb.NewLineSenderConfig(qdb.TcpSenderType)
+			default:
+				assert.FailNow(t, "happy case configs must start with either 'http' or 'tcp'")
+			}
 			for _, opt := range tc.expectedOpts {
 				opt(expected)
 			}

export_test.go

@@ -29,10 +29,16 @@ type (
 	ConfigData       = configData
 	TcpLineSender    = tcpLineSender
 	LineSenderConfig = lineSenderConfig
+	SenderType       = senderType
 )
 
 var (
-	GlobalTransport = globalTransport
+	GlobalTransport                     = globalTransport
+	NoSenderType             SenderType = noSenderType
+	HttpSenderType           SenderType = httpSenderType
+	TcpSenderType            SenderType = tcpSenderType
+	DefaultAutoFlushInterval            = defaultAutoFlushInterval
+	DefaultAutoFlushRows                = defaultAutoFlushRows
 )
 
 func NewBuffer(initBufSize int, maxBufSize int, fileNameLimit int) Buffer {
@@ -58,6 +64,10 @@ func Messages(s LineSender) string {
 }
 
 func MsgCount(s LineSender) int {
+	if ps, ok := s.(*pooledSender); ok {
+		hs, _ := ps.wrapped.(*httpLineSender)
+		return hs.MsgCount()
+	}
 	if hs, ok := s.(*httpLineSender); ok {
 		return hs.MsgCount()
 	}
@@ -76,3 +86,7 @@ func BufLen(s LineSender) int {
 	}
 	panic("unexpected struct")
 }
+
+func NewLineSenderConfig(t SenderType) *LineSenderConfig {
+	return newLineSenderConfig(t)
+}

http_sender.go

@@ -29,7 +29,6 @@ import (
 	"context"
 	"crypto/tls"
 	"encoding/json"
-	"errors"
 	"fmt"
 	"io"
 	"math/big"
@@ -178,7 +177,7 @@ func (s *httpLineSender) flush0(ctx context.Context, closing bool) error {
 	)
 
 	if s.closed {
-		return errors.New("cannot flush a closed LineSender")
+		return errClosedSenderFlush
 	}
 
 	err := s.buf.LastErr()
@@ -189,7 +188,7 @@ func (s *httpLineSender) flush0(ctx context.Context, closing bool) error {
 	}
 	if s.buf.HasTable() {
 		s.buf.DiscardPendingMsg()
-		return errors.New("pending ILP message must be finalized with At or AtNow before calling Flush")
+		return errFlushWithPendingMessage
 	}
 
 	if s.buf.msgCount == 0 {
@@ -287,7 +286,7 @@ func (s *httpLineSender) BoolColumn(name string, val bool) LineSender {
 
 func (s *httpLineSender) Close(ctx context.Context) error {
 	if s.closed {
-		return nil
+		return errDoubleSenderClose
 	}
 
 	var err error
@@ -311,7 +310,7 @@ func (s *httpLineSender) AtNow(ctx context.Context) error {
 
 func (s *httpLineSender) At(ctx context.Context, ts time.Time) error {
 	if s.closed {
-		return errors.New("cannot queue new messages on a closed LineSender")
+		return errClosedSenderAt
 	}
 
 	sendTs := true

http_sender_test.go

@@ -502,6 +502,12 @@ func TestRowBasedAutoFlushWithTimeBasedFlushDisabled(t *testing.T) {
 
 	assert.Equal(t, autoFlushRows-1, qdb.MsgCount(sender))
 
+	// Sleep past the default interval
+	time.Sleep(qdb.DefaultAutoFlushInterval + time.Millisecond)
+
+	// Check that the number of messages hasn't changed
+	assert.Equal(t, autoFlushRows-1, qdb.MsgCount(sender))
+
 	// Send one additional message and ensure that all are flushed
 	err = sender.Table(testTable).StringColumn("bar", "baz").AtNow(ctx)
 	assert.NoError(t, err)
@@ -511,8 +517,6 @@ func TestRowBasedAutoFlushWithTimeBasedFlushDisabled(t *testing.T) {
 
 func TestNoFlushWhenAutoFlushDisabled(t *testing.T) {
 	ctx := context.Background()
-	autoFlushRows := 10
-	autoFlushInterval := time.Duration(autoFlushRows-1) * time.Millisecond
 
 	srv, err := newTestHttpServer(readAndDiscard)
 	assert.NoError(t, err)
@@ -524,21 +528,54 @@ func TestNoFlushWhenAutoFlushDisabled(t *testing.T) {
 		ctx,
 		qdb.WithHttp(),
 		qdb.WithAddress(srv.Addr()),
-		qdb.WithAutoFlushRows(autoFlushRows),
-		qdb.WithAutoFlushInterval(autoFlushInterval),
+		qdb.WithAutoFlushRows(qdb.DefaultAutoFlushRows),
+		qdb.WithAutoFlushInterval(qdb.DefaultAutoFlushInterval),
 		qdb.WithAutoFlushDisabled(),
 	)
 	assert.NoError(t, err)
 	defer sender.Close(ctx)
 
 	// Send autoFlushRows + 1 messages and ensure all are buffered
-	for i := 0; i < autoFlushRows+1; i++ {
+	for i := 0; i < qdb.DefaultAutoFlushRows+1; i++ {
 		err = sender.Table(testTable).StringColumn("bar", "baz").AtNow(ctx)
 		assert.NoError(t, err)
-		time.Sleep(time.Millisecond)
 	}
 
-	assert.Equal(t, autoFlushRows+1, qdb.MsgCount(sender))
+	// Sleep past the default interval
+	time.Sleep(qdb.DefaultAutoFlushInterval + time.Millisecond)
+
+	assert.Equal(t, qdb.DefaultAutoFlushRows+1, qdb.MsgCount(sender))
+}
+
+func TestNoFlushWhenAutoFlushRowsAndIntervalAre0(t *testing.T) {
+	ctx := context.Background()
+
+	srv, err := newTestHttpServer(readAndDiscard)
+	assert.NoError(t, err)
+	defer srv.Close()
+
+	// opts are processed sequentially, so AutoFlushDisabled will
+	// override AutoFlushRows
+	sender, err := qdb.NewLineSender(
+		ctx,
+		qdb.WithHttp(),
+		qdb.WithAddress(srv.Addr()),
+		qdb.WithAutoFlushRows(0),
+		qdb.WithAutoFlushInterval(0),
+	)
+	assert.NoError(t, err)
+	defer sender.Close(ctx)
+
+	// Send autoFlushRows + 1 messages and ensure all are buffered
+	for i := 0; i < qdb.DefaultAutoFlushRows+1; i++ {
+		err = sender.Table(testTable).StringColumn("bar", "baz").AtNow(ctx)
+		assert.NoError(t, err)
+	}
+
+	// Sleep past the default interval
+	time.Sleep(qdb.DefaultAutoFlushInterval + time.Millisecond)
+
+	assert.Equal(t, qdb.DefaultAutoFlushRows+1, qdb.MsgCount(sender))
 }
 
 func TestSenderDoubleClose(t *testing.T) {
@@ -564,7 +601,7 @@ func TestSenderDoubleClose(t *testing.T) {
 	assert.NoError(t, err)
 
 	err = sender.Close(ctx)
-	assert.NoError(t, err)
+	assert.Error(t, err)
 }
 
 func TestErrorOnFlushWhenSenderIsClosed(t *testing.T) {
@@ -625,7 +662,7 @@ func TestNoFlushWhenSenderIsClosedAndAutoFlushIsDisabled(t *testing.T) {
 
 	err = sender.Close(ctx)
 	assert.NoError(t, err)
-	assert.Empty(t, qdb.Messages(sender))
+	assert.NotEmpty(t, qdb.Messages(sender))
 }
 
 func TestSuccessAfterRetries(t *testing.T) {

integration_test.go

@@ -192,7 +192,7 @@ func setupQuestDB0(ctx context.Context, auth ilpAuthType, setupProxy bool) (*que
 	)
 	if setupProxy || auth == httpBasicAuth || auth == httpBearerAuth {
 		req = testcontainers.ContainerRequest{
-			Image:        "haproxy:2.6.0",
+			Image:        "haproxy:2.6.4",
 			ExposedPorts: []string{"8443/tcp", "8444/tcp", "8445/tcp", "8888/tcp"},
 			WaitingFor:   wait.ForHTTP("/").WithPort("8888"),
 			Networks:     []string{networkName},