From 1f998b3b6ac41e0e61691f5ceed3fd91b36080e2 Mon Sep 17 00:00:00 2001
From: postmannen <postmannen@gmail.com>
Date: Tue, 16 Feb 2021 14:29:32 +0100
Subject: [PATCH] changed types used for ringbuffer, and fixed mutex on message
 index

---
 incommmingBuffer.db | Bin 65536 -> 32768 bytes
 publisher.go        |   5 ++--
 ringbuffer.go       |  56 ++++++++++++++++++++++++++++++--------------
 3 files changed, 42 insertions(+), 19 deletions(-)

diff --git a/incommmingBuffer.db b/incommmingBuffer.db
index e24a958bf2043bda208d43da2cb3da79626e9834..d62e593519051fb17af92f2cccfb39743245421a 100644
GIT binary patch
delta 233
zcmZo@U}<PznxMeQGWnywxC9#m1o%N|C8upE0k>aTZZ<U7z&}aA0<ORxs^D;zp*~}(
zRwqP(f)1m?WL*VMexu^V+_1!)(p0C?<m}XviGdQ66gW0UMDtHF;NoWj>t%tOz`EIR
z!*l*g4qTJD6qK0c87J#0cwo^#IYL1|2&D5Z5E~g)D|xyoSs58kZgf?coELI*Q3l7x
Xgc!C-0UUw?3=j(-Bm)C8$O3%;=s`Cp

