Add ErrorHandlerFunc

Author: daichirata

benchmark/benchmark_test.go

@@ -140,6 +140,9 @@ func BenchmarkStructDaichirata(b *testing.B) {
 		panic(err)
 	}
 	defer logger.Close()
+	logger.ErrorHandler = daichirata.ErrorHandlerFunc(func(err error, _ []byte) error {
+		panic(err)
+	})
 
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {
@@ -170,6 +173,9 @@ func BenchmarkMapDaichirata(b *testing.B) {
 		panic(err)
 	}
 	defer logger.Close()
+	logger.ErrorHandler = daichirata.ErrorHandlerFunc(func(err error, _ []byte) error {
+		panic(err)
+	})
 
 	b.ResetTimer()
 	for i := 0; i < b.N; i++ {

buffer.go

@@ -38,11 +38,11 @@ func (buffer *buffer) Remove() []*Message {
 	}
 
 	var messages []*Message
-	messages = append(messages, buffer.new...)
 	messages = append(messages, buffer.pending...)
+	messages = append(messages, buffer.new...)
 
-	buffer.new = buffer.new[:0]
 	buffer.pending = buffer.pending[:0]
+	buffer.new = buffer.new[:0]
 	return messages
 }
 

error_handler.go

@@ -0,0 +1,93 @@
+package fluent
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"io"
+
+	"gopkg.in/vmihailenco/msgpack.v2"
+)
+
+type ErrorHandler interface {
+	HandleError(error, []byte) error
+}
+
+type ErrorHandlerFunc func(error, []byte) error
+
+func (f ErrorHandlerFunc) HandleError(err error, data []byte) error {
+	return f(err, data)
+}
+
+type FallbackHandler struct {
+	logger *Logger
+}
+
+func NewFallbackHandler(logger *Logger) *FallbackHandler {
+	return &FallbackHandler{
+		logger: logger,
+	}
+}
+
+func (h *FallbackHandler) HandleError(_ error, data []byte) error {
+	return h.logger.writeWithBreaker(data)
+}
+
+type FallbackJSONHandler struct {
+	io io.Writer
+}
+
+func NewFallbackJSONHandler(io io.Writer) *FallbackJSONHandler {
+	return &FallbackJSONHandler{io: io}
+}
+
+func (h *FallbackJSONHandler) HandleError(_ error, data []byte) error {
+	r := bytes.NewReader(data)
+	d := msgpack.NewDecoder(r)
+
+	for r.Len() > 0 {
+		var tag string
+		var timestamp uint64
+		iRecord := map[interface{}]interface{}{}
+
+		message := []interface{}{&tag, &timestamp, &iRecord}
+		if err := d.Decode(&message); err != nil {
+			return err
+		}
+		record := h.castMap(iRecord)
+
+		str, err := json.Marshal([]interface{}{tag, timestamp, record})
+		if err != nil {
+			return err
+		}
+		_, err = h.io.Write(append(str, "\n"...))
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (h *FallbackJSONHandler) castMap(in map[interface{}]interface{}) map[string]interface{} {
+	res := make(map[string]interface{})
+	for k, v := range in {
+		res[fmt.Sprintf("%v", k)] = h.castValue(v)
+	}
+	return res
+}
+
+func (h *FallbackJSONHandler) castValue(iv interface{}) interface{} {
+	switch val := iv.(type) {
+	case map[interface{}]interface{}:
+		return h.castMap(val)
+	case []interface{}:
+		ia := make([]interface{}, len(val))
+		for i, v := range val {
+			ia[i] = h.castValue(v)
+		}
+		return ia
+	default:
+		return iv
+	}
+}

logger.go

@@ -1,6 +1,7 @@
 package fluent
 
 import (
+	"fmt"
 	"io"
 	"net"
 	"strings"
@@ -19,8 +20,9 @@ var (
 type Config struct {
 	Address           string
 	ConnectionTimeout time.Duration
-	FlushInterval     time.Duration
 	FailureThreshold  int64
+	FlushInterval     time.Duration
+	PendingLimit      int
 }
 
 func withDefaultConfig(c Config) Config {
@@ -37,6 +39,8 @@ func withDefaultConfig(c Config) Config {
 }
 
 type Logger struct {
+	ErrorHandler ErrorHandler
+
 	conf    Config
 	conn    io.WriteCloser
 	buf     buffer
@@ -153,10 +157,17 @@ func (logger *Logger) send() error {
 	for _, m := range messages {
 		data = append(data, m.buf.Bytes()...)
 	}
-	err := logger.write(data)
+
+	err := logger.writeWithBreaker(data)
 	if err != nil {
-		logger.buf.Back(messages)
-		return err
+		if logger.ErrorHandler != nil && len(messages) > logger.conf.PendingLimit {
+			err = logger.ErrorHandler.HandleError(err, data)
+		}
+		if err != nil {
+			fmt.Println(err)
+			logger.buf.Back(messages)
+			return err
+		}
 	}
 
 	for _, m := range messages {
@@ -173,18 +184,12 @@ func (logger *Logger) start() {
 		for {
 			select {
 			case <-logger.done:
-				err := logger.send()
-				if err != nil {
-					panic(err)
-				}
+				logger.send()
 				return
 			case <-logger.buf.Dirty:
 			case <-ticker.C:
 			}
-			err := logger.send()
-			if err != nil {
-				panic(err)
-			}
+			logger.send()
 		}
 	}()
 }