From 6a81c0e9db100a9d8a927d525891ddf71785cd01 Mon Sep 17 00:00:00 2001
From: postmannen <postmannen@gmail.com>
Date: Wed, 17 Feb 2021 12:02:34 +0100
Subject: [PATCH] retreival and handling of previosly unhandled messages on
 program restart implemented

---
 incommmingBuffer.db | Bin 32768 -> 32768 bytes
 publisher.go        |   2 +-
 ringbuffer.go       |  64 +++++++++++++++++++++++++++++++++++++++-----
 textlogging.log     |   1 +
 4 files changed, 60 insertions(+), 7 deletions(-)

diff --git a/incommmingBuffer.db b/incommmingBuffer.db
index 2a70a16c3b0aefb4d63808926e55d60acee7bed2..0c1eac2f8ddf49209f5c8f281dc94bd320d67c0f 100644
GIT binary patch
delta 86
zcmZo@U}|V!n&2R?i~$0?ww>2n|2j}?v!cNU{z(E30!yJ1E3>zqi};YN1d&iEWi**w
dreM!$Y+0@3>7rz1ym_C(c6mmV&1DW<_5f)48pi+t

delta 86
zcmZo@U}|V!n&2SN!2khEY-QbNDV?(3tZ1--f0BTMKs!`|+ojH_Cw`MXL_(pI(Rgy1
df<32^Wwny0i;|Vm=6wp=<r$4PmpOFV0|44q7#RQn

diff --git a/publisher.go b/publisher.go
index 28465de..7c451a1 100644
--- a/publisher.go
+++ b/publisher.go
@@ -141,7 +141,7 @@ func (s *server) printProcessesMap() {
 // given to the system, and route them to the correct subject queue.
 func (s *server) handleMessagesInRingbuffer() {
 	// Prepare and start a new ring buffer
-	const bufferSize int = 100
+	const bufferSize int = 1000
 	rb := newringBuffer(bufferSize)
 	inCh := make(chan subjectAndMessage)
 	outCh := make(chan samDBValue)
diff --git a/ringbuffer.go b/ringbuffer.go
index b6acc3b..ac5a430 100644
--- a/ringbuffer.go
+++ b/ringbuffer.go
@@ -14,6 +14,7 @@ import (
 	"fmt"
 	"log"
 	"os"
+	"sort"
 	"strconv"
 	"sync"
 	"time"
@@ -79,6 +80,22 @@ func (r *ringBuffer) start(inCh chan subjectAndMessage, outCh chan samDBValue) {
 // fillBuffer will fill the buffer in the ringbuffer  reading from the inchannel.
 // It will also store the messages in a K/V DB while being processed.
 func (r *ringBuffer) fillBuffer(inCh chan subjectAndMessage, samValueBucket string, indexValueBucket string) {
+	// At startup get all the values that might be in the K/V store so we can
+	// put them into the buffer before we start to fill up with new incomming
+	// messages to the system.
+	// This is needed when the program have been restarted, and we need to check
+	// if there where previously unhandled messages that need to be handled first.
+	func() {
+		s, err := r.dumpBucket(samValueBucket)
+		if err != nil {
+			log.Printf("error: retreival of values from k/v store failed: %v\n", err)
+		}
+
+		for _, v := range s {
+			r.bufData <- v
+		}
+	}()
+
 	// Check for incomming messages. These are typically comming from
 	// the go routine who reads inmsg.txt.
 	for v := range inCh {
@@ -108,7 +125,7 @@ func (r *ringBuffer) fillBuffer(inCh chan subjectAndMessage, samValueBucket stri
 			log.Printf("error: dbUpdate samValue failed: %v\n", err)
 		}
 
-		// Send the message to some process to consume it.
+		// Put the message on the inmemory buffer.
 		r.bufData <- samV
 
 		// Increment index, and store the new value to the database.
@@ -158,13 +175,13 @@ func (r *ringBuffer) processBufferMessages(samValueBucket string, outCh chan sam
 			r.deleteKeyFromBucket(samValueBucket, strconv.Itoa(v.ID))
 			fmt.Printf("******* DELETED KEY %v FROM BUCKET*******\n", v.ID)
 
-			r.permStore <- fmt.Sprintf("%v : %+v\n", time.Now(), v)
+			r.permStore <- fmt.Sprintf("%v : %+v\n", time.Now().UTC(), v)
 
 			// TODO: Write the Key/Value we just deleted to a file acting
 			// as the transaction log for the system.
 
 			// REMOVE: Dump the whole KV store
-			err := r.dumpBucket(samValueBucket)
+			err := r.printBucketContent(samValueBucket)
 			if err != nil {
 				fmt.Printf("* Error: dump of db failed: %v\n", err)
 			}
@@ -175,9 +192,44 @@ func (r *ringBuffer) processBufferMessages(samValueBucket string, outCh chan sam
 	close(outCh)
 }
 
-// dumpBucket will print out all they keys and values in the
-// specified bucker.
-func (r *ringBuffer) dumpBucket(bucket string) error {
+// dumpBucket will dump out all they keys and values in the
+// specified bucket, and return a sorted []samDBValue
+func (r *ringBuffer) dumpBucket(bucket string) ([]samDBValue, error) {
+	samDBValues := []samDBValue{}
+
+	err := r.db.View(func(tx *bolt.Tx) error {
+		bu := tx.Bucket([]byte(bucket))
+
+		// For each element found in the DB, unmarshal, and put on slice.
+		bu.ForEach(func(k, v []byte) error {
+			var vv samDBValue
+			err := json.Unmarshal(v, &vv)
+			if err != nil {
+				log.Printf("error: dumpBucket json.Umarshal failed: %v\n", err)
+			}
+			samDBValues = append(samDBValues, vv)
+			return nil
+		})
+
+		// Sort the order of the slice items based on ID, since they where retreived from a map.
+		sort.SliceStable(samDBValues, func(i, j int) bool {
+			return samDBValues[i].ID > samDBValues[j].ID
+		})
+
+		fmt.Println("--------------------------K/V DUMP TO VARIABLE SORTED---------------------------")
+		for _, v := range samDBValues {
+			fmt.Printf("%#v\n", v)
+		}
+		fmt.Println("----------------------------------------------------------------------------------")
+		return nil
+	})
+
+	return samDBValues, err
+}
+
+// printBuckerContent will print out all they keys and values in the
+// specified bucket.
+func (r *ringBuffer) printBucketContent(bucket string) error {
 	err := r.db.View(func(tx *bolt.Tx) error {
 		bu := tx.Bucket([]byte(bucket))
 
diff --git a/textlogging.log b/textlogging.log
index 6365e5e..38dfdd3 100644
--- a/textlogging.log
+++ b/textlogging.log
@@ -19,3 +19,4 @@ some message sent from a ship for testing
 some message sent from a ship for testing
 some message sent from a ship for testing
 some message sent from a ship for testing
+some message sent from a ship for testing