literal 65536
zcmeI*O^X~w7{Kw)_!3{@vOy5^H1<u{#O>Z>Uq(-xD42tKkmTShbY@z2lX;Py8Ch5c
zJme&Z;#vFxLJs-?^dl&G5)XplMHf6McoBT6`>9ZAW>Raj6K8t<4L#G<Z#DJnPR%@3
z)!87(rw8@@=*KfhHM6{&uXs$?+qcJc;f%gF^!=w>UtIZk=hw4#Ki&~Q009ILKmY**
z5I_I{1Q0-Ab_7o8N0}7=XKew_$9+JTZT$a&rvG{U=U}OR>knsF77;)I0R#|0009IL
zKmY**5I|r?1Z;d?3-Tq?9#EIrbpLP2Y|__cUXt0Y+v~Q{gR4n<m~Ic7H`9Tv8avgp
zBlAt>xQApmIb_GjTm7WtoaE7bKVA_)009ILKmY**5I_I{1Q0-AIs&#oe>uoU0N8N=
zk7#yw{J$OdKb=p(SrI@00R#|0009ILKmY**5I|s>0yf^S1$IP#5PTyUn!ftqHRI1^
z5P(S^X?i3HH@zc`2=4w3IjOt-Y3cko;|PA!<Nqq8zk4{}Zu7f+`tJQdl*S16_;+Ny
zfScZyMhcsb*ABy1E{BbHZ6&;%43e;M81{$PZ=}tES?u;&Y1jz+yL<1&;Y!%-bvjA6
z_1gYR@26ea!LE=^opi9<Yi0YU?e?X7$8cme^!v%Xa+bqz(0ldn^X<t2t1DqE%lOUk
zdeYyOQ=Mzd%V0lE>*vouAHG#0+m-O;{a!~iYsyF6Puk&VG>R{1X)pLn8=Y`z&H3(8
z+IGKwu6^>wy{N^#xERG<*W#}Gid%_$c|%LP;VW$=?q#i~%2iXgI=au*lrP&!)s)s!
z-b!lhZS2c*N>#Ei+eu~BDrzxmD6OE%)lWX*8N0ArJ$Y-VD%`}Ic~!jWtBI>n)~&p(
zTcyg{)Zc~oxgz#uJE?NgAK<(pfB*s_V8-Q}sTSBC{~-84(l(wp+jb=3rZ+YIxAFQ@
zh4}wZP22W@?+fh#UrMHKd%-6X|J(FxUK>jgT_Sp)vtJZZUe;(K%BmMODtBdlUwBjc
zt?0{k()Vm+FKxMN>&tf5N*n8q9Q*F%N<{9FhVGF!cCSWpHF}TAh41rrFKhfBc?0;d
zvCo{9)_l7&A%Fk^2q1vKq6tXoolV8~-^Sx+<J%JayXhAt;ijJ{#Q%Rw+HEiRp%DLn
zrSZRQANbUX|0C^Eh|2XT%-Qd+NIMmxa=i-kR$A>=h|2XV%v))-V<B3Io&{|J@U{Sq
zjhZg~KfNfvY$vtu#M+h>d)u-q(Et!@V^-{K%&JCl?X6tyUY@gBD*yK96#)bgKmY**
z7D+&c+GkTOc;Zg6zm+97zQ3*Uzm3;F()iz|+lBc5N+I5VuF&6qTF3j_=K+2xq`%hj
z|8{)AXNB?q2ZizfZ#d)sO(9n0KP#~94?GVr{^^Hm{8Z<8fbmy%Y?M7`kx9H*&jZBT
zI2JEN;}}mr9P@!RUmMkwh64GrozyhO(+|ge($FV7V;7dEAFBBemZu-?ggq=Euy6u)
z)c>hKf_Rx$Wj3+1eJ<cxSvG9}cAUVInwAhhn^vRscukjiLjVB;5I_I{1Q0*~0R#|0
z0D;*Quzz~*zsCP|+yL?aY+hGB6#@t#fB*srAb<b@2q1s}0<M7l`+l{omp||Ougeb0
zGH=Li#sXZEc}ZroZm-))53VNdVY)qR-b@FwYV1_Yj?6ci<IH#hGmaoTKHlsn9p@yr
z5hCB6R|F71009ILKmY**5I_I{1Q0l30eMtDo0fxoM8ApA&6Gb45a`;#`Rf4rc3u%c
z009ILKmY**5I_I{1Q0*~0R#|0009ILKmY**5I_I{1Q0*~0R#|0009ILKmY**5I_I{
z1Q0*~0R#|0009ILKmY**5I_I{1Q0*~0R#|0009ILKmY**5I_I{1Q0;r_yYd|83VG$

diff --git a/publisher.go b/publisher.go
index 3e454e5..c11869c 100644
--- a/publisher.go
+++ b/publisher.go
@@ -144,7 +144,7 @@ func (s *server) handleMessagesInRingbuffer() {
 	const bufferSize int = 100
 	rb := newringBuffer(bufferSize)
 	inCh := make(chan subjectAndMessage)
-	outCh := make(chan subjectAndMessage)
+	outCh := make(chan samDBValue)
 	rb.start(inCh, outCh)
 
 	// Start reading new messages received on the incomming message
@@ -165,7 +165,8 @@ func (s *server) handleMessagesInRingbuffer() {
 	// send if there are a specific subject for it, and no subject
 	// exist throw an error.
 	go func() {
-		for sam := range outCh {
+		for samTmp := range outCh {
+			sam := samTmp.Data
 			// Check if the format of the message is correct.
 			// TODO: Send a message to the error kernel here that
 			// it was unable to process the message with the reason
diff --git a/ringbuffer.go b/ringbuffer.go
index 8dff90a..bcffb13 100644
--- a/ringbuffer.go
+++ b/ringbuffer.go
@@ -14,6 +14,7 @@ import (
 	"fmt"
 	"log"
 	"strconv"
+	"sync"
 
 	bolt "go.etcd.io/bbolt"
 )
@@ -28,9 +29,10 @@ type samDBValue struct {
 
 // ringBuffer holds the data of the buffer,
 type ringBuffer struct {
-	bufData            chan subjectAndMessage
+	bufData            chan samDBValue
 	db                 *bolt.DB
 	totalMessagesIndex int
+	mu                 sync.Mutex
 }
 
 // newringBuffer is a push/pop storage for values.
@@ -40,7 +42,7 @@ func newringBuffer(size int) *ringBuffer {
 		log.Printf("error: failed to open db: %v\n", err)
 	}
 	return &ringBuffer{
-		bufData: make(chan subjectAndMessage, size),
+		bufData: make(chan samDBValue, size),
 		db:      db,
 	}
 }
@@ -48,7 +50,7 @@ func newringBuffer(size int) *ringBuffer {
 // start will process incomming messages through the inCh,
 // put the messages on a buffered channel
 // and deliver messages out when requested on the outCh.
-func (r *ringBuffer) start(inCh chan subjectAndMessage, outCh chan subjectAndMessage) {
+func (r *ringBuffer) start(inCh chan subjectAndMessage, outCh chan samDBValue) {
 	// Starting both writing and reading in separate go routines so we
 	// can write and read concurrently.
 
@@ -71,9 +73,15 @@ func (r *ringBuffer) fillBuffer(inCh chan subjectAndMessage, samValueBucket stri
 	for v := range inCh {
 		// --- Store the incomming message in the k/v store ---
 
+		// Get a unique number for the message to use when storing
+		// it in the databases, and also use when further processing.
+		r.mu.Lock()
+		dbID := r.totalMessagesIndex
+		r.mu.Unlock()
+
 		// Create a structure for JSON marshaling.
 		samV := samDBValue{
-			ID:   r.totalMessagesIndex,
+			ID:   dbID,
 			Data: v,
 		}
 
@@ -83,20 +91,23 @@ func (r *ringBuffer) fillBuffer(inCh chan subjectAndMessage, samValueBucket stri
 		}
 
 		// Store the incomming message in key/value store
-		err = r.dbUpdate(r.db, samValueBucket, strconv.Itoa(r.totalMessagesIndex), js)
+		err = r.dbUpdate(r.db, samValueBucket, strconv.Itoa(dbID), js)
 		if err != nil {
 			// TODO: Handle error
 			log.Printf("error: dbUpdate samValue failed: %v\n", err)
 		}
 
 		// Send the message to some process to consume it.
-		r.bufData <- v
+		r.bufData <- samV
 
 		// Increment index, and store the new value to the database.
+		r.mu.Lock()
 		r.totalMessagesIndex++
 		fmt.Printf("*** NEXT INDEX NUMBER INCREMENTED: %v\n", r.totalMessagesIndex)
+		r.mu.Unlock()
+
 		fmt.Println("---------------------------------------------------------")
-		r.dbUpdate(r.db, indexValueBucket, "index", []byte(strconv.Itoa(r.totalMessagesIndex)))
+		r.dbUpdate(r.db, indexValueBucket, "index", []byte(strconv.Itoa(dbID)))
 	}
 
 	// When done close the buffer channel
@@ -107,7 +118,7 @@ func (r *ringBuffer) fillBuffer(inCh chan subjectAndMessage, samValueBucket stri
 // one by one. The messages will be delivered on the outCh, and it will wait
 // until a signal is received on the done channel before it continues with the
 // next message.
-func (r *ringBuffer) processBufferMessages(samValueBucket string, outCh chan subjectAndMessage) {
+func (r *ringBuffer) processBufferMessages(samValueBucket string, outCh chan samDBValue) {
 	// Range over the buffer of messages to pass on to processes.
 	for v := range r.bufData {
 		// Create a done channel per message. A process started by the
@@ -120,16 +131,19 @@ func (r *ringBuffer) processBufferMessages(samValueBucket string, outCh chan sub
 		// it's own go routine. Reason is that we don't want to block other
 		// messages being blocked while waiting for the done signal, or if an
 		// error with an individual message occurs.
-		go func(v subjectAndMessage) {
-			v.Message.done = make(chan struct{})
+		go func(v samDBValue) {
+			v.Data.Message.done = make(chan struct{})
 			outCh <- v
 
 			// ----------TESTING
-			<-v.done
+			<-v.Data.done
 			fmt.Println("-----------------------------------------------------------")
 			fmt.Printf("### DONE WITH THE MESSAGE\n")
 			fmt.Println("-----------------------------------------------------------")
 
+			r.deleteKeyFromBucket(samValueBucket, strconv.Itoa(v.ID))
+			fmt.Printf("******* DELETED KEY %v FROM BUCKET*******", v.ID)
+
 			// TODO: Delete the messages here. The SAM handled here, do
 			// not contain the totalMessageID, so we might need to change
 			// the struct we pass around.
@@ -165,12 +179,20 @@ func (r *ringBuffer) dumpBucket(bucket string) error {
 			return nil
 		})
 
-		// c := bu.Cursor()
-		//
-		// fmt.Println("--------------------------DUMP---------------------------")
-		// for k, v := c.First(); k != nil; c.Next() {
-		// 	fmt.Printf("k: %s, v: %v\n", k, v)
-		// }
+		return nil
+	})
+
+	return err
+}
+
+func (r *ringBuffer) deleteKeyFromBucket(bucket string, key string) error {
+	err := r.db.Update(func(tx *bolt.Tx) error {
+		bu := tx.Bucket([]byte(bucket))
+
+		err := bu.Delete([]byte(key))
+		if err != nil {
+			log.Printf("error: delete key in bucket %v failed: %v\n", bucket, err)
+		}
 
 		return nil
 	})