Compare commits

...

5 Commits

Author SHA1 Message Date
Philipp Heckel
ffe0c72a5a Bump version 2021-11-16 13:11:31 -05:00
Philipp Heckel
52136030be Subscribe to more than one topic 2021-11-15 07:56:58 -05:00
Philipp Heckel
a481f4c448 Theme color in Chrome Android 2021-11-12 14:43:32 -05:00
Philipp Heckel
9b171dee8b Merge branch 'main' of github.com:binwiederhier/ntfy into main 2021-11-09 14:48:48 -05:00
Philipp Heckel
c0ee174b13 Make publishing asynchronous 2021-11-09 14:48:25 -05:00
4 changed files with 71 additions and 42 deletions

View File

@@ -137,13 +137,13 @@ sudo apt install ntfy
**Debian/Ubuntu** (*manual install*)**:**
```bash
sudo apt install tmux
wget https://github.com/binwiederhier/ntfy/releases/download/v1.3.0/ntfy_1.3.0_amd64.deb
dpkg -i ntfy_1.3.0_amd64.deb
wget https://github.com/binwiederhier/ntfy/releases/download/v1.4.0/ntfy_1.3.0_amd64.deb
dpkg -i ntfy_1.4.0_amd64.deb
```
**Fedora/RHEL/CentOS:**
```bash
rpm -ivh https://github.com/binwiederhier/ntfy/releases/download/v1.3.0/ntfy_1.3.0_amd64.rpm
rpm -ivh https://github.com/binwiederhier/ntfy/releases/download/v1.4.0/ntfy_1.3.0_amd64.rpm
```
**Docker:**
@@ -158,8 +158,8 @@ go get -u heckel.io/ntfy
**Manual install** (*any x86_64-based Linux*)**:**
```bash
wget https://github.com/binwiederhier/ntfy/releases/download/v1.3.0/ntfy_1.3.0_linux_x86_64.tar.gz
sudo tar -C /usr/bin -zxf ntfy_1.3.0_linux_x86_64.tar.gz ntfy
wget https://github.com/binwiederhier/ntfy/releases/download/v1.4.0/ntfy_1.3.0_linux_x86_64.tar.gz
sudo tar -C /usr/bin -zxf ntfy_1.4.0_linux_x86_64.tar.gz ntfy
./ntfy
```

View File

@@ -13,9 +13,9 @@
<meta name="HandheldFriendly" content="true">
<!-- Mobile browsers, background color -->
<meta name="theme-color" content="#39005a">
<meta name="msapplication-navbutton-color" content="#39005a">
<meta name="apple-mobile-web-app-status-bar-style" content="#39005a">
<meta name="theme-color" content="#317f6f">
<meta name="msapplication-navbutton-color" content="#317f6f">
<meta name="apple-mobile-web-app-status-bar-style" content="#317f6f">
<!-- Favicon, see favicon.io -->
<link rel="icon" type="image/png" href="static/img/favicon.png">

View File

@@ -78,9 +78,9 @@ const (
var (
topicRegex = regexp.MustCompile(`^/[-_A-Za-z0-9]{1,64}$`) // Regex must match JS & Android app!
jsonRegex = regexp.MustCompile(`^/[-_A-Za-z0-9]{1,64}/json$`)
sseRegex = regexp.MustCompile(`^/[-_A-Za-z0-9]{1,64}/sse$`)
rawRegex = regexp.MustCompile(`^/[-_A-Za-z0-9]{1,64}/raw$`)
jsonRegex = regexp.MustCompile(`^/[-_A-Za-z0-9]{1,64}(,[-_A-Za-z0-9]{1,64})*/json$`)
sseRegex = regexp.MustCompile(`^/[-_A-Za-z0-9]{1,64}(,[-_A-Za-z0-9]{1,64})*/sse$`)
rawRegex = regexp.MustCompile(`^/[-_A-Za-z0-9]{1,64}(,[-_A-Za-z0-9]{1,64})*/raw$`)
staticRegex = regexp.MustCompile(`^/static/.+`)
@@ -223,7 +223,7 @@ func (s *Server) handleStatic(w http.ResponseWriter, r *http.Request) error {
}
func (s *Server) handlePublish(w http.ResponseWriter, r *http.Request, v *visitor) error {
t, err := s.topic(r.URL.Path[1:])
t, err := s.topicFromID(r.URL.Path[1:])
if err != nil {
return err
}
@@ -289,7 +289,9 @@ func (s *Server) handleSubscribe(w http.ResponseWriter, r *http.Request, v *visi
return errHTTPTooManyRequests
}
defer v.RemoveSubscription()
t, err := s.topic(strings.TrimSuffix(r.URL.Path[1:], "/"+format)) // Hack
topicsStr := strings.TrimSuffix(r.URL.Path[1:], "/"+format) // Hack
topicIDs := strings.Split(topicsStr, ",")
topics, err := s.topicsFromIDs(topicIDs...)
if err != nil {
return err
}
@@ -314,14 +316,21 @@ func (s *Server) handleSubscribe(w http.ResponseWriter, r *http.Request, v *visi
w.Header().Set("Access-Control-Allow-Origin", "*") // CORS, allow cross-origin requests
w.Header().Set("Content-Type", contentType+"; charset=utf-8") // Android/Volley client needs charset!
if poll {
return s.sendOldMessages(t, since, sub)
return s.sendOldMessages(topics, since, sub)
}
subscriberID := t.Subscribe(sub)
defer t.Unsubscribe(subscriberID)
if err := sub(newOpenMessage(t.id)); err != nil { // Send out open message
subscriberIDs := make([]int, 0)
for _, t := range topics {
subscriberIDs = append(subscriberIDs, t.Subscribe(sub))
}
defer func() {
for i, subscriberID := range subscriberIDs {
topics[i].Unsubscribe(subscriberID) // Order!
}
}()
if err := sub(newOpenMessage(topicsStr)); err != nil { // Send out open message
return err
}
if err := s.sendOldMessages(t, since, sub); err != nil {
if err := s.sendOldMessages(topics, since, sub); err != nil {
return err
}
for {
@@ -330,25 +339,27 @@ func (s *Server) handleSubscribe(w http.ResponseWriter, r *http.Request, v *visi
return nil
case <-time.After(s.config.KeepaliveInterval):
v.Keepalive()
if err := sub(newKeepaliveMessage(t.id)); err != nil { // Send keepalive message
if err := sub(newKeepaliveMessage(topicsStr)); err != nil { // Send keepalive message
return err
}
}
}
}
func (s *Server) sendOldMessages(t *topic, since sinceTime, sub subscriber) error {
func (s *Server) sendOldMessages(topics []*topic, since sinceTime, sub subscriber) error {
if since.IsNone() {
return nil
}
messages, err := s.cache.Messages(t.id, since)
if err != nil {
return err
}
for _, m := range messages {
if err := sub(m); err != nil {
for _, t := range topics {
messages, err := s.cache.Messages(t.id, since)
if err != nil {
return err
}
for _, m := range messages {
if err := sub(m); err != nil {
return err
}
}
}
return nil
}
@@ -382,19 +393,31 @@ func (s *Server) handleOptions(w http.ResponseWriter, r *http.Request) error {
return nil
}
func (s *Server) topic(id string) (*topic, error) {
func (s *Server) topicFromID(id string) (*topic, error) {
topics, err := s.topicsFromIDs(id)
if err != nil {
return nil, err
}
return topics[0], nil
}
func (s *Server) topicsFromIDs(ids... string) ([]*topic, error) {
s.mu.Lock()
defer s.mu.Unlock()
if _, ok := s.topics[id]; !ok {
if len(s.topics) >= s.config.GlobalTopicLimit {
return nil, errHTTPTooManyRequests
}
s.topics[id] = newTopic(id, time.Now())
if s.firebase != nil {
s.topics[id].Subscribe(s.firebase)
topics := make([]*topic, 0)
for _, id := range ids {
if _, ok := s.topics[id]; !ok {
if len(s.topics) >= s.config.GlobalTopicLimit {
return nil, errHTTPTooManyRequests
}
s.topics[id] = newTopic(id, time.Now())
if s.firebase != nil {
s.topics[id].Subscribe(s.firebase)
}
}
topics = append(topics, s.topics[id])
}
return s.topics[id], nil
return topics, nil
}
func (s *Server) updateStatsAndExpire() {

View File

@@ -28,6 +28,7 @@ func newTopic(id string, last time.Time) *topic {
}
}
// Subscribe subscribes to this topic
func (t *topic) Subscribe(s subscriber) int {
t.mu.Lock()
defer t.mu.Unlock()
@@ -37,24 +38,29 @@ func (t *topic) Subscribe(s subscriber) int {
return subscriberID
}
// Unsubscribe removes the subscription from the list of subscribers
func (t *topic) Unsubscribe(id int) {
t.mu.Lock()
defer t.mu.Unlock()
delete(t.subscribers, id)
}
// Publish asynchronously publishes to all subscribers
func (t *topic) Publish(m *message) error {
t.mu.Lock()
defer t.mu.Unlock()
t.last = time.Now()
for _, s := range t.subscribers {
if err := s(m); err != nil {
log.Printf("error publishing message to subscriber")
go func() {
t.mu.Lock()
defer t.mu.Unlock()
t.last = time.Now()
for _, s := range t.subscribers {
if err := s(m); err != nil {
log.Printf("error publishing message to subscriber")
}
}
}
}()
return nil
}
// Subscribers returns the number of subscribers to this topic
func (t *topic) Subscribers() int {
t.mu.Lock()
defer t.mu.Unlock()