func (f *Fluent) Close() (err error) {
	defer f.close(f.conn)
	if f.Config.Async {
		f.pendingMutex.Lock()
		if f.chanClosed {
			f.pendingMutex.Unlock()
			return nil
		}
		f.chanClosed = true
		f.pendingMutex.Unlock()
		if f.Config.ForceStopAsyncSend {
			f.stopRunning <- true
			close(f.stopRunning)
		}
		close(f.pending)
		f.wg.Wait()
	}
	return nil
}
Close closes the connection, waiting for pending logs to be sent
package fluent
import (
	"encoding/json"
	"errors"
	"fmt"
	"math"
	"net"
	"os"
	"reflect"
	"strconv"
	"sync"
	"time"
	"bytes"
	"encoding/base64"
	"encoding/binary"
	"math/rand"
	"github.com/tinylib/msgp/msgp"
)
const (
	defaultHost                   = "127.0.0.1"
	defaultNetwork                = "tcp"
	defaultSocketPath             = ""
	defaultPort                   = 24224
	defaultTimeout                = 3 * time.Second
	defaultWriteTimeout           = time.Duration(0)  
	defaultBufferLimit            = 8 * 1024
	defaultRetryWait              = 500
	defaultMaxRetryWait           = 60000
	defaultMaxRetry               = 13
	defaultReconnectWaitIncreRate = 1.5
	defaultSubSecondPrecision = false
)
type Config struct {
	FluentPort          int           `json:"fluent_port"`
	FluentHost          string        `json:"fluent_host"`
	FluentNetwork       string        `json:"fluent_network"`
	FluentSocketPath    string        `json:"fluent_socket_path"`
	Timeout             time.Duration `json:"timeout"`
	WriteTimeout        time.Duration `json:"write_timeout"`
	BufferLimit         int           `json:"buffer_limit"`
	RetryWait           int           `json:"retry_wait"`
	MaxRetry            int           `json:"max_retry"`
	MaxRetryWait        int           `json:"max_retry_wait"`
	TagPrefix           string        `json:"tag_prefix"`
	Async               bool          `json:"async"`
	ForceStopAsyncSend  bool          `json:"force_stop_async_send"`
	AsyncResultCallback func(data []byte, err error)
	AsyncConnect  bool `json:"async_connect"`
	MarshalAsJSON bool `json:"marshal_as_json"`
	SubSecondPrecision bool `json:"sub_second_precision"`
	RequestAck bool `json:"request_ack"`
}
type ErrUnknownNetwork struct {
	network string
}
func (e *ErrUnknownNetwork) Error() string {
	return "unknown network " + e.network
}
func NewErrUnknownNetwork(network string) error {
	return &ErrUnknownNetwork{network}
}
type msgToSend struct {
	data []byte
	ack  string
}
type Fluent struct {
	Config
	dialer       dialer
	stopRunning  chan bool
	pending      chan *msgToSend
	pendingMutex sync.RWMutex
	chanClosed   bool
	wg           sync.WaitGroup
	muconn sync.Mutex
	conn   net.Conn
}
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
	}
	if config.FluentHost == "" {
		config.FluentHost = defaultHost
	}
	if config.FluentPort == 0 {
		config.FluentPort = defaultPort
	}
	if config.FluentSocketPath == "" {
		config.FluentSocketPath = defaultSocketPath
	}
	if config.WriteTimeout == 0 {
		config.WriteTimeout = defaultWriteTimeout
	}
	if config.BufferLimit == 0 {
		config.BufferLimit = defaultBufferLimit
	}
	if config.RetryWait == 0 {
		config.RetryWait = defaultRetryWait
	}
	if config.MaxRetry == 0 {
		config.MaxRetry = defaultMaxRetry
	}
	if config.MaxRetryWait == 0 {
		config.MaxRetryWait = defaultMaxRetryWait
	}
	if config.AsyncConnect {
		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),
			pendingMutex: sync.RWMutex{},
			stopRunning:  make(chan bool, 1),
		}
		f.wg.Add(1)
		go f.run()
	} else {
		f = &Fluent{
			Config: config,
			dialer: d,
		}
		err = f.connect()
	}
	return
}
func (f *Fluent) Post(tag string, message interface{}) error {
	timeNow := time.Now()
	return f.PostWithTime(tag, timeNow, message)
}
func (f *Fluent) PostWithTime(tag string, tm time.Time, message interface{}) error {
	if len(f.TagPrefix) > 0 {
		tag = f.TagPrefix + "." + tag
	}
	if m, ok := message.(msgp.Marshaler); ok {
		return f.EncodeAndPostData(tag, tm, m)
	}
	msg := reflect.ValueOf(message)
	msgtype := msg.Type()
	if msgtype.Kind() == reflect.Struct {
		kv := make(map[string]interface{})
		fields := msgtype.NumField()
		for i := 0; i < fields; i++ {
			field := msgtype.Field(i)
			value := msg.FieldByIndex(field.Index)
			if !value.CanInterface() {
				continue
			}
			name := field.Name
			if n1 := field.Tag.Get("msg"); n1 != "" {
				name = n1
			} else if n2 := field.Tag.Get("codec"); n2 != "" {
				name = n2
			}
			kv[name] = value.Interface()
		}
		return f.EncodeAndPostData(tag, tm, kv)
	}
	if msgtype.Kind() != reflect.Map {
		return errors.New("fluent#PostWithTime: message must be a map")
	} else if msgtype.Key().Kind() != reflect.String {
		return errors.New("fluent#PostWithTime: map keys must be strings")
	}
	kv := make(map[string]interface{})
	for _, k := range msg.MapKeys() {
		kv[k.String()] = msg.MapIndex(k).Interface()
	}
	return f.EncodeAndPostData(tag, tm, kv)
}
func (f *Fluent) EncodeAndPostData(tag string, tm time.Time, message interface{}) error {
	var msg *msgToSend
	var err error
	if msg, err = f.EncodeData(tag, tm, message); err != nil {
		return fmt.Errorf("fluent#EncodeAndPostData: can't convert '%#v' to msgpack:%v", message, err)
	}
	return f.postRawData(msg)
}
func (f *Fluent) PostRawData(msg *msgToSend) {
	f.postRawData(msg)
}
func (f *Fluent) postRawData(msg *msgToSend) error {
	if f.Config.Async {
		return f.appendBuffer(msg)
	}
	return f.write(msg)
}
type MessageChunk struct {
	message Message
}
func (chunk *MessageChunk) MarshalJSON() ([]byte, error) {
	data, err := json.Marshal(chunk.message.Record)
	if err != nil {
		return nil, err
	}
	option, err := json.Marshal(chunk.message.Option)
	if err != nil {
		return nil, err
	}
	return []byte(fmt.Sprintf("[\"%s\",%d,%s,%s]", chunk.message.Tag,
		chunk.message.Time, data, option)), err
}
func getUniqueID(timeUnix int64) (string, error) {
	buf := bytes.NewBuffer(nil)
	enc := base64.NewEncoder(base64.StdEncoding, buf)
	if err := binary.Write(enc, binary.LittleEndian, timeUnix); err != nil {
		enc.Close()
		return "", err
	}
	if err := binary.Write(enc, binary.LittleEndian, rand.Uint64()); err != nil {
		enc.Close()
		return "", err
	}
	enc.Close()
	return buf.String(), nil
}
func (f *Fluent) EncodeData(tag string, tm time.Time, message interface{}) (msg *msgToSend, err error) {
	option := make(map[string]string)
	msg = &msgToSend{}
	timeUnix := tm.Unix()
	if f.Config.RequestAck {
		var err error
		msg.ack, err = getUniqueID(timeUnix)
		if err != nil {
			return nil, err
		}
		option["chunk"] = msg.ack
	}
	if f.Config.MarshalAsJSON {
		m := Message{Tag: tag, Time: timeUnix, Record: message, Option: option}
		chunk := &MessageChunk{message: m}
		msg.data, err = json.Marshal(chunk)
	} else if f.Config.SubSecondPrecision {
		m := &MessageExt{Tag: tag, Time: EventTime(tm), Record: message, Option: option}
		msg.data, err = m.MarshalMsg(nil)
	} else {
		m := &Message{Tag: tag, Time: timeUnix, Record: message, Option: option}
		msg.data, err = m.MarshalMsg(nil)
	}
	return
}