storing to message kv buffer seems to work

example-inmessage/orig-ship2.json

@@ -2,7 +2,7 @@
     {
         "toNode": "ship2",
         "data": ["bash","-c","tree ../"],
-        "commandOrEvent":"Command",
+        "commandOrEvent":"command",
         "method":"shellCommand"
 
     }

publisher.go

@@ -26,6 +26,10 @@ type Message struct {
 	// method, what is this message doing, etc. shellCommand, syslog, etc.
 	Method   Method `json:"method" yaml:"method"`
 	FromNode node
+	// done is used to signal when a message is fully processed.
+	// This is used when choosing when to move the message from
+	// the ringbuffer into the time series log.
+	done chan struct{}
 }
 
 // server is the structure that will hold the state about spawned
@@ -118,7 +122,7 @@ func (s *server) Start() {
 	time.Sleep(time.Second * 2)
 	s.printProcessesMap()
 
-	s.handleMessagesToPublish()
+	s.handleMessagesInRingbuffer()
 
 	select {}
 
@@ -135,7 +139,7 @@ func (s *server) printProcessesMap() {
 
 // handleNewOperatorMessages will handle all the new operator messages
 // given to the system, and route them to the correct subject queue.
-func (s *server) handleMessagesToPublish() {
+func (s *server) handleMessagesInRingbuffer() {
 	// Prepare and start a new ring buffer
 	const bufferSize int = 100
 	rb := newringBuffer(bufferSize)
@@ -147,6 +151,9 @@ func (s *server) handleMessagesToPublish() {
 	// pipe requested by operator, and fill them into the buffer.
 	go func() {
 		for samSlice := range s.inputFromFileCh {
+			fmt.Println("----------------------DEBUG1--------------------------------")
+			fmt.Printf("DEBUG!!!!!!!!!!!!!!\n")
+			fmt.Println("----------------------DEBUG1END-----------------------------")
 			for _, sam := range samSlice {
 				inCh <- sam
 			}
@@ -307,6 +314,7 @@ func (s *server) processSpawnWorker(proc process) {
 			m := <-proc.subject.messageCh
 			m.ID = s.processes[proc.subject.name()].messageID
 			messageDeliver(proc, m, s.natsConn)
+			m.done <- struct{}{}
 
 			// Increment the counter for the next message to be sent.
 			proc.messageID++

ringbuffer.go

@@ -98,9 +98,23 @@ func (r *ringBuffer) start(inCh chan subjectAndMessage, outCh chan subjectAndMes
 
 	// Empty the buffer when data asked for
 	go func() {
+		// 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
+			// spawnProcess function will handle incomming messages sequentaly.
+			// So in the spawnProcess function we put a struct{} value when a
+			// message is processed on the "done" channel and an ack is received
+			// for a message, and we wait here for the "done" to be received.
+
+			v.Message.done = make(chan struct{})
 			outCh <- v
 
+			// ----------TESTING
+			<-v.done
+			fmt.Println("-----------------------------------------------------------")
+			fmt.Printf("### DONE WITH THE MESSAGE\n")
+			fmt.Println("-----------------------------------------------------------")
+
 			// 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.
@@ -110,7 +124,7 @@ func (r *ringBuffer) start(inCh chan subjectAndMessage, outCh chan subjectAndMes
 			// been processed, and that we then can delete it out of the K/V Store.
 
 			// Dump the whole KV store
-			err := r.dumpCursor(samValueBucket)
+			err := r.dumpBucket(samValueBucket)
 			if err != nil {
 				fmt.Printf("* Error: dump of db failed: %v\n", err)
 			}
@@ -121,7 +135,7 @@ func (r *ringBuffer) start(inCh chan subjectAndMessage, outCh chan subjectAndMes
 	}()
 }
 
-func (r *ringBuffer) dumpCursor(bucket string) error {
+func (r *ringBuffer) dumpBucket(bucket string) error {
 	err := r.db.View(func(tx *bolt.Tx) error {
 		bu := tx.Bucket([]byte(bucket))
 

textlogging.log

@@ -17,3 +17,5 @@ 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
+some message sent from a ship for testing