Test how the logger behaves in async mode

New testcases have been written to test if PostWithTime() and Close()
are working as expected. More specifically, they test if Close() succeed
with all combination of options (Async + ForceStopAsyncSend).

Tests on PostWithTime() are divided into two parts:

* With transient network failure ;
* And without any ;

And tests on Close() are divided into two parts:

* CloseOnAsyncConnect: When no logs have been written yet ;
* CloseOnAsyncReconnect: When there're pending logs to write ;

Finally, the test `Test_send_WritePendingToConn` has been removed
because it was relying too much on the logger internals to test the
logger's behavior from an API user PoV. And the init() that was starting
a tcp listener has been removed to make the tests easier to execute
(otherwise, we've to wait for go test to timeout before re-running it).

Signed-off-by: Albin Kerouanton <[email protected]>

Author: Albin Kerouanton

fluent/fluent.go

@@ -15,8 +15,9 @@ import (
 	"bytes"
 	"encoding/base64"
 	"encoding/binary"
-	"github.com/tinylib/msgp/msgp"
 	"math/rand"
+
+	"github.com/tinylib/msgp/msgp"
 )
 
 const (
@@ -84,6 +85,7 @@ type msgToSend struct {
 type Fluent struct {
 	Config
 
+	dialer      dialer
 	stopRunning chan bool
 	pending     chan *msgToSend
 	wg          sync.WaitGroup
@@ -93,7 +95,20 @@ type Fluent struct {
 }
 
 // New creates a new Logger.
-func New(config Config) (f *Fluent, err error) {
+func New(config Config) (*Fluent, error) {
+	if config.Timeout == 0 {
+		config.Timeout = defaultTimeout
+	}
+	return newWithDialer(config, &net.Dialer{
+		Timeout: config.Timeout,
+	})
+}
+
+type dialer interface {
+	Dial(string, string) (net.Conn, error)
+}
+
+func newWithDialer(config Config, d dialer) (f *Fluent, err error) {
 	if config.FluentNetwork == "" {
 		config.FluentNetwork = defaultNetwork
 	}
@@ -106,9 +121,6 @@ func New(config Config) (f *Fluent, err error) {
 	if config.FluentSocketPath == "" {
 		config.FluentSocketPath = defaultSocketPath
 	}
-	if config.Timeout == 0 {
-		config.Timeout = defaultTimeout
-	}
 	if config.WriteTimeout == 0 {
 		config.WriteTimeout = defaultWriteTimeout
 	}
@@ -128,15 +140,20 @@ func New(config Config) (f *Fluent, err error) {
 		fmt.Fprintf(os.Stderr, "fluent#New: AsyncConnect is now deprecated, please use Async instead")
 		config.Async = config.Async || config.AsyncConnect
 	}
+
 	if config.Async {
 		f = &Fluent{
 			Config:  config,
+			dialer:  d,
 			pending: make(chan *msgToSend, config.BufferLimit),
 		}
 		f.wg.Add(1)
 		go f.run()
 	} else {
-		f = &Fluent{Config: config}
+		f = &Fluent{
+			Config: config,
+			dialer: d,
+		}
 		err = f.connect()
 	}
 	return
@@ -340,12 +357,15 @@ func (f *Fluent) close(c net.Conn) {
 
 // connect establishes a new connection using the specified transport.
 func (f *Fluent) connect() (err error) {
-
 	switch f.Config.FluentNetwork {
 	case "tcp":
-		f.conn, err = net.DialTimeout(f.Config.FluentNetwork, f.Config.FluentHost+":"+strconv.Itoa(f.Config.FluentPort), f.Config.Timeout)
+		f.conn, err = f.dialer.Dial(
+			f.Config.FluentNetwork,
+			f.Config.FluentHost+":"+strconv.Itoa(f.Config.FluentPort))
 	case "unix":
-		f.conn, err = net.DialTimeout(f.Config.FluentNetwork, f.Config.FluentSocketPath, f.Config.Timeout)
+		f.conn, err = f.dialer.Dial(
+			f.Config.FluentNetwork,
+			f.Config.FluentSocketPath)
 	default:
 		err = NewErrUnknownNetwork(f.Config.FluentNetwork)
 	}