Wait with timeout

ringbuffer/ringbuffer.go

@@ -4,7 +4,6 @@ import (
 	"sync"
 )
 
-
 type RingBuffer struct {
 	buffer []string
 	size   int
@@ -13,7 +12,6 @@ type RingBuffer struct {
 	mu     sync.Mutex
 }
 
-
 func NewRingBuffer(size int) *RingBuffer {
 	return &RingBuffer{
 		buffer: make([]string, size),
@@ -23,7 +21,6 @@ func NewRingBuffer(size int) *RingBuffer {
 	}
 }
 
-
 func (rb *RingBuffer) Add(item string) {
 	rb.mu.Lock()
 	defer rb.mu.Unlock()
@@ -40,7 +37,6 @@ func (rb *RingBuffer) Add(item string) {
 	}
 }
 
-
 func (rb *RingBuffer) GetLast() (string, int) {
 	rb.mu.Lock()
 	defer rb.mu.Unlock()
@@ -52,19 +48,28 @@ func (rb *RingBuffer) GetLast() (string, int) {
 	return rb.buffer[rb.tail], rb.tail
 }
 
-
 func (rb *RingBuffer) WaitForNewItem() string {
-	var newItem string
+	rb.mu.Lock()
+	defer rb.mu.Unlock()
 
-	for {
-		rb.mu.Lock()
-		if rb.head != -1 && rb.tail != -1 && rb.buffer[rb.tail] != newItem {
-			newItem = rb.buffer[rb.tail]
-			rb.mu.Unlock()
-			break
-		}
-		rb.mu.Unlock()
+	if rb.tail == -1 {
+		return ""
 	}
-	return newItem
+
+	return rb.buffer[rb.tail]
+}
+
+func (rb *RingBuffer) IsEmpty() bool {
+	rb.mu.Lock()
+	defer rb.mu.Unlock()
+
+	return rb.head == -1
+}
+
+func (rb *RingBuffer) GetLastPosition() int {
+	rb.mu.Lock()
+	defer rb.mu.Unlock()
+
+	return rb.tail
 }
 

smshook/smshook.go

@@ -4,23 +4,21 @@ import (
 	"encoding/json"
 	"net/http"
 	"sync"
+	"time"
 
 	"SmsHook/ringbuffer"
 )
 
-
 type WebhookServer struct {
 	buffer *ringbuffer.RingBuffer
 	cond   *sync.Cond
 	mux    *http.ServeMux
 }
 
-
 type WebhookPayload struct {
 	Content string `json:"content"`
 }
 
-
 func NewWebhookServer(bufferSize int) *WebhookServer {
 	mutex := sync.Mutex{}
 	server := &WebhookServer{
@@ -33,7 +31,6 @@ func NewWebhookServer(bufferSize int) *WebhookServer {
 	return server
 }
 
-
 func (s *WebhookServer) webhookHandler(w http.ResponseWriter, r *http.Request) {
 	var payload WebhookPayload
 
@@ -47,12 +44,10 @@ func (s *WebhookServer) webhookHandler(w http.ResponseWriter, r *http.Request) {
 	s.cond.Broadcast()
 }
 
-
 func (s *WebhookServer) GetLastItem() (string, int) {
 	return s.buffer.GetLast()
 }
 
-
 func (s *WebhookServer) WaitForNewItem() string {
 	s.cond.L.Lock()
 	s.cond.Wait()
@@ -61,30 +56,54 @@ func (s *WebhookServer) WaitForNewItem() string {
 	return newItem
 }
 
+func (s *WebhookServer) WaitForNewItemWithTimeout(timeout time.Duration) (string, bool) {
+	s.cond.L.Lock()
+	defer s.cond.L.Unlock()
+
+	initialPos := s.buffer.GetLastPosition()
+
+	timedOut := false
+	timer := time.AfterFunc(timeout, func() {
+		s.cond.Broadcast()
+		timedOut = true
+	})
+
+	for s.buffer.GetLastPosition() == initialPos {
+		if timedOut {
+			return "", false
+		}
+		s.cond.Wait()
+	}
+
+	timer.Stop()
+	newItem := s.buffer.WaitForNewItem()
+	return newItem, true
+}
 
 func (s *WebhookServer) Start(address string) error {
 	return http.ListenAndServe(address, s.mux)
 }
 
-var server *WebhookServer
-
+var Server *WebhookServer
 
 func Init(bufferSize int, address string) {
-	server = NewWebhookServer(bufferSize)
+	Server = NewWebhookServer(bufferSize)
 	go func() {
-		if err := server.Start(address); err != nil {
+		if err := Server.Start(address); err != nil {
 			panic(err)
 		}
 	}()
 }
 
-
 func GetLast() (string, int) {
-	return server.GetLastItem()
+	return Server.GetLastItem()
 }
 
-
 func WaitForNew() string {
-	return server.WaitForNewItem()
+	return Server.WaitForNewItem()
+}
+
+func WaitForNewWithTimeout(timeout time.Duration) (string, bool) {
+	return Server.WaitForNewItemWithTimeout(timeout)
 }