ntfy/client/client.go

292 lines
9.3 KiB
Go
Raw Normal View History

// Package client provides a ntfy client to publish and subscribe to topics
2021-12-17 02:33:01 +01:00
package client
import (
"bufio"
"context"
"encoding/json"
2022-02-02 05:39:57 +01:00
"errors"
2021-12-17 02:33:01 +01:00
"fmt"
"heckel.io/ntfy/log"
2021-12-21 21:22:27 +01:00
"heckel.io/ntfy/util"
"io"
2021-12-17 02:33:01 +01:00
"net/http"
"strings"
"sync"
"time"
)
// Event type constants
2021-12-17 02:33:01 +01:00
const (
2022-02-04 04:26:22 +01:00
MessageEvent = "message"
KeepaliveEvent = "keepalive"
OpenEvent = "open"
PollRequestEvent = "poll_request"
2021-12-17 02:33:01 +01:00
)
const (
maxResponseBytes = 4096
)
// Client is the ntfy client that can be used to publish and subscribe to ntfy topics
2021-12-17 02:33:01 +01:00
type Client struct {
Messages chan *Message
2021-12-18 20:43:27 +01:00
config *Config
2021-12-17 02:33:01 +01:00
subscriptions map[string]*subscription
mu sync.Mutex
}
// Message is a struct that represents a ntfy message
2021-12-22 23:20:43 +01:00
type Message struct { // TODO combine with server.message
2022-01-16 04:33:35 +01:00
ID string
Event string
Time int64
Topic string
Message string
Title string
Priority int
Tags []string
Click string
Attachment *Attachment
2021-12-21 21:22:27 +01:00
// Additional fields
TopicURL string
SubscriptionID string
Raw string
2021-12-17 02:33:01 +01:00
}
2022-01-16 05:53:40 +01:00
// Attachment represents a message attachment
2022-01-16 04:33:35 +01:00
type Attachment struct {
Name string `json:"name"`
Type string `json:"type,omitempty"`
Size int64 `json:"size,omitempty"`
Expires int64 `json:"expires,omitempty"`
URL string `json:"url"`
Owner string `json:"-"` // IP address of uploader, used for rate limiting
}
2021-12-17 02:33:01 +01:00
type subscription struct {
2021-12-21 21:22:27 +01:00
ID string
topicURL string
cancel context.CancelFunc
2021-12-17 02:33:01 +01:00
}
// New creates a new Client using a given Config
2021-12-18 20:43:27 +01:00
func New(config *Config) *Client {
2021-12-17 02:33:01 +01:00
return &Client{
2021-12-22 23:20:43 +01:00
Messages: make(chan *Message, 50), // Allow reading a few messages
2021-12-18 20:43:27 +01:00
config: config,
2021-12-17 02:33:01 +01:00
subscriptions: make(map[string]*subscription),
}
}
// Publish sends a message to a specific topic, optionally using options.
2022-01-13 03:24:48 +01:00
// See PublishReader for details.
func (c *Client) Publish(topic, message string, options ...PublishOption) (*Message, error) {
return c.PublishReader(topic, strings.NewReader(message), options...)
}
// PublishReader sends a message to a specific topic, optionally using options.
//
// A topic can be either a full URL (e.g. https://myhost.lan/mytopic), a short URL which is then prepended https://
// (e.g. myhost.lan -> https://myhost.lan), or a short name which is expanded using the default host in the
// config (e.g. mytopic -> https://ntfy.sh/mytopic).
//
// To pass title, priority and tags, check out WithTitle, WithPriority, WithTagsList, WithDelay, WithNoCache,
// WithNoFirebase, and the generic WithHeader.
2022-01-13 03:24:48 +01:00
func (c *Client) PublishReader(topic string, body io.Reader, options ...PublishOption) (*Message, error) {
2021-12-19 04:02:36 +01:00
topicURL := c.expandTopicURL(topic)
2022-01-13 03:24:48 +01:00
req, _ := http.NewRequest("POST", topicURL, body)
2021-12-17 02:33:01 +01:00
for _, option := range options {
if err := option(req); err != nil {
return nil, err
2021-12-17 02:33:01 +01:00
}
}
log.Debug("%s Publishing message with headers %s", util.ShortTopicURL(topicURL), req.Header)
2021-12-17 02:33:01 +01:00
resp, err := http.DefaultClient.Do(req)
if err != nil {
return nil, err
2021-12-17 02:33:01 +01:00
}
defer resp.Body.Close()
b, err := io.ReadAll(io.LimitReader(resp.Body, maxResponseBytes))
if err != nil {
return nil, err
}
2022-02-02 05:39:57 +01:00
if resp.StatusCode != http.StatusOK {
return nil, errors.New(strings.TrimSpace(string(b)))
}
2021-12-21 21:22:27 +01:00
m, err := toMessage(string(b), topicURL, "")
if err != nil {
return nil, err
}
return m, nil
2021-12-17 02:33:01 +01:00
}
// Poll queries a topic for all (or a limited set) of messages. Unlike Subscribe, this method only polls for
// messages and does not subscribe to messages that arrive after this call.
//
// A topic can be either a full URL (e.g. https://myhost.lan/mytopic), a short URL which is then prepended https://
// (e.g. myhost.lan -> https://myhost.lan), or a short name which is expanded using the default host in the
// config (e.g. mytopic -> https://ntfy.sh/mytopic).
//
// By default, all messages will be returned, but you can change this behavior using a SubscribeOption.
// See WithSince, WithSinceAll, WithSinceUnixTime, WithScheduled, and the generic WithQueryParam.
2021-12-18 20:43:27 +01:00
func (c *Client) Poll(topic string, options ...SubscribeOption) ([]*Message, error) {
2021-12-17 15:32:59 +01:00
ctx := context.Background()
messages := make([]*Message, 0)
msgChan := make(chan *Message)
errChan := make(chan error)
2021-12-18 20:43:27 +01:00
topicURL := c.expandTopicURL(topic)
log.Debug("%s Polling from topic", util.ShortTopicURL(topicURL))
2021-12-22 23:45:19 +01:00
options = append(options, WithPoll())
2021-12-17 15:32:59 +01:00
go func() {
2021-12-21 21:22:27 +01:00
err := performSubscribeRequest(ctx, msgChan, topicURL, "", options...)
2021-12-17 15:32:59 +01:00
close(msgChan)
errChan <- err
}()
for m := range msgChan {
messages = append(messages, m)
}
return messages, <-errChan
}
// Subscribe subscribes to a topic to listen for newly incoming messages. The method starts a connection in the
// background and returns new messages via the Messages channel.
//
// A topic can be either a full URL (e.g. https://myhost.lan/mytopic), a short URL which is then prepended https://
// (e.g. myhost.lan -> https://myhost.lan), or a short name which is expanded using the default host in the
// config (e.g. mytopic -> https://ntfy.sh/mytopic).
//
// By default, only new messages will be returned, but you can change this behavior using a SubscribeOption.
// See WithSince, WithSinceAll, WithSinceUnixTime, WithScheduled, and the generic WithQueryParam.
//
2021-12-21 21:22:27 +01:00
// The method returns a unique subscriptionID that can be used in Unsubscribe.
//
// Example:
// c := client.New(client.NewConfig())
2021-12-21 21:22:27 +01:00
// subscriptionID := c.Subscribe("mytopic")
// for m := range c.Messages {
// fmt.Printf("New message: %s", m.Message)
// }
2021-12-18 20:43:27 +01:00
func (c *Client) Subscribe(topic string, options ...SubscribeOption) string {
2021-12-17 02:33:01 +01:00
c.mu.Lock()
defer c.mu.Unlock()
2021-12-21 21:22:27 +01:00
subscriptionID := util.RandomString(10)
2021-12-18 20:43:27 +01:00
topicURL := c.expandTopicURL(topic)
log.Debug("%s Subscribing to topic", util.ShortTopicURL(topicURL))
2021-12-17 02:33:01 +01:00
ctx, cancel := context.WithCancel(context.Background())
2021-12-21 21:22:27 +01:00
c.subscriptions[subscriptionID] = &subscription{
ID: subscriptionID,
topicURL: topicURL,
cancel: cancel,
}
go handleSubscribeConnLoop(ctx, c.Messages, topicURL, subscriptionID, options...)
return subscriptionID
2021-12-17 02:33:01 +01:00
}
2021-12-21 21:22:27 +01:00
// Unsubscribe unsubscribes from a topic that has been previously subscribed to using the unique
// subscriptionID returned in Subscribe.
func (c *Client) Unsubscribe(subscriptionID string) {
c.mu.Lock()
defer c.mu.Unlock()
sub, ok := c.subscriptions[subscriptionID]
if !ok {
return
}
delete(c.subscriptions, subscriptionID)
sub.cancel()
}
// UnsubscribeAll unsubscribes from a topic that has been previously subscribed with Subscribe.
// If there are multiple subscriptions matching the topic, all of them are unsubscribed from.
//
// A topic can be either a full URL (e.g. https://myhost.lan/mytopic), a short URL which is then prepended https://
// (e.g. myhost.lan -> https://myhost.lan), or a short name which is expanded using the default host in the
// config (e.g. mytopic -> https://ntfy.sh/mytopic).
2021-12-21 21:22:27 +01:00
func (c *Client) UnsubscribeAll(topic string) {
2021-12-17 02:33:01 +01:00
c.mu.Lock()
defer c.mu.Unlock()
2021-12-18 20:43:27 +01:00
topicURL := c.expandTopicURL(topic)
2021-12-21 21:22:27 +01:00
for _, sub := range c.subscriptions {
if sub.topicURL == topicURL {
delete(c.subscriptions, sub.ID)
sub.cancel()
}
2021-12-17 02:33:01 +01:00
}
}
2021-12-18 20:43:27 +01:00
func (c *Client) expandTopicURL(topic string) string {
if strings.HasPrefix(topic, "http://") || strings.HasPrefix(topic, "https://") {
return topic
} else if strings.Contains(topic, "/") {
return fmt.Sprintf("https://%s", topic)
}
return fmt.Sprintf("%s/%s", c.config.DefaultHost, topic)
}
2021-12-21 21:22:27 +01:00
func handleSubscribeConnLoop(ctx context.Context, msgChan chan *Message, topicURL, subcriptionID string, options ...SubscribeOption) {
2021-12-17 02:33:01 +01:00
for {
2021-12-21 21:22:27 +01:00
// TODO The retry logic is crude and may lose messages. It should record the last message like the
// Android client, use since=, and do incremental backoff too
if err := performSubscribeRequest(ctx, msgChan, topicURL, subcriptionID, options...); err != nil {
log.Warn("%s Connection failed: %s", util.ShortTopicURL(topicURL), err.Error())
2021-12-17 02:33:01 +01:00
}
select {
case <-ctx.Done():
2022-06-02 20:38:38 +02:00
log.Info("%s Connection exited", util.ShortTopicURL(topicURL))
2021-12-17 02:33:01 +01:00
return
case <-time.After(10 * time.Second): // TODO Add incremental backoff
2021-12-17 02:33:01 +01:00
}
}
}
2021-12-21 21:22:27 +01:00
func performSubscribeRequest(ctx context.Context, msgChan chan *Message, topicURL string, subscriptionID string, options ...SubscribeOption) error {
streamURL := fmt.Sprintf("%s/json", topicURL)
log.Debug("%s Listening to %s", util.ShortTopicURL(topicURL), streamURL)
req, err := http.NewRequestWithContext(ctx, http.MethodGet, streamURL, nil)
2021-12-17 02:33:01 +01:00
if err != nil {
return err
}
2021-12-17 15:32:59 +01:00
for _, option := range options {
if err := option(req); err != nil {
return err
}
}
2021-12-17 02:33:01 +01:00
resp, err := http.DefaultClient.Do(req)
if err != nil {
return err
}
defer resp.Body.Close()
2022-02-04 04:26:22 +01:00
if resp.StatusCode != http.StatusOK {
b, err := io.ReadAll(io.LimitReader(resp.Body, maxResponseBytes))
if err != nil {
return err
}
return errors.New(strings.TrimSpace(string(b)))
}
2021-12-17 02:33:01 +01:00
scanner := bufio.NewScanner(resp.Body)
for scanner.Scan() {
messageJSON := scanner.Text()
m, err := toMessage(messageJSON, topicURL, subscriptionID)
if err != nil {
2021-12-17 02:33:01 +01:00
return err
}
log.Trace("%s Message received: %s", util.ShortTopicURL(topicURL), messageJSON)
2021-12-22 23:20:43 +01:00
if m.Event == MessageEvent {
msgChan <- m
}
2021-12-17 02:33:01 +01:00
}
return nil
}
2021-12-21 21:22:27 +01:00
func toMessage(s, topicURL, subscriptionID string) (*Message, error) {
var m *Message
if err := json.NewDecoder(strings.NewReader(s)).Decode(&m); err != nil {
return nil, err
}
m.TopicURL = topicURL
2021-12-21 21:22:27 +01:00
m.SubscriptionID = subscriptionID
m.Raw = s
return m, nil
}