renabled ringbuffer done channel triggered by persiststore flag

process.go

@@ -1005,13 +1005,15 @@ func (p process) publishAMessage(m Message, zEnc *zstd.Encoder, once sync.Once,
 	// sending of the message.
 	p.messageDeliverNats(natsMsgPayloadCompressed, natsMsgHeader, natsConn, m)
 
-	// select {
+	if p.configuration.RingBufferPersistStore {
-	// case m.done <- struct{}{}:
+		select {
-	// 	// Signaling back to the ringbuffer that we are done with the
+		case m.done <- struct{}{}:
-	// 	// current message, and it can remove it from the ringbuffer.
+			// Signaling back to the ringbuffer that we are done with the
-	// case <-p.ctx.Done():
+			// current message, and it can remove it from the ringbuffer.
-	// 	return
+		case <-p.ctx.Done():
-	// }
+			return
+		}
+	}
 
 	// Increment the counter for the next message to be sent.
 	p.messageID++

ringbuffer.go

@@ -281,6 +281,12 @@ func (r *ringBuffer) processBufferMessages(ctx context.Context, ringBufferOutCh
 					},
 				}
 
+				// Create a ticker that will kick in when a message have been in the
+				// system for it's maximum time. This will allow us to continue, and
+				// remove the message if it takes longer than it should to get delivered.
+				ticker := time.NewTicker(time.Duration(v.SAM.ACKTimeout) * time.Duration(v.SAM.Retries) * time.Second)
+				defer ticker.Stop()
+
 				ringBufferOutCh <- sd
 				// Just to confirm here that the message was picked up, to know if the
 				// the read process have stalled or not.
@@ -305,14 +311,16 @@ func (r *ringBuffer) processBufferMessages(ctx context.Context, ringBufferOutCh
 				// amount of time a message should be allowed to be using for getting published so
 				// we don't get stuck go routines here.
 				//
-				// TODO: Figure out why what the reason for not receceving the done signals might be.
+				if r.configuration.RingBufferPersistStore {
-				// select {
+					select {
-				// case <-v.SAM.done:
+					case <-v.SAM.done:
-				// case <-ticker.C:
+						fmt.Printf("---\n DONE with\n---\n")
-				// 	log.Printf("----------------------------------------------\n")
+					case <-ticker.C:
-				// 	log.Printf("Error: ringBuffer message id: %v, subject: %v seems to be stuck, did not receive done signal from publishAMessage process, exited on ticker\n", v.SAM.ID, v.SAM.Subject)
+						log.Printf("----------------------------------------------\n")
-				// 	log.Printf("----------------------------------------------\n")
+						log.Printf("Error: ringBuffer message id: %v, subject: %v seems to be stuck, did not receive done signal from publishAMessage process, exited on ticker\n", v.SAM.ID, v.SAM.Subject)
-				// }
+						log.Printf("----------------------------------------------\n")
+					}
+				}
 				// log.Printf("info: processBufferMessages: done with message, deleting key from bucket, %v\n", v.ID)
 				r.metrics.promMessagesProcessedIDLast.Set(float64(v.ID))
 

scripts/clifolder.sh

@@ -33,9 +33,9 @@ function sendMessage() {
             ],
         "replyMethod": "REQToFileAppend",
         "retryWait": 5,
-        "ACKTimeout": 10,
+        "ACKTimeout": 30,
         "retries": 1,
-        "replyACKTimeout": 10,
+        "replyACKTimeout": 30,
         "replyRetries": 1,
         "methodTimeout": 10,
         "replyMethodTimeout": 10,