fix(conf): disable auto_flush when set to off in config (#51)

Author: sklarsa

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 {
@@ -76,3 +82,7 @@ func BufLen(s LineSender) int {
 	}
 	panic("unexpected struct")
 }
+
+func NewLineSenderConfig(t SenderType) *LineSenderConfig {
+	return newLineSenderConfig(t)
+}

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) {
@@ -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) {

sender.go

@@ -481,13 +481,61 @@ func LineSenderFromConf(ctx context.Context, conf string) (LineSender, error) {
 // sender corresponds to a single client connection. LineSender should
 // not be called concurrently by multiple goroutines.
 func NewLineSender(ctx context.Context, opts ...LineSenderOption) (LineSender, error) {
-	conf := &lineSenderConfig{}
+	var conf *lineSenderConfig
+
+	// Iterate over all options to determine the sender type
+	// This is used to set defaults based on the type of sender (http vs tcp)
+	// Worst case performance is 2N for the number of LineSenderOptions
+	tmp := newLineSenderConfig(noSenderType)
+	for _, opt := range opts {
+		opt(tmp)
+		switch tmp.senderType {
+		case httpSenderType:
+			conf = newLineSenderConfig(httpSenderType)
+		case tcpSenderType:
+			conf = newLineSenderConfig(tcpSenderType)
+		}
+
+		if conf != nil {
+			break
+		}
+	}
+
+	if tmp.senderType == noSenderType {
+		return nil, errors.New("sender type is not specified: use WithHttp or WithTcp")
+	}
+
 	for _, opt := range opts {
 		opt(conf)
 	}
 	return newLineSender(ctx, conf)
 }
 
+func newLineSenderConfig(t senderType) *lineSenderConfig {
+	switch t {
+	case tcpSenderType:
+		return &lineSenderConfig{
+			senderType:    t,
+			address:       defaultTcpAddress,
+			initBufSize:   defaultInitBufferSize,
+			fileNameLimit: defaultFileNameLimit,
+		}
+	default:
+		return &lineSenderConfig{
+			senderType:        t,
+			address:           defaultHttpAddress,
+			requestTimeout:    defaultRequestTimeout,
+			retryTimeout:      defaultRetryTimeout,
+			minThroughput:     defaultMinThroughput,
+			autoFlushRows:     defaultAutoFlushRows,
+			autoFlushInterval: defaultAutoFlushInterval,
+			initBufSize:       defaultInitBufferSize,
+			maxBufSize:        defaultMaxBufferSize,
+			fileNameLimit:     defaultFileNameLimit,
+		}
+	}
+}
+
 func newLineSender(ctx context.Context, conf *lineSenderConfig) (LineSender, error) {
 	switch conf.senderType {
 	case tcpSenderType:
@@ -538,17 +586,6 @@ func sanitizeTcpConf(conf *lineSenderConfig) error {
 		return errors.New("tcpKeyId is empty and tcpKey is not. both (or none) must be provided")
 	}
 
-	// Set defaults
-	if conf.address == "" {
-		conf.address = defaultTcpAddress
-	}
-	if conf.initBufSize == 0 {
-		conf.initBufSize = defaultInitBufferSize
-	}
-	if conf.fileNameLimit == 0 {
-		conf.fileNameLimit = defaultFileNameLimit
-	}
-
 	return nil
 }
 
@@ -563,35 +600,6 @@ func sanitizeHttpConf(conf *lineSenderConfig) error {
 		return errors.New("both basic and token authentication cannot be used")
 	}
 
-	// Set defaults
-	if conf.address == "" {
-		conf.address = defaultHttpAddress
-	}
-	if conf.requestTimeout == 0 {
-		conf.requestTimeout = defaultRequestTimeout
-	}
-	if conf.retryTimeout == 0 {
-		conf.retryTimeout = defaultRetryTimeout
-	}
-	if conf.minThroughput == 0 {
-		conf.minThroughput = defaultMinThroughput
-	}
-	if conf.autoFlushRows == 0 {
-		conf.autoFlushRows = defaultAutoFlushRows
-	}
-	if conf.autoFlushInterval == 0 {
-		conf.autoFlushInterval = defaultAutoFlushInterval
-	}
-	if conf.initBufSize == 0 {
-		conf.initBufSize = defaultInitBufferSize
-	}
-	if conf.maxBufSize == 0 {
-		conf.maxBufSize = defaultMaxBufferSize
-	}
-	if conf.fileNameLimit == 0 {
-		conf.fileNameLimit = defaultFileNameLimit
-	}
-
 	return nil
 }