Address unbounded queue (Issue #20)

.travis.yml

@@ -6,11 +6,11 @@ go:
     - tip
 
 script:
-    - go test -coverprofile=coverage.txt -covermode=atomic
+    - go test -v -coverprofile=coverage.txt -covermode=atomic
 
 after_success:
     - bash <(curl -s https://codecov.io/bash)
 
 addons:
     code_climate:
-        repo_token: 42136a24528a7d52e98adbe0d5126c8dadda18158a8559eec3ee4f73a2053436
+        repo_token: 42136a24528a7d52e98adbe0d5126c8dadda18158a8559eec3ee4f73a2053436

README.md

@@ -89,19 +89,39 @@ func main() {
 	gomol.SetAttr("another_attr", 1234)
 
 	// Configure gomol to add the filename and line number that the
-	// log was generated from as well as the internal sequence number
-	// to help with ordering events if your log system doesn't support
-	// a small enough sub-second resolution.
+	// log was generated from, the internal sequence number to help
+	// with ordering events if your log system doesn't support a
+	// small enough sub-second resolution, and set the size of the
+	// internal queue (default is 10k messages).
 	cfg := gomol.NewConfig()
 	cfg.FilenameAttr = "filename"
 	cfg.LineNumberAttr = "line"
 	cfg.SequenceAttr = "sequence"
+	cfg.MaxQueueSize = 50000
 	gomol.SetConfig(cfg)
 
 	// Initialize the loggers
 	gomol.InitLoggers()
 	defer gomol.ShutdownLoggers()
 
+	// Create a channel on which to receive internal (asynchronous)
+	// logger errors. This is optional, but recommended in order to
+	// determine when logging may be dropping messages.
+	ch := make(chan error)
+
+	go func() {
+		// This consumer is expected to be efficient as writes to 
+		// the channel are blocking. If this handler is slow, the
+		// user should add a buffer to the channel, or manually
+		// queue and batch errors for processing.
+
+		for err := range ch {
+			fmt.Printf("[Internal Error] %s\n", err.Error())
+		}
+	}()
+
+	gomol.SetErrorChan(ch)
+
 	// Log some debug messages with message-level attrs
 	// that will be sent only with that message
 	for idx := 1; idx <= 10; idx++ {

base.go

@@ -12,6 +12,7 @@ are desired.
 type Base struct {
 	isInitialized bool
 	config        *Config
+	errorChan     chan<- error
 	queue         *queue
 	logLevel      LogLevel
 	sequence      uint64
@@ -25,7 +26,6 @@ type Base struct {
 func NewBase() *Base {
 	b := &Base{
 		config:       NewConfig(),
-		queue:        newQueue(),
 		logLevel:     LevelDebug,
 		sequence:     0,
 		BaseAttrs:    NewAttrs(),
@@ -55,6 +55,20 @@ func (b *Base) SetConfig(config *Config) {
 	b.config = config
 }
 
+// SetErrorChan will register a channel as the consumer of internal error
+// events.  This channel will be closed once ShutdownLoggers has finished.
+// The consumer of this channel is expected to be efficient as writing to
+// this channel will block.
+func (b *Base) SetErrorChan(ch chan<- error) {
+	b.errorChan = ch
+}
+
+func (b *Base) report(err error) {
+	if b.errorChan != nil {
+		b.errorChan <- err
+	}
+}
+
 /*
 SetLogLevel sets the level messages will be logged at.  It will log any message
 that is at the level or more severe than the level.
@@ -136,6 +150,7 @@ func (b *Base) ClearLoggers() error {
 			return err
 		}
 	}
+
 	b.loggers = make([]Logger, 0)
 	b.hookPreQueue = make([]HookPreQueue, 0)
 
@@ -153,14 +168,18 @@ If an error occurs in initializing a logger, the loggers that have already been
 initialized will continue to be initialized.
 */
 func (b *Base) InitLoggers() error {
+	if b.queue == nil {
+		b.queue = newQueue(b, b.config.MaxQueueSize)
+	}
+
 	for _, logger := range b.loggers {
 		err := logger.InitLogger()
 		if err != nil {
 			return err
 		}
 	}
 
-	b.queue.startQueueWorkers()
+	b.queue.startWorker()
 	b.isInitialized = true
 
 	return nil
@@ -169,7 +188,9 @@ func (b *Base) InitLoggers() error {
 // Flush will wait until all messages currently queued are distributed to
 // all initialized loggers
 func (b *Base) Flush() {
-	b.queue.Flush()
+	if b.queue != nil {
+		b.queue.flush()
+	}
 }
 
 /*
@@ -178,17 +199,23 @@ while shutting down a Logger, the error will be returned and all the loggers tha
 were already shut down will remain shut down.
 */
 func (b *Base) ShutdownLoggers() error {
-	b.queue.stopQueueWorkers()
-
 	for _, logger := range b.loggers {
 		err := logger.ShutdownLogger()
 		if err != nil {
 			return err
 		}
 	}
 
-	b.isInitialized = false
+	if b.queue != nil {
+		b.queue.stopWorker()
+	}
 
+	if b.errorChan != nil {
+		close(b.errorChan)
+		b.errorChan = nil
+	}
+
+	b.isInitialized = false
 	return nil
 }
 
@@ -262,7 +289,7 @@ func (b *Base) LogWithTime(level LogLevel, ts time.Time, m *Attrs, msg string, a
 		}
 	}
 
-	return b.queue.QueueMessage(nm)
+	return b.queue.queueMessage(nm)
 }
 
 // Log will log a message at the provided level to all added loggers with the timestamp set to the time

base_test.go

@@ -1,6 +1,7 @@
 package gomol
 
 import (
+	"errors"
 	"time"
 
 	"github.com/aphistic/sweet"
@@ -76,7 +77,6 @@ func (s *BaseSuite) TestNewBase(t sweet.T) {
 	Expect(b.config).ToNot(BeNil())
 	Expect(b.config.FilenameAttr).To(Equal(""))
 	Expect(b.config.LineNumberAttr).To(Equal(""))
-	Expect(b.queue).ToNot(BeNil())
 	Expect(b.logLevel).To(Equal(LevelDebug))
 	Expect(b.loggers).To(HaveLen(0))
 	Expect(b.BaseAttrs.Attrs()).To(HaveLen(0))
@@ -99,6 +99,34 @@ func (s *BaseSuite) TestSetConfig(t sweet.T) {
 	Expect(b.config.LineNumberAttr).To(Equal("line_number"))
 }
 
+func (s *BaseSuite) TestErrorChannel(t sweet.T) {
+	ch := make(chan error)
+	received := make(chan error, 3)
+
+	b := NewBase()
+	b.SetErrorChan(ch)
+
+	go func() {
+		defer close(received)
+
+		for val := range ch {
+			received <- val
+		}
+	}()
+
+	b.report(errors.New("error1"))
+	b.report(errors.New("error2"))
+	b.report(errors.New("error3"))
+
+	Eventually(received).Should(Receive(MatchError("error1")))
+	Eventually(received).Should(Receive(MatchError("error2")))
+	Eventually(received).Should(Receive(MatchError("error3")))
+
+	b.ShutdownLoggers()
+	Eventually(ch).Should(BeClosed())
+	Eventually(received).Should(BeClosed())
+}
+
 func (s *BaseSuite) TestSetLogLevel(t sweet.T) {
 	b := NewBase()
 	b.InitLoggers()
@@ -260,6 +288,7 @@ func (s *BaseSuite) TestBaseClearLoggers(t sweet.T) {
 func (s *BaseSuite) TestInitLoggers(t sweet.T) {
 	b := NewBase()
 	Expect(b.IsInitialized()).To(Equal(false))
+	Expect(b.queue).To(BeNil())
 
 	ml1 := newDefaultMemLogger()
 	ml2 := newDefaultMemLogger()
@@ -268,6 +297,7 @@ func (s *BaseSuite) TestInitLoggers(t sweet.T) {
 	b.AddLogger(ml2)
 
 	b.InitLoggers()
+	Expect(b.queue).ToNot(BeNil())
 
 	Expect(b.IsInitialized()).To(Equal(true))
 	Expect(ml1.IsInitialized()).To(Equal(true))

config.go

@@ -5,13 +5,20 @@ type Config struct {
 	// FilenameAttr is the name of the attribute to put the log location's
 	// filename in.  This comes at a slight performance penalty.
 	FilenameAttr string
+
 	// LineNumberAttr is the name of the attribute to put the log location's
 	// line number in.  This comes at a slight performance penalty.
 	LineNumberAttr string
+
 	// SequenceAttr is the name of the attribute to put the log message's sequence
 	// number in.  The sequence number is an incrementing number for each log message
 	// processed by a Base.
 	SequenceAttr string
+
+	// MaxQueueSize is the number of log messages which will be queued before old
+	// messages are discarded.  This value takes effect once InitLoggers is called.
+	// Further changes to this value will not increase or decrease the queue size.
+	MaxQueueSize uint
 }
 
 // NewConfig creates a new configuration with default settings
@@ -20,5 +27,6 @@ func NewConfig() *Config {
 		FilenameAttr:   "",
 		LineNumberAttr: "",
 		SequenceAttr:   "",
+		MaxQueueSize:   10000,
 	}
 }

default.go

@@ -16,6 +16,11 @@ func SetConfig(config *Config) {
 	curDefault.SetConfig(config)
 }
 
+// SetErrorChan executes the same function on the default Base instance
+func SetErrorChan(ch chan<- error) {
+	curDefault.SetErrorChan(ch)
+}
+
 // SetLogLevel executes the same function on the default Base instance
 func SetLogLevel(level LogLevel) {
 	curDefault.SetLogLevel(level)