1 files changed, 96 insertions, 0 deletions
diff --git a/ntfy/watcher.go b/ntfy/watcher.go
new file mode 100644
index 0000000..2b3b5eb
--- /dev/null
+++ b/ntfy/watcher.go
@@ -0,0 +1,96 @@
+package ntfy
+
+import (
+	"bufio"
+	"encoding/json"
+	"log"
+	"net/http"
+	"net/url"
+	"path"
+	"time"
+)
+
+type Message struct {
+	Id      string `json:"id"`
+	Time    int    `json:"time"`
+	Message string `json:"message"`
+	Event   string `json:"event"`
+}
+
+type NtfyWatcher struct {
+	Endpoint string
+	Topics   []string
+}
+
+func (w *NtfyWatcher) Watch() chan Message {
+	notifications := make(chan Message)
+
+	for _, topic := range w.Topics {
+		log.Println("subscribing to topic:", topic)
+
+		go func() {
+			retryCount := 5
+			retryTime := 5 * time.Second
+			retries := retryCount
+
+			sleepAndDecrementRetry := func() {
+				log.Println("waiting 5 seconds before reconnecting. retries left:", retries, "topic:", topic, "endpoint:", w.Endpoint)
+				time.Sleep(retryTime)
+				retries--
+			}
+
+			for true {
+				if retries == 0 {
+					log.Fatal("too many retries, exiting")
+				}
+
+				endpoint, _ := url.JoinPath(w.Endpoint, path.Join(topic, "json"))
+				resp, err := http.Get(endpoint)
+				if err != nil {
+					log.Println("error connecting to endpoint:", err)
+					sleepAndDecrementRetry()
+					continue
+				}
+
+				defer resp.Body.Close()
+				scanner := bufio.NewScanner(resp.Body)
+				for scanner.Scan() {
+					bytes := scanner.Bytes()
+					var msg Message
+					err := json.Unmarshal(bytes, &msg)
+					if err != nil {
+						log.Println("could not unmarshal message:", err)
+						continue
+					}
+
+					if msg.Event == "keepalive" {
+						log.Println("received keepalive message")
+						continue
+					}
+					if msg.Event != "message" {
+						log.Println("received unknown event:", msg.Event)
+						continue
+					}
+
+					log.Println("received notification:", msg)
+					notifications <- msg
+					retries = retryCount // reset retries
+				}
+
+				if err := scanner.Err(); err != nil {
+					log.Println("error reading response body:", err)
+					sleepAndDecrementRetry()
+				}
+			}
+		}()
+	}
+
+	return notifications
+}
+
+func MakeNtfyWatcher(endpoint string, topics []string) *NtfyWatcher {
+	return &NtfyWatcher{
+		Endpoint: endpoint,
+		Topics:   topics,
+	}
+}