Merge pull request #97 from ivan-valkov/async-error-callback

Add callback for error handling when using async

Author: tagomoris

.gitignore

@@ -1,4 +1,5 @@
 .DS_Store
 *~
 .project
-.settings
+.settings
+.idea

CHANGELOG.md

@@ -1,5 +1,8 @@
 # CHANGELOG
 
+## 1.6.2
+* Add `AsyncResultCallback` to allow users to handle errors when using asynchronous message sending. 
+
 ## 1.6.1
 * Add another fix for `Close` called twice in Async
 

README.md

@@ -121,6 +121,12 @@ The default is false.
 When Async is enabled, immediately discard the event queue on close() and return (instead of trying MaxRetry times for each event in the queue before returning)
 The default is false.
 
+### AsyncResultCallback
+
+When Async is enabled, if this is callback is provided, it will be called on every write to Fluentd. The callback function
+takes two arguments - a `[]byte` of the message that was to be sent and an `error`. If the `error` is not nil this means the 
+delivery of the message was unsuccessful.
+
 ### SubSecondPrecision
 
 Enable time encoding as EventTime, which contains sub-second precision values. The messages encoded with this option can be received only by Fluentd v0.14 or later.

fluent/fluent.go

@@ -38,19 +38,20 @@ const (
 )
 
 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"`
+	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)
 	// Deprecated: Use Async instead
 	AsyncConnect  bool `json:"async_connect"`
 	MarshalAsJSON bool `json:"marshal_as_json"`
@@ -406,6 +407,13 @@ func (f *Fluent) run() {
 			if err != nil {
 				fmt.Fprintf(os.Stderr, "[%s] Unable to send logs to fluentd, reconnecting...\n", time.Now().Format(time.RFC3339))
 			}
+			if f.AsyncResultCallback != nil {
+				var data []byte
+				if entry != nil {
+					data = entry.data
+				}
+				f.AsyncResultCallback(data, err)
+			}
 		}
 		select {
 		case stopRunning, ok := <-f.stopRunning: