Fix stuck on startup (#81)

* fix getting stuck during processing large pending queues

* updated dependencies

beater/journalbeat.go

@@ -15,6 +15,7 @@
 package beater
 
 import (
+	"encoding/json"
 	"fmt"
 	"io/ioutil"
 	"os"
@@ -122,6 +123,26 @@ func (jb *Journalbeat) initJournal() error {
 	return nil
 }
 
+func (jb *Journalbeat) publishPending() error {
+	pending := map[string]common.MapStr{}
+	file, err := os.Open(jb.config.PendingQueue.File)
+	if err != nil {
+		return err
+	}
+	defer file.Close()
+
+	if err = json.NewDecoder(file).Decode(&pending); err != nil {
+		return err
+	}
+
+	logp.Info("Loaded %d events, trying to publish", len(pending))
+	for cursor, event := range pending {
+		jb.client.PublishEvent(event, publisher.Signal(&eventSignal{&eventReference{cursor, event}, jb.completed}), publisher.Guaranteed)
+	}
+
+	return nil
+}
+
 // New creates beater
 func New(b *beat.Beat, cfg *common.Config) (beat.Beater, error) {
 	config := config.DefaultConfig
@@ -165,6 +186,11 @@ func (jb *Journalbeat) Run(b *beat.Beat) error {
 		go jb.writeCursorLoop()
 	}
 
+	// load the previously saved queue of unsent events and try to publish them if any
+	if err := jb.publishPending(); err != nil {
+		logp.Warn("could not read the pending queue: %s", err)
+	}
+
 	for rawEvent := range journal.Follow(jb.journal, jb.done) {
 		//convert sdjournal.JournalEntry to common.MapStr
 		event := MapStrFromJournalEntry(

beater/state.go

@@ -25,7 +25,6 @@ import (
 
 	"github.com/elastic/beats/libbeat/common"
 	"github.com/elastic/beats/libbeat/logp"
-	"github.com/elastic/beats/libbeat/publisher"
 )
 
 // eventSignal implements the op.Signaler interface
@@ -86,17 +85,6 @@ func (jb *Journalbeat) managePendingQueueLoop() {
 		return os.Rename(tempFile.Name(), dest)
 	}
 
-	// load loads the map[string]common.MapStr from the JSON file on disk
-	load := func(source string, dest *map[string]common.MapStr) error {
-		file, err := os.Open(source)
-		if err != nil {
-			return err
-		}
-		defer file.Close()
-
-		return json.NewDecoder(file).Decode(dest)
-	}
-
 	// on exit fully consume both queues and flush to disk the pending queue
 	defer func() {
 		var wg sync.WaitGroup
@@ -123,15 +111,6 @@ func (jb *Journalbeat) managePendingQueueLoop() {
 		}
 	}()
 
-	// load the previously saved queue of unsent events and try to publish them if any
-	if err := load(jb.config.PendingQueue.File, &pending); err != nil {
-		logp.Warn("could not read the pending queue: %s", err)
-	}
-	logp.Info("Loaded %d events, trying to publish", len(pending))
-	for cursor, event := range pending {
-		jb.client.PublishEvent(event, publisher.Signal(&eventSignal{&eventReference{cursor, event}, jb.completed}), publisher.Guaranteed)
-	}
-
 	// flush the pending queue to disk periodically
 	tick := time.Tick(jb.config.PendingQueue.FlushPeriod)
 	for {