Fix race condition (marcinwyszynski#11)

group.go

@@ -75,6 +75,7 @@ func (g *groupImpl) Open(ctx context.Context, streamName string) io.ReadCloser {
 func (g *groupImpl) create(ctx context.Context, streamName string) (*writerImpl, error) {
 	ret := &writerImpl{
 		client:     g,
+		closeChan:  make(chan struct{}),
 		ctx:        ctx,
 		events:     newEventsBuffer(),
 		groupName:  aws.String(g.groupName),

group_test.go

@@ -56,8 +56,6 @@ func (gs *groupTestSuite) TestCreateWithExistingStream_OK() {
 }
 
 func (gs *groupTestSuite) TestCreateWithExistingStream_UnexpectedFailure() {
-	const sequenceToken = "sequenceToken"
-
 	gs.creatingLogStreamReturns(errors.New("bacon"))
 
 	writer, err := gs.sut.Create(gs.ctx, gs.streamName)

interface.go

@@ -2,7 +2,6 @@ package cloudwatch
 
 import (
 	"context"
-	"fmt"
 	"io"
 
 	"github.com/aws/aws-sdk-go/service/cloudwatchlogs/cloudwatchlogsiface"
@@ -17,7 +16,7 @@ type RejectedLogEventsInfoError struct {
 }
 
 func (e *RejectedLogEventsInfoError) Error() string {
-	return fmt.Sprintf("log messages were rejected")
+	return "log messages were rejected"
 }
 
 // CreateOption allows setting various options on the resulting writer.

writer.go

@@ -20,8 +20,9 @@ type writerImpl struct {
 
 	ctx context.Context
 
-	closed bool
-	err    error
+	closeChan chan (struct{})
+	closed    bool
+	err       error
 
 	events  *eventsBuffer
 	nowFunc func() time.Time
@@ -71,15 +72,15 @@ func (w *writerImpl) Write(b []byte) (int, error) {
 }
 
 // Start continuously flushing the buffered events.
-func (w *writerImpl) start() error {
+func (w *writerImpl) start() (err error) {
 	for {
-		// Exit if the stream is closed.
-		if w.closed {
-			return nil
-		}
-
-		if err := w.flushTrottled(); err != nil {
-			return err
+		select {
+		case <-w.closeChan:
+			return
+		case <-w.throttle.C:
+			if err = w.flushBatch(); err != nil {
+				return
+			}
 		}
 	}
 }
@@ -90,11 +91,14 @@ func (w *writerImpl) Close() error {
 	defer w.throttle.Stop()
 
 	w.closed = true
+	close(w.closeChan)
+
 	for w.events.hasMore() {
 		if w.flushTrottled() != nil {
 			break
 		}
 	}
+
 	return w.err
 }