mirror of
https://github.com/binwiederhier/ntfy.git
synced 2026-01-19 00:27:25 +01:00
Compare commits
9 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
17839e634c | ||
|
7b810acfb5 | ||
|
|
1c7695c1f3 | ||
|
|
0e9fa1c4dc | ||
|
|
67922b0ae5 | ||
|
|
b775e6dfce | ||
|
|
fa7a45902f | ||
|
|
5f2bb4f876 | ||
|
|
91541f9c69 |
10
README.md
10
README.md
@@ -64,13 +64,13 @@ sudo apt install ntfy
|
||||
**Debian/Ubuntu** (*manual install*)**:**
|
||||
```bash
|
||||
sudo apt install tmux
|
||||
wget https://github.com/binwiederhier/ntfy/releases/download/v1.0.0/ntfy_1.0.0_amd64.deb
|
||||
dpkg -i ntfy_1.0.0_amd64.deb
|
||||
wget https://github.com/binwiederhier/ntfy/releases/download/v1.1.2/ntfy_1.1.2_amd64.deb
|
||||
dpkg -i ntfy_1.1.2_amd64.deb
|
||||
```
|
||||
|
||||
**Fedora/RHEL/CentOS:**
|
||||
```bash
|
||||
rpm -ivh https://github.com/binwiederhier/ntfy/releases/download/v1.0.0/ntfy_1.0.0_amd64.rpm
|
||||
rpm -ivh https://github.com/binwiederhier/ntfy/releases/download/v1.1.2/ntfy_1.1.2_amd64.rpm
|
||||
```
|
||||
|
||||
**Docker:**
|
||||
@@ -85,8 +85,8 @@ go get -u heckel.io/ntfy
|
||||
|
||||
**Manual install** (*any x86_64-based Linux*)**:**
|
||||
```bash
|
||||
wget https://github.com/binwiederhier/ntfy/releases/download/v1.0.0/ntfy_1.0.0_linux_x86_64.tar.gz
|
||||
sudo tar -C /usr/bin -zxf ntfy_1.0.0_linux_x86_64.tar.gz ntfy
|
||||
wget https://github.com/binwiederhier/ntfy/releases/download/v1.1.2/ntfy_1.1.2_linux_x86_64.tar.gz
|
||||
sudo tar -C /usr/bin -zxf ntfy_1.1.2_linux_x86_64.tar.gz ntfy
|
||||
./ntfy
|
||||
```
|
||||
|
||||
|
||||
@@ -19,6 +19,7 @@ func New() *cli.App {
|
||||
flags := []cli.Flag{
|
||||
&cli.StringFlag{Name: "config", Aliases: []string{"c"}, EnvVars: []string{"NTFY_CONFIG_FILE"}, Value: "/etc/ntfy/config.yml", DefaultText: "/etc/ntfy/config.yml", Usage: "config file"},
|
||||
altsrc.NewStringFlag(&cli.StringFlag{Name: "listen-http", Aliases: []string{"l"}, EnvVars: []string{"NTFY_LISTEN_HTTP"}, Value: config.DefaultListenHTTP, Usage: "ip:port used to as listen address"}),
|
||||
altsrc.NewStringFlag(&cli.StringFlag{Name: "cache-file", Aliases: []string{"C"}, EnvVars: []string{"NTFY_CACHE_FILE"}, Usage: "cache file used for message caching"}),
|
||||
altsrc.NewStringFlag(&cli.StringFlag{Name: "firebase-key-file", Aliases: []string{"F"}, EnvVars: []string{"NTFY_FIREBASE_KEY_FILE"}, Usage: "Firebase credentials file; if set additionally publish to FCM topic"}),
|
||||
altsrc.NewDurationFlag(&cli.DurationFlag{Name: "message-buffer-duration", Aliases: []string{"b"}, EnvVars: []string{"NTFY_MESSAGE_BUFFER_DURATION"}, Value: config.DefaultMessageBufferDuration, Usage: "buffer messages in memory for this time to allow `since` requests"}),
|
||||
altsrc.NewDurationFlag(&cli.DurationFlag{Name: "keepalive-interval", Aliases: []string{"k"}, EnvVars: []string{"NTFY_KEEPALIVE_INTERVAL"}, Value: config.DefaultKeepaliveInterval, Usage: "default interval of keepalive messages"}),
|
||||
@@ -44,6 +45,7 @@ func New() *cli.App {
|
||||
func execRun(c *cli.Context) error {
|
||||
// Read all the options
|
||||
listenHTTP := c.String("listen-http")
|
||||
cacheFile := c.String("cache-file")
|
||||
firebaseKeyFile := c.String("firebase-key-file")
|
||||
messageBufferDuration := c.Duration("message-buffer-duration")
|
||||
keepaliveInterval := c.Duration("keepalive-interval")
|
||||
@@ -62,6 +64,7 @@ func execRun(c *cli.Context) error {
|
||||
|
||||
// Run server
|
||||
conf := config.New(listenHTTP)
|
||||
conf.CacheFile = cacheFile
|
||||
conf.FirebaseKeyFile = firebaseKeyFile
|
||||
conf.MessageBufferDuration = messageBufferDuration
|
||||
conf.KeepaliveInterval = keepaliveInterval
|
||||
|
||||
@@ -14,33 +14,43 @@ const (
|
||||
DefaultManagerInterval = time.Minute
|
||||
)
|
||||
|
||||
// Defines the max number of requests, here:
|
||||
// 50 requests bucket, replenished at a rate of 1 per second
|
||||
// Defines all the limits
|
||||
// - request limit: max number of PUT/GET/.. requests (here: 50 requests bucket, replenished at a rate of 1 per second)
|
||||
// - global topic limit: max number of topics overall
|
||||
// - subscription limit: max number of subscriptions (active HTTP connections) per per-visitor/IP
|
||||
var (
|
||||
defaultLimit = rate.Every(time.Second)
|
||||
defaultLimitBurst = 50
|
||||
defaultGlobalTopicLimit = 5000
|
||||
defaultVisitorRequestLimit = rate.Every(time.Second)
|
||||
defaultVisitorRequestLimitBurst = 50
|
||||
defaultVisitorSubscriptionLimit = 30
|
||||
)
|
||||
|
||||
// Config is the main config struct for the application. Use New to instantiate a default config struct.
|
||||
type Config struct {
|
||||
ListenHTTP string
|
||||
FirebaseKeyFile string
|
||||
MessageBufferDuration time.Duration
|
||||
KeepaliveInterval time.Duration
|
||||
ManagerInterval time.Duration
|
||||
Limit rate.Limit
|
||||
LimitBurst int
|
||||
ListenHTTP string
|
||||
CacheFile string
|
||||
FirebaseKeyFile string
|
||||
MessageBufferDuration time.Duration
|
||||
KeepaliveInterval time.Duration
|
||||
ManagerInterval time.Duration
|
||||
GlobalTopicLimit int
|
||||
VisitorRequestLimit rate.Limit
|
||||
VisitorRequestLimitBurst int
|
||||
VisitorSubscriptionLimit int
|
||||
}
|
||||
|
||||
// New instantiates a default new config
|
||||
func New(listenHTTP string) *Config {
|
||||
return &Config{
|
||||
ListenHTTP: listenHTTP,
|
||||
FirebaseKeyFile: "",
|
||||
MessageBufferDuration: DefaultMessageBufferDuration,
|
||||
KeepaliveInterval: DefaultKeepaliveInterval,
|
||||
ManagerInterval: DefaultManagerInterval,
|
||||
Limit: defaultLimit,
|
||||
LimitBurst: defaultLimitBurst,
|
||||
ListenHTTP: listenHTTP,
|
||||
CacheFile: "",
|
||||
FirebaseKeyFile: "",
|
||||
MessageBufferDuration: DefaultMessageBufferDuration,
|
||||
KeepaliveInterval: DefaultKeepaliveInterval,
|
||||
ManagerInterval: DefaultManagerInterval,
|
||||
GlobalTopicLimit: defaultGlobalTopicLimit,
|
||||
VisitorRequestLimit: defaultVisitorRequestLimit,
|
||||
VisitorRequestLimitBurst: defaultVisitorRequestLimitBurst,
|
||||
VisitorSubscriptionLimit: defaultVisitorSubscriptionLimit,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -3,5 +3,5 @@
|
||||
# a simple script. The notify-send command sends any arriving message as a desktop notification.
|
||||
|
||||
while read msg; do
|
||||
notify-send "$msg"
|
||||
[ -n "$msg" ] && notify-send "$msg"
|
||||
done < <(stdbuf -i0 -o0 curl -s ntfy.sh/mytopic/raw)
|
||||
|
||||
30
go.mod
30
go.mod
@@ -1,16 +1,42 @@
|
||||
module heckel.io/ntfy
|
||||
|
||||
go 1.16
|
||||
go 1.17
|
||||
|
||||
require (
|
||||
cloud.google.com/go/firestore v1.6.0 // indirect
|
||||
cloud.google.com/go/firestore v1.6.1 // indirect
|
||||
cloud.google.com/go/storage v1.18.2 // indirect
|
||||
firebase.google.com/go v3.13.0+incompatible
|
||||
github.com/BurntSushi/toml v0.4.1 // indirect
|
||||
github.com/cpuguy83/go-md2man/v2 v2.0.1 // indirect
|
||||
github.com/mattn/go-sqlite3 v1.14.9
|
||||
github.com/urfave/cli/v2 v2.3.0
|
||||
golang.org/x/oauth2 v0.0.0-20211028175245-ba495a64dcb5 // indirect
|
||||
golang.org/x/time v0.0.0-20210723032227-1f47c861a9ac
|
||||
google.golang.org/api v0.60.0
|
||||
gopkg.in/yaml.v2 v2.4.0 // indirect
|
||||
)
|
||||
|
||||
require (
|
||||
cloud.google.com/go v0.97.0 // indirect
|
||||
github.com/census-instrumentation/opencensus-proto v0.3.0 // indirect
|
||||
github.com/cespare/xxhash v1.1.0 // indirect
|
||||
github.com/cespare/xxhash/v2 v2.1.2 // indirect
|
||||
github.com/cncf/udpa/go v0.0.0-20210930031921-04548b0d99d4 // indirect
|
||||
github.com/cncf/xds/go v0.0.0-20211011173535-cb28da3451f1 // indirect
|
||||
github.com/envoyproxy/go-control-plane v0.10.0 // indirect
|
||||
github.com/envoyproxy/protoc-gen-validate v0.6.2 // indirect
|
||||
github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
|
||||
github.com/golang/protobuf v1.5.2 // indirect
|
||||
github.com/google/go-cmp v0.5.6 // indirect
|
||||
github.com/googleapis/gax-go/v2 v2.1.1 // indirect
|
||||
github.com/russross/blackfriday/v2 v2.1.0 // indirect
|
||||
go.opencensus.io v0.23.0 // indirect
|
||||
golang.org/x/net v0.0.0-20210813160813-60bc85c4be6d // indirect
|
||||
golang.org/x/sys v0.0.0-20211025201205-69cdffdb9359 // indirect
|
||||
golang.org/x/text v0.3.7 // indirect
|
||||
golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1 // indirect
|
||||
google.golang.org/appengine v1.6.7 // indirect
|
||||
google.golang.org/genproto v0.0.0-20211101144312-62acf1d99145 // indirect
|
||||
google.golang.org/grpc v1.41.0 // indirect
|
||||
google.golang.org/protobuf v1.27.1 // indirect
|
||||
)
|
||||
|
||||
47
go.sum
47
go.sum
@@ -36,6 +36,8 @@ cloud.google.com/go/datastore v1.0.0/go.mod h1:LXYbyblFSglQ5pkeyhO+Qmw7ukd3C+pD7
|
||||
cloud.google.com/go/datastore v1.1.0/go.mod h1:umbIZjpQpHh4hmRpGhH4tLFup+FVzqBi1b3c64qFpCk=
|
||||
cloud.google.com/go/firestore v1.6.0 h1:dMIWvm+3O0E3DM7kcZPH0FBQ94Xg/OMkdTNDaY9itbI=
|
||||
cloud.google.com/go/firestore v1.6.0/go.mod h1:afJwI0vaXwAG54kI7A//lP/lSPDkQORQuMkv56TxEPU=
|
||||
cloud.google.com/go/firestore v1.6.1 h1:8rBq3zRjnHx8UtBvaOWqBB1xq9jH6/wltfQLlTMh2Fw=
|
||||
cloud.google.com/go/firestore v1.6.1/go.mod h1:asNXNOzBdyVQmEU+ggO8UPodTkEVFW5Qx+rwHnAz+EY=
|
||||
cloud.google.com/go/pubsub v1.0.1/go.mod h1:R0Gpsv3s54REJCy4fxDixWD93lHJMoZTyQ2kNxGRt3I=
|
||||
cloud.google.com/go/pubsub v1.1.0/go.mod h1:EwwdRX2sKPjnvnqCa270oGRyludottCI76h+R3AArQw=
|
||||
cloud.google.com/go/pubsub v1.2.0/go.mod h1:jhfEVHT8odbXTkndysNHCcx0awwzvfOlguIAii9o8iA=
|
||||
@@ -59,8 +61,13 @@ github.com/OneOfOne/xxhash v1.2.2/go.mod h1:HSdplMjZKSmBqAxg5vPj2TmRDmfkzw+cTzAE
|
||||
github.com/antihax/optional v1.0.0/go.mod h1:uupD/76wgC+ih3iEmQUL+0Ugr19nfwCT1kdvxnR2qWY=
|
||||
github.com/census-instrumentation/opencensus-proto v0.2.1 h1:glEXhBS5PSLLv4IXzLA5yPRVX4bilULVyxxbrfOtDAk=
|
||||
github.com/census-instrumentation/opencensus-proto v0.2.1/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
|
||||
github.com/census-instrumentation/opencensus-proto v0.3.0 h1:t/LhUZLVitR1Ow2YOnduCsavhwFUklBMoGVYUCqmCqk=
|
||||
github.com/census-instrumentation/opencensus-proto v0.3.0/go.mod h1:f6KPmirojxKA12rnyqOA5BBL4O983OfeGPqjHWSTneU=
|
||||
github.com/cespare/xxhash v1.1.0 h1:a6HrQnmkObjyL+Gs60czilIUGqrzKutQD6XZog3p+ko=
|
||||
github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghfAqPWnc=
|
||||
github.com/cespare/xxhash/v2 v2.1.1/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
|
||||
github.com/cespare/xxhash/v2 v2.1.2 h1:YRXhKfTDauu4ajMg1TPgFO5jnlC2HCbmLXMcTG5cbYE=
|
||||
github.com/cespare/xxhash/v2 v2.1.2/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs=
|
||||
github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI=
|
||||
github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI=
|
||||
github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU=
|
||||
@@ -69,8 +76,15 @@ github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGX
|
||||
github.com/cncf/udpa/go v0.0.0-20200629203442-efcf912fb354/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
|
||||
github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403 h1:cqQfy1jclcSy/FwLjemeg3SR1yaINm74aQyupQ0Bl8M=
|
||||
github.com/cncf/udpa/go v0.0.0-20201120205902-5459f2c99403/go.mod h1:WmhPx2Nbnhtbo57+VJT5O0JRkEi1Wbu0z5j0R8u5Hbk=
|
||||
github.com/cncf/udpa/go v0.0.0-20210930031921-04548b0d99d4 h1:hzAQntlaYRkVSFEfj9OTWlVV1H155FMD8BTKktLv0QI=
|
||||
github.com/cncf/udpa/go v0.0.0-20210930031921-04548b0d99d4/go.mod h1:6pvJx4me5XPnfI9Z40ddWsdw2W/uZgQLFXToKeRcDiI=
|
||||
github.com/cncf/xds/go v0.0.0-20210312221358-fbca930ec8ed h1:OZmjad4L3H8ncOIR8rnb5MREYqG8ixi5+WbeUsquF0c=
|
||||
github.com/cncf/xds/go v0.0.0-20210312221358-fbca930ec8ed/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
|
||||
github.com/cncf/xds/go v0.0.0-20210805033703-aa0b78936158/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
|
||||
github.com/cncf/xds/go v0.0.0-20210922020428-25de7278fc84/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
|
||||
github.com/cncf/xds/go v0.0.0-20211001041855-01bcc9b48dfe/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
|
||||
github.com/cncf/xds/go v0.0.0-20211011173535-cb28da3451f1 h1:zH8ljVhhq7yC0MIeUL/IviMtY8hx2mK8cN9wEYb8ggw=
|
||||
github.com/cncf/xds/go v0.0.0-20211011173535-cb28da3451f1/go.mod h1:eXthEFrGJvWHgFFCl3hGmgk+/aYT6PnTQLykKQRLhEs=
|
||||
github.com/cpuguy83/go-md2man/v2 v2.0.0-20190314233015-f79a8a8ca69d/go.mod h1:maD7wRr/U5Z6m/iR4s+kqSMx2CaBsrgA7czyZG/E6dU=
|
||||
github.com/cpuguy83/go-md2man/v2 v2.0.1 h1:r/myEWzV9lfsM1tFLgDyu0atFtJ1fXn261LKYj/3DxU=
|
||||
github.com/cpuguy83/go-md2man/v2 v2.0.1/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=
|
||||
@@ -83,8 +97,13 @@ github.com/envoyproxy/go-control-plane v0.9.9-0.20201210154907-fd9021fe5dad/go.m
|
||||
github.com/envoyproxy/go-control-plane v0.9.9-0.20210217033140-668b12f5399d/go.mod h1:cXg6YxExXjJnVBQHBLXeUAgxn2UodCpnH306RInaBQk=
|
||||
github.com/envoyproxy/go-control-plane v0.9.9-0.20210512163311-63b5d3c536b0 h1:dulLQAYQFYtG5MTplgNGHWuV2D+OBD+Z8lmDBmbLg+s=
|
||||
github.com/envoyproxy/go-control-plane v0.9.9-0.20210512163311-63b5d3c536b0/go.mod h1:hliV/p42l8fGbc6Y9bQ70uLwIvmJyVE5k4iMKlh8wCQ=
|
||||
github.com/envoyproxy/go-control-plane v0.9.10-0.20210907150352-cf90f659a021/go.mod h1:AFq3mo9L8Lqqiid3OhADV3RfLJnjiw63cSpi+fDTRC0=
|
||||
github.com/envoyproxy/go-control-plane v0.10.0 h1:WVt4HEPbdRbRD/PKKPbPnIVavO6gk/h673jWyIJ016k=
|
||||
github.com/envoyproxy/go-control-plane v0.10.0/go.mod h1:AY7fTTXNdv/aJ2O5jwpxAPOWUZ7hQAEvzN5Pf27BkQQ=
|
||||
github.com/envoyproxy/protoc-gen-validate v0.1.0 h1:EQciDnbrYxy13PgWoY8AqoxGiPrpgBZ1R8UNe3ddc+A=
|
||||
github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c=
|
||||
github.com/envoyproxy/protoc-gen-validate v0.6.2 h1:JiO+kJTpmYGjEodY7O1Zk8oZcNz1+f30UtwtXoFUPzE=
|
||||
github.com/envoyproxy/protoc-gen-validate v0.6.2/go.mod h1:2t7qjJNvHPx8IjnBOzl9E9/baC+qXE/TeeyBRzgJDws=
|
||||
github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
|
||||
github.com/go-gl/glfw v0.0.0-20190409004039-e6da0acd62b1/go.mod h1:vR7hzQXu2zJy9AVAgeJqvqgH9Q5CA+iKCZ2gyEVpxRU=
|
||||
github.com/go-gl/glfw/v3.3/glfw v0.0.0-20191125211704-12ad95a8df72/go.mod h1:tQ2UAYgL5IevRw8kRxooKSPJfGvJ9fJQFa0TUsXzTg8=
|
||||
@@ -94,6 +113,8 @@ github.com/golang/groupcache v0.0.0-20190702054246-869f871628b6/go.mod h1:cIg4er
|
||||
github.com/golang/groupcache v0.0.0-20191227052852-215e87163ea7/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
|
||||
github.com/golang/groupcache v0.0.0-20200121045136-8c9f03a8e57e h1:1r7pUrabqp18hOBcwBwiTsbnFeTZHV9eER/QT5JVZxY=
|
||||
github.com/golang/groupcache v0.0.0-20200121045136-8c9f03a8e57e/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
|
||||
github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da h1:oI5xCqsCo564l8iNU+DwB5epxmsaqB+rhGL0m5jtYqE=
|
||||
github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da/go.mod h1:cIg4eruTrX1D+g88fzRXU5OdNfaM+9IcxsU14FzY7Hc=
|
||||
github.com/golang/mock v1.1.1/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
|
||||
github.com/golang/mock v1.2.0/go.mod h1:oTYuIxOrZwtPieC+H1uAHpcLFnEyAGVDL/k47Jfbm0A=
|
||||
github.com/golang/mock v1.3.1/go.mod h1:sBzyDLLjw3U8JLTeZvSv8jJB+tU5PVekmnlKIyFUx0Y=
|
||||
@@ -168,16 +189,24 @@ github.com/googleapis/gax-go/v2 v2.1.1/go.mod h1:hddJymUZASv3XPyGkUpKj8pPO47Rmb0
|
||||
github.com/grpc-ecosystem/grpc-gateway v1.16.0/go.mod h1:BDjrQk3hbvj6Nolgz8mAMFbcEtjT1g+wF4CSlocrBnw=
|
||||
github.com/hashicorp/golang-lru v0.5.0/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
|
||||
github.com/hashicorp/golang-lru v0.5.1/go.mod h1:/m3WP610KZHVQ1SGc6re/UDhFvYD7pJ4Ao+sR/qLZy8=
|
||||
github.com/iancoleman/strcase v0.2.0/go.mod h1:iwCmte+B7n89clKwxIoIXy/HfoL7AsD47ZCWhYzw7ho=
|
||||
github.com/ianlancetaylor/demangle v0.0.0-20181102032728-5e5cf60278f6/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
|
||||
github.com/ianlancetaylor/demangle v0.0.0-20200824232613-28f6c0f3b639/go.mod h1:aSSvb/t6k1mPoxDqO4vJh6VOCGPwU4O0C2/Eqndh1Sc=
|
||||
github.com/jstemmer/go-junit-report v0.0.0-20190106144839-af01ea7f8024/go.mod h1:6v2b51hI/fHJwM22ozAgKL4VKDeJcHhJFhtBdhmNjmU=
|
||||
github.com/jstemmer/go-junit-report v0.9.1/go.mod h1:Brl9GWCQeLvo8nXZwPNNblvFj/XSXhF0NWZEnDohbsk=
|
||||
github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
|
||||
github.com/kr/fs v0.1.0/go.mod h1:FFnZGqtBN9Gxj7eW1uZ42v5BccTP0vu6NEaFoC2HwRg=
|
||||
github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
|
||||
github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
|
||||
github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
|
||||
github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
|
||||
github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
|
||||
github.com/lyft/protoc-gen-star v0.5.3/go.mod h1:V0xaHgaf5oCCqmcxYcWiDfTiKsZsRc87/1qhoTACD8w=
|
||||
github.com/mattn/go-sqlite3 v1.14.8/go.mod h1:NyWgC/yNuGj7Q9rpYnZvas74GogHl5/Z4A/KQRfk6bU=
|
||||
github.com/mattn/go-sqlite3 v1.14.9 h1:10HX2Td0ocZpYEjhilsuo6WWtUqttj2Kb0KtD86/KYA=
|
||||
github.com/mattn/go-sqlite3 v1.14.9/go.mod h1:NyWgC/yNuGj7Q9rpYnZvas74GogHl5/Z4A/KQRfk6bU=
|
||||
github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
|
||||
github.com/pkg/sftp v1.10.1/go.mod h1:lYOWFsE0bwd1+KfKJaKeuokY15vzFx25BLbzYYoAxZI=
|
||||
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA=
|
||||
github.com/rogpeppe/fastuuid v1.2.0/go.mod h1:jVj6XXZzXRy/MSR5jhDC/2q6DgLz+nrA6LYCDYWNEvQ=
|
||||
@@ -188,10 +217,13 @@ github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQD
|
||||
github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc=
|
||||
github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72 h1:qLC7fQah7D6K1B0ujays3HV9gkFtllcxhzImRR7ArPQ=
|
||||
github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
|
||||
github.com/spf13/afero v1.3.3/go.mod h1:5KUK8ByomD5Ti5Artl0RtHeI5pTF7MIDuXL3yY520V4=
|
||||
github.com/spf13/afero v1.6.0/go.mod h1:Ai8FlHk4v/PARR026UzYexafAt9roJ7LcLMAmO6Z93I=
|
||||
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
|
||||
github.com/stretchr/testify v1.4.0/go.mod h1:j7eGeouHqKxXV5pUuKE4zz7dFj8WfuZ+81PSLYec5m4=
|
||||
github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
|
||||
github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
|
||||
github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
|
||||
github.com/urfave/cli/v2 v2.3.0 h1:qph92Y649prgesehzOrQjdWyxFOp/QVM+6imKHad91M=
|
||||
github.com/urfave/cli/v2 v2.3.0/go.mod h1:LJmUH05zAU44vOAcrfzZQKsZbVcdbOG8rtL3/XcUArI=
|
||||
github.com/yuin/goldmark v1.1.25/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
|
||||
@@ -211,6 +243,7 @@ go.opentelemetry.io/proto/otlp v0.7.0/go.mod h1:PqfVotwruBrMGOCsRd/89rSnXhoiJIqe
|
||||
golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
|
||||
golang.org/x/crypto v0.0.0-20190510104115-cbcb75029529/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
|
||||
golang.org/x/crypto v0.0.0-20190605123033-f99c8df09eb5/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
|
||||
golang.org/x/crypto v0.0.0-20190820162420-60c769a6c586/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
|
||||
golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
|
||||
golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
|
||||
golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA=
|
||||
@@ -248,6 +281,7 @@ golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
|
||||
golang.org/x/mod v0.4.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
|
||||
golang.org/x/mod v0.4.1/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
|
||||
golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
|
||||
golang.org/x/mod v0.5.0/go.mod h1:5OXOZSfqPIIbmVBIIKWRFfZjPR0E5r58TLhUjH0a2Ro=
|
||||
golang.org/x/net v0.0.0-20180724234803-3673e40ba225/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
|
||||
golang.org/x/net v0.0.0-20180826012351-8a410e7b638d/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
|
||||
golang.org/x/net v0.0.0-20190108225652-1e06a53dbb7e/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
|
||||
@@ -284,6 +318,8 @@ golang.org/x/net v0.0.0-20210316092652-d523dce5a7f4/go.mod h1:RBQZq4jEuRlivfhVLd
|
||||
golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
|
||||
golang.org/x/net v0.0.0-20210503060351-7fd8e65b6420 h1:a8jGStKg0XqKDlKqjLrXn0ioF5MH36pT7Z0BRTqLhbk=
|
||||
golang.org/x/net v0.0.0-20210503060351-7fd8e65b6420/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
|
||||
golang.org/x/net v0.0.0-20210813160813-60bc85c4be6d h1:LO7XpTYMwTqxjLcGWPijK3vRXg1aWdlNOVOHRq45d7c=
|
||||
golang.org/x/net v0.0.0-20210813160813-60bc85c4be6d/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
|
||||
golang.org/x/oauth2 v0.0.0-20180821212333-d2e6202438be/go.mod h1:N/0e6XlmueqKjAGxoOufVs8QHGRruUQn6yWY3a++T0U=
|
||||
golang.org/x/oauth2 v0.0.0-20190226205417-e64efc72b421/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
|
||||
golang.org/x/oauth2 v0.0.0-20190604053449-0f29369cfe45/go.mod h1:gOpvHmFTYa4IltrdGE7lF6nIHvwfUNPOp7c8zoXwtLw=
|
||||
@@ -356,9 +392,11 @@ golang.org/x/sys v0.0.0-20210603125802-9665404d3644/go.mod h1:oPkhp1MJrh7nUepCBc
|
||||
golang.org/x/sys v0.0.0-20210616094352-59db8d763f22/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.0.0-20210630005230-0f9fa26af87c/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.0.0-20210806184541-e5e7981a1069/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.0.0-20210816183151-1e6c022a8912/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.0.0-20210823070655-63515b42dcdf/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.0.0-20210908233432-aa78b53d3365/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.0.0-20210917161153-d61c044b1678/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.0.0-20211007075335-d3039528d8ac/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.0.0-20211025201205-69cdffdb9359 h1:2B5p2L5IfGiD7+b9BOoRMC6DgObAVZV+Fsp050NqXik=
|
||||
golang.org/x/sys v0.0.0-20211025201205-69cdffdb9359/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
|
||||
@@ -371,6 +409,8 @@ golang.org/x/text v0.3.4/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
|
||||
golang.org/x/text v0.3.5/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
|
||||
golang.org/x/text v0.3.6 h1:aRYxNxv6iGQlyVaZmk6ZgYEDa+Jg18DxebPSrd6bg1M=
|
||||
golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
|
||||
golang.org/x/text v0.3.7 h1:olpwvP2KacW1ZWvsR7uQhoyTYvKAupfQrRGBFM352Gk=
|
||||
golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
|
||||
golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
|
||||
golang.org/x/time v0.0.0-20190308202827-9d24e82272b4/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
|
||||
golang.org/x/time v0.0.0-20191024005414-555d28b269f0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ=
|
||||
@@ -462,6 +502,7 @@ google.golang.org/api v0.55.0/go.mod h1:38yMfeP1kfjsl8isn0tliTjIb1rJXcQi4UXlbqiv
|
||||
google.golang.org/api v0.56.0/go.mod h1:38yMfeP1kfjsl8isn0tliTjIb1rJXcQi4UXlbqivdVE=
|
||||
google.golang.org/api v0.57.0/go.mod h1:dVPlbZyBo2/OjBpmvNdpn2GRm6rPy75jyU7bmhdrMgI=
|
||||
google.golang.org/api v0.58.0/go.mod h1:cAbP2FsxoGVNwtgNAmmn3y5G1TWAiVYRmg4yku3lv+E=
|
||||
google.golang.org/api v0.59.0/go.mod h1:sT2boj7M9YJxZzgeZqXogmhfmRWDtPzT31xkieUbuZU=
|
||||
google.golang.org/api v0.60.0 h1:eq/zs5WPH4J9undYM9IP1O7dSr7Yh8Y0GtSCpzGzIUk=
|
||||
google.golang.org/api v0.60.0/go.mod h1:d7rl65NZAkEQ90JFzqBjcRq1TVeG5ZoGV3sSpEnnVb4=
|
||||
google.golang.org/appengine v1.1.0/go.mod h1:EbEs0AVv82hx2wNQdGPgUI5lhzA/G0D9YwlJXL52JkM=
|
||||
@@ -529,9 +570,13 @@ google.golang.org/genproto v0.0.0-20210903162649-d08c68adba83/go.mod h1:eFjDcFEc
|
||||
google.golang.org/genproto v0.0.0-20210909211513-a8c4777a87af/go.mod h1:eFjDcFEctNawg4eG61bRv87N7iHBWyVhJu7u1kqDUXY=
|
||||
google.golang.org/genproto v0.0.0-20210917145530-b395a37504d4/go.mod h1:eFjDcFEctNawg4eG61bRv87N7iHBWyVhJu7u1kqDUXY=
|
||||
google.golang.org/genproto v0.0.0-20210924002016-3dee208752a0/go.mod h1:5CzLGKJ67TSI2B9POpiiyGha0AjJvZIUgRMt1dSmuhc=
|
||||
google.golang.org/genproto v0.0.0-20211008145708-270636b82663/go.mod h1:5CzLGKJ67TSI2B9POpiiyGha0AjJvZIUgRMt1dSmuhc=
|
||||
google.golang.org/genproto v0.0.0-20211016002631-37fc39342514/go.mod h1:5CzLGKJ67TSI2B9POpiiyGha0AjJvZIUgRMt1dSmuhc=
|
||||
google.golang.org/genproto v0.0.0-20211021150943-2b146023228c h1:FqrtZMB5Wr+/RecOM3uPJNPfWR8Upb5hAPnt7PU6i4k=
|
||||
google.golang.org/genproto v0.0.0-20211021150943-2b146023228c/go.mod h1:5CzLGKJ67TSI2B9POpiiyGha0AjJvZIUgRMt1dSmuhc=
|
||||
google.golang.org/genproto v0.0.0-20211028162531-8db9c33dc351/go.mod h1:5CzLGKJ67TSI2B9POpiiyGha0AjJvZIUgRMt1dSmuhc=
|
||||
google.golang.org/genproto v0.0.0-20211101144312-62acf1d99145 h1:vum3nDKdleYb+aePXKFEDT2+ghuH00EgYp9B7Q7EZZE=
|
||||
google.golang.org/genproto v0.0.0-20211101144312-62acf1d99145/go.mod h1:5CzLGKJ67TSI2B9POpiiyGha0AjJvZIUgRMt1dSmuhc=
|
||||
google.golang.org/grpc v1.19.0/go.mod h1:mqu4LbDTu4XGKhr4mRzUsmM4RtVoemTSY81AxZiDr8c=
|
||||
google.golang.org/grpc v1.20.1/go.mod h1:10oTOabMzJvdu6/UiuZezV6QK5dSlG84ov/aaiqXj38=
|
||||
google.golang.org/grpc v1.21.1/go.mod h1:oYelfM1adQP15Ek0mdvEgi9Df8B9CZIaU1084ijfRaM=
|
||||
@@ -558,6 +603,8 @@ google.golang.org/grpc v1.39.0/go.mod h1:PImNr+rS9TWYb2O4/emRugxiyHZ5JyHW5F+RPnD
|
||||
google.golang.org/grpc v1.39.1/go.mod h1:PImNr+rS9TWYb2O4/emRugxiyHZ5JyHW5F+RPnDzfrE=
|
||||
google.golang.org/grpc v1.40.0 h1:AGJ0Ih4mHjSeibYkFGh1dD9KJ/eOtZ93I6hoHhukQ5Q=
|
||||
google.golang.org/grpc v1.40.0/go.mod h1:ogyxbiOoUXAkP+4+xa6PZSE9DZgIHtSpzjDTB9KAK34=
|
||||
google.golang.org/grpc v1.41.0 h1:f+PlOh7QV4iIJkPrx5NQ7qaNGFQ3OTse67yaDHfju4E=
|
||||
google.golang.org/grpc v1.41.0/go.mod h1:U3l9uK9J0sini8mHphKoXyaqDA/8VyGnDee1zzIUK6k=
|
||||
google.golang.org/grpc/cmd/protoc-gen-go-grpc v1.1.0/go.mod h1:6Kw0yEErY5E/yWrBtf03jp27GLLJujG4z/JK95pnjjw=
|
||||
google.golang.org/protobuf v0.0.0-20200109180630-ec00e32a8dfd/go.mod h1:DFci5gLYBciE7Vtevhsrf46CRTquxDuWsQurQQe4oz8=
|
||||
google.golang.org/protobuf v0.0.0-20200221191635-4d8936d0db64/go.mod h1:kwYJMbMJ01Woi6D6+Kah6886xMZcty6N08ah7+eCXa0=
|
||||
|
||||
14
server/cache.go
Normal file
14
server/cache.go
Normal file
@@ -0,0 +1,14 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
_ "github.com/mattn/go-sqlite3" // SQLite driver
|
||||
"time"
|
||||
)
|
||||
|
||||
type cache interface {
|
||||
AddMessage(m *message) error
|
||||
Messages(topic string, since time.Time) ([]*message, error)
|
||||
MessageCount(topic string) (int, error)
|
||||
Topics() (map[string]*topic, error)
|
||||
Prune(keep time.Duration) error
|
||||
}
|
||||
80
server/cache_mem.go
Normal file
80
server/cache_mem.go
Normal file
@@ -0,0 +1,80 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
_ "github.com/mattn/go-sqlite3" // SQLite driver
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
type memCache struct {
|
||||
messages map[string][]*message
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
var _ cache = (*memCache)(nil)
|
||||
|
||||
func newMemCache() *memCache {
|
||||
return &memCache{
|
||||
messages: make(map[string][]*message),
|
||||
}
|
||||
}
|
||||
|
||||
func (s *memCache) AddMessage(m *message) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if _, ok := s.messages[m.Topic]; !ok {
|
||||
s.messages[m.Topic] = make([]*message, 0)
|
||||
}
|
||||
s.messages[m.Topic] = append(s.messages[m.Topic], m)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *memCache) Messages(topic string, since time.Time) ([]*message, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if _, ok := s.messages[topic]; !ok {
|
||||
return make([]*message, 0), nil
|
||||
}
|
||||
messages := make([]*message, 0) // copy!
|
||||
for _, m := range s.messages[topic] {
|
||||
msgTime := time.Unix(m.Time, 0)
|
||||
if msgTime == since || msgTime.After(since) {
|
||||
messages = append(messages, m)
|
||||
}
|
||||
}
|
||||
return messages, nil
|
||||
}
|
||||
|
||||
func (s *memCache) MessageCount(topic string) (int, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if _, ok := s.messages[topic]; !ok {
|
||||
return 0, nil
|
||||
}
|
||||
return len(s.messages[topic]), nil
|
||||
}
|
||||
|
||||
func (s *memCache) Topics() (map[string]*topic, error) {
|
||||
// Hack since we know when this is called there are no messages!
|
||||
return make(map[string]*topic), nil
|
||||
}
|
||||
|
||||
func (s *memCache) Prune(keep time.Duration) error {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
for topic, _ := range s.messages {
|
||||
s.pruneTopic(topic, keep)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *memCache) pruneTopic(topic string, keep time.Duration) {
|
||||
for i, m := range s.messages[topic] {
|
||||
msgTime := time.Unix(m.Time, 0)
|
||||
if time.Since(msgTime) < keep {
|
||||
s.messages[topic] = s.messages[topic][i:]
|
||||
return
|
||||
}
|
||||
}
|
||||
s.messages[topic] = make([]*message, 0) // all messages expired
|
||||
}
|
||||
127
server/cache_sqlite.go
Normal file
127
server/cache_sqlite.go
Normal file
@@ -0,0 +1,127 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"database/sql"
|
||||
"errors"
|
||||
_ "github.com/mattn/go-sqlite3" // SQLite driver
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
createTableQuery = `
|
||||
BEGIN;
|
||||
CREATE TABLE IF NOT EXISTS messages (
|
||||
id VARCHAR(20) PRIMARY KEY,
|
||||
time INT NOT NULL,
|
||||
topic VARCHAR(64) NOT NULL,
|
||||
message VARCHAR(1024) NOT NULL
|
||||
);
|
||||
CREATE INDEX IF NOT EXISTS idx_topic ON messages (topic);
|
||||
COMMIT;
|
||||
`
|
||||
insertMessageQuery = `INSERT INTO messages (id, time, topic, message) VALUES (?, ?, ?, ?)`
|
||||
pruneMessagesQuery = `DELETE FROM messages WHERE time < ?`
|
||||
selectMessagesSinceTimeQuery = `
|
||||
SELECT id, time, message
|
||||
FROM messages
|
||||
WHERE topic = ? AND time >= ?
|
||||
ORDER BY time ASC
|
||||
`
|
||||
selectMessageCountQuery = `SELECT count(*) FROM messages WHERE topic = ?`
|
||||
selectTopicsQuery = `SELECT topic, MAX(time) FROM messages GROUP BY TOPIC`
|
||||
)
|
||||
|
||||
type sqliteCache struct {
|
||||
db *sql.DB
|
||||
}
|
||||
|
||||
var _ cache = (*sqliteCache)(nil)
|
||||
|
||||
func newSqliteCache(filename string) (*sqliteCache, error) {
|
||||
db, err := sql.Open("sqlite3", filename)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if _, err := db.Exec(createTableQuery); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &sqliteCache{
|
||||
db: db,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (c *sqliteCache) AddMessage(m *message) error {
|
||||
_, err := c.db.Exec(insertMessageQuery, m.ID, m.Time, m.Topic, m.Message)
|
||||
return err
|
||||
}
|
||||
|
||||
func (c *sqliteCache) Messages(topic string, since time.Time) ([]*message, error) {
|
||||
rows, err := c.db.Query(selectMessagesSinceTimeQuery, topic, since.Unix())
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
messages := make([]*message, 0)
|
||||
for rows.Next() {
|
||||
var timestamp int64
|
||||
var id, msg string
|
||||
if err := rows.Scan(&id, ×tamp, &msg); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
messages = append(messages, &message{
|
||||
ID: id,
|
||||
Time: timestamp,
|
||||
Event: messageEvent,
|
||||
Topic: topic,
|
||||
Message: msg,
|
||||
})
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return messages, nil
|
||||
}
|
||||
|
||||
func (c *sqliteCache) MessageCount(topic string) (int, error) {
|
||||
rows, err := c.db.Query(selectMessageCountQuery, topic)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
defer rows.Close()
|
||||
var count int
|
||||
if !rows.Next() {
|
||||
return 0, errors.New("no rows found")
|
||||
}
|
||||
if err := rows.Scan(&count); err != nil {
|
||||
return 0, err
|
||||
} else if err := rows.Err(); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return count, nil
|
||||
}
|
||||
|
||||
func (s *sqliteCache) Topics() (map[string]*topic, error) {
|
||||
rows, err := s.db.Query(selectTopicsQuery)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
topics := make(map[string]*topic, 0)
|
||||
for rows.Next() {
|
||||
var id string
|
||||
var last int64
|
||||
if err := rows.Scan(&id, &last); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
topics[id] = newTopic(id, time.Unix(last, 0))
|
||||
}
|
||||
if err := rows.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return topics, nil
|
||||
}
|
||||
|
||||
func (c *sqliteCache) Prune(keep time.Duration) error {
|
||||
_, err := c.db.Exec(pruneMessagesQuery, time.Now().Add(-1 * keep).Unix())
|
||||
return err
|
||||
}
|
||||
@@ -25,12 +25,12 @@
|
||||
<meta property="og:site_name" content="ntfy.sh" />
|
||||
<meta property="og:title" content="ntfy.sh | simple HTTP-based pub-sub" />
|
||||
<meta property="og:description" content="ntfy is a simple HTTP-based pub-sub notification service. It allows you to send desktop notifications via scripts from any computer, entirely without signup or cost. Made with ❤ by Philipp C. Heckel, Apache License 2.0, source at https://heckel.io/ntfy." />
|
||||
<meta property="og:image" content="/static/img/favicon.png" />
|
||||
<meta property="og:image" content="/static/img/ntfy.png" />
|
||||
<meta property="og:url" content="https://ntfy.sh" />
|
||||
</head>
|
||||
<body>
|
||||
<div id="main">
|
||||
<h1>ntfy.sh - simple HTTP-based pub-sub</h1>
|
||||
<h1><img src="static/img/ntfy.png" alt="ntfy"/><br/>ntfy.sh - simple HTTP-based pub-sub</h1>
|
||||
<p>
|
||||
<b>ntfy</b> (pronounce: <i>notify</i>) is a simple HTTP-based <a href="https://en.wikipedia.org/wiki/Publish%E2%80%93subscribe_pattern">pub-sub</a> notification service.
|
||||
It allows you to send <b>desktop notifications via scripts from any computer</b>, entirely <b>without signup or cost</b>.
|
||||
@@ -81,6 +81,12 @@
|
||||
<ul id="topicsList"></ul>
|
||||
<audio id="notifySound" src="static/sound/mixkit-message-pop-alert-2354.mp3"></audio>
|
||||
|
||||
<h3>Subscribe via phone</h3>
|
||||
<p>
|
||||
Once it's approved, you can use the <b>Ntfy Android App</b> to receive notifications directly on your phone. Just like
|
||||
the server, this app is also <a href="https://github.com/binwiederhier/ntfy-android">open source</a>.
|
||||
</p>
|
||||
|
||||
<h3>Subscribe via your app, or via the CLI</h3>
|
||||
<p class="smallMarginBottom">
|
||||
Using <a href="https://developer.mozilla.org/en-US/docs/Web/API/EventSource">EventSource</a> in JS, you can consume
|
||||
@@ -142,6 +148,7 @@
|
||||
$ curl -s "ntfy.sh/mytopic/json?poll=1&since=10m"<br/>
|
||||
# Returns messages from up to 10 minutes ago and ends the connection
|
||||
</code>
|
||||
|
||||
<h2>FAQ</h2>
|
||||
<p>
|
||||
<b>Isn't this like ...?</b><br/>
|
||||
@@ -165,6 +172,28 @@
|
||||
That said, the logs do not contain any topic names or other details about you. Check the code if you don't believe me.
|
||||
</p>
|
||||
|
||||
<p>
|
||||
<b>Why is Firebase used?</b><br/>
|
||||
In addition to caching messages locally and delivering them to long-polling subscribers, all messages are also
|
||||
published to Firebase Cloud Messaging (FCM) (if <tt>FirebaseKeyFile</tt> is set, which it is on ntfy.sh). This
|
||||
is to facilitate instant notifications on Android. I tried really, really hard to avoid using FCM, but newer
|
||||
versions of Android made it impossible to implement <a href="https://developer.android.com/guide/background">background services</a>>.
|
||||
I'm sorry.
|
||||
</p>
|
||||
|
||||
<h2>Privacy policy</h2>
|
||||
<p>
|
||||
Neither the server nor the app record any personal information, or share any of the messages and topics with
|
||||
any outside service. All data is exclusively used to make the service function properly. The notable exception
|
||||
is the Firebase Cloud Messaging (FCM) service, which is required to provide instant Android notifications (see
|
||||
FAQ for details).
|
||||
</p>
|
||||
|
||||
<p>
|
||||
The web server does not log or otherwise store request paths, remote IP addresses or even topics or messages,
|
||||
aside from a short on-disk cache (up to a day) to support service restarts.
|
||||
</p>
|
||||
|
||||
<center id="ironicCenterTagDontFreakOut"><i>Made with ❤️ by <a href="https://heckel.io">Philipp C. Heckel</a></i></center>
|
||||
</div>
|
||||
<script src="static/js/app.js"></script>
|
||||
|
||||
141
server/server.go
141
server/server.go
@@ -9,7 +9,6 @@ import (
|
||||
firebase "firebase.google.com/go"
|
||||
"firebase.google.com/go/messaging"
|
||||
"fmt"
|
||||
"golang.org/x/time/rate"
|
||||
"google.golang.org/api/option"
|
||||
"heckel.io/ntfy/config"
|
||||
"io"
|
||||
@@ -23,9 +22,8 @@ import (
|
||||
"time"
|
||||
)
|
||||
|
||||
// TODO add "max connections open" limit
|
||||
// TODO add "max messages in a topic" limit
|
||||
// TODO add "max topics" limit
|
||||
// TODO implement "since=<ID>"
|
||||
|
||||
// Server is the main server
|
||||
type Server struct {
|
||||
@@ -34,15 +32,10 @@ type Server struct {
|
||||
visitors map[string]*visitor
|
||||
firebase subscriber
|
||||
messages int64
|
||||
cache cache
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
// visitor represents an API user, and its associated rate.Limiter used for rate limiting
|
||||
type visitor struct {
|
||||
limiter *rate.Limiter
|
||||
seen time.Time
|
||||
}
|
||||
|
||||
// errHTTP is a generic HTTP error for any non-200 HTTP error
|
||||
type errHTTP struct {
|
||||
Code int
|
||||
@@ -54,8 +47,7 @@ func (e errHTTP) Error() string {
|
||||
}
|
||||
|
||||
const (
|
||||
messageLimit = 1024
|
||||
visitorExpungeAfter = 30 * time.Minute
|
||||
messageLimit = 1024
|
||||
)
|
||||
|
||||
var (
|
||||
@@ -86,14 +78,30 @@ func New(conf *config.Config) (*Server, error) {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
cache, err := createCache(conf)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
topics, err := cache.Topics()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &Server{
|
||||
config: conf,
|
||||
cache: cache,
|
||||
firebase: firebaseSubscriber,
|
||||
topics: make(map[string]*topic),
|
||||
topics: topics,
|
||||
visitors: make(map[string]*visitor),
|
||||
}, nil
|
||||
}
|
||||
|
||||
func createCache(conf *config.Config) (cache, error) {
|
||||
if conf.CacheFile != "" {
|
||||
return newSqliteCache(conf.CacheFile)
|
||||
}
|
||||
return newMemCache(), nil
|
||||
}
|
||||
|
||||
func createFirebaseSubscriber(conf *config.Config) (subscriber, error) {
|
||||
fb, err := firebase.NewApp(context.Background(), nil, option.WithCredentialsFile(conf.FirebaseKeyFile))
|
||||
if err != nil {
|
||||
@@ -147,21 +155,21 @@ func (s *Server) handle(w http.ResponseWriter, r *http.Request) {
|
||||
|
||||
func (s *Server) handleInternal(w http.ResponseWriter, r *http.Request) error {
|
||||
v := s.visitor(r.RemoteAddr)
|
||||
if !v.limiter.Allow() {
|
||||
return errHTTPTooManyRequests
|
||||
if err := v.RequestAllowed(); err != nil {
|
||||
return err
|
||||
}
|
||||
if r.Method == http.MethodGet && r.URL.Path == "/" {
|
||||
return s.handleHome(w, r)
|
||||
} else if r.Method == http.MethodGet && staticRegex.MatchString(r.URL.Path) {
|
||||
return s.handleStatic(w, r)
|
||||
} else if (r.Method == http.MethodPut || r.Method == http.MethodPost) && topicRegex.MatchString(r.URL.Path) {
|
||||
return s.handlePublish(w, r)
|
||||
return s.handlePublish(w, r, v)
|
||||
} else if r.Method == http.MethodGet && jsonRegex.MatchString(r.URL.Path) {
|
||||
return s.handleSubscribeJSON(w, r)
|
||||
return s.handleSubscribeJSON(w, r, v)
|
||||
} else if r.Method == http.MethodGet && sseRegex.MatchString(r.URL.Path) {
|
||||
return s.handleSubscribeSSE(w, r)
|
||||
return s.handleSubscribeSSE(w, r, v)
|
||||
} else if r.Method == http.MethodGet && rawRegex.MatchString(r.URL.Path) {
|
||||
return s.handleSubscribeRaw(w, r)
|
||||
return s.handleSubscribeRaw(w, r, v)
|
||||
} else if r.Method == http.MethodOptions {
|
||||
return s.handleOptions(w, r)
|
||||
}
|
||||
@@ -178,14 +186,21 @@ func (s *Server) handleStatic(w http.ResponseWriter, r *http.Request) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Server) handlePublish(w http.ResponseWriter, r *http.Request) error {
|
||||
t := s.createTopic(r.URL.Path[1:])
|
||||
func (s *Server) handlePublish(w http.ResponseWriter, r *http.Request, v *visitor) error {
|
||||
t, err := s.topic(r.URL.Path[1:])
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
reader := io.LimitReader(r.Body, messageLimit)
|
||||
b, err := io.ReadAll(reader)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := t.Publish(newDefaultMessage(t.id, string(b))); err != nil {
|
||||
m := newDefaultMessage(t.id, string(b))
|
||||
if err := t.Publish(m); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := s.cache.AddMessage(m); err != nil {
|
||||
return err
|
||||
}
|
||||
w.Header().Set("Access-Control-Allow-Origin", "*") // CORS, allow cross-origin requests
|
||||
@@ -195,7 +210,7 @@ func (s *Server) handlePublish(w http.ResponseWriter, r *http.Request) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Server) handleSubscribeJSON(w http.ResponseWriter, r *http.Request) error {
|
||||
func (s *Server) handleSubscribeJSON(w http.ResponseWriter, r *http.Request, v *visitor) error {
|
||||
encoder := func(msg *message) (string, error) {
|
||||
var buf bytes.Buffer
|
||||
if err := json.NewEncoder(&buf).Encode(&msg); err != nil {
|
||||
@@ -203,10 +218,10 @@ func (s *Server) handleSubscribeJSON(w http.ResponseWriter, r *http.Request) err
|
||||
}
|
||||
return buf.String(), nil
|
||||
}
|
||||
return s.handleSubscribe(w, r, "json", "application/stream+json", encoder)
|
||||
return s.handleSubscribe(w, r, v, "json", "application/stream+json", encoder)
|
||||
}
|
||||
|
||||
func (s *Server) handleSubscribeSSE(w http.ResponseWriter, r *http.Request) error {
|
||||
func (s *Server) handleSubscribeSSE(w http.ResponseWriter, r *http.Request, v *visitor) error {
|
||||
encoder := func(msg *message) (string, error) {
|
||||
var buf bytes.Buffer
|
||||
if err := json.NewEncoder(&buf).Encode(&msg); err != nil {
|
||||
@@ -217,21 +232,28 @@ func (s *Server) handleSubscribeSSE(w http.ResponseWriter, r *http.Request) erro
|
||||
}
|
||||
return fmt.Sprintf("data: %s\n", buf.String()), nil
|
||||
}
|
||||
return s.handleSubscribe(w, r, "sse", "text/event-stream", encoder)
|
||||
return s.handleSubscribe(w, r, v, "sse", "text/event-stream", encoder)
|
||||
}
|
||||
|
||||
func (s *Server) handleSubscribeRaw(w http.ResponseWriter, r *http.Request) error {
|
||||
func (s *Server) handleSubscribeRaw(w http.ResponseWriter, r *http.Request, v *visitor) error {
|
||||
encoder := func(msg *message) (string, error) {
|
||||
if msg.Event == "" { // only handle default events
|
||||
if msg.Event == messageEvent { // only handle default events
|
||||
return strings.ReplaceAll(msg.Message, "\n", " ") + "\n", nil
|
||||
}
|
||||
return "\n", nil // "keepalive" and "open" events just send an empty line
|
||||
}
|
||||
return s.handleSubscribe(w, r, "raw", "text/plain", encoder)
|
||||
return s.handleSubscribe(w, r, v, "raw", "text/plain", encoder)
|
||||
}
|
||||
|
||||
func (s *Server) handleSubscribe(w http.ResponseWriter, r *http.Request, format string, contentType string, encoder messageEncoder) error {
|
||||
t := s.createTopic(strings.TrimSuffix(r.URL.Path[1:], "/"+format)) // Hack
|
||||
func (s *Server) handleSubscribe(w http.ResponseWriter, r *http.Request, v *visitor, format string, contentType string, encoder messageEncoder) error {
|
||||
if err := v.AddSubscription(); err != nil {
|
||||
return errHTTPTooManyRequests
|
||||
}
|
||||
defer v.RemoveSubscription()
|
||||
t, err := s.topic(strings.TrimSuffix(r.URL.Path[1:], "/"+format)) // Hack
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
since, err := parseSince(r)
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -253,23 +275,22 @@ func (s *Server) handleSubscribe(w http.ResponseWriter, r *http.Request, format
|
||||
w.Header().Set("Access-Control-Allow-Origin", "*") // CORS, allow cross-origin requests
|
||||
w.Header().Set("Content-Type", contentType)
|
||||
if poll {
|
||||
return sendOldMessages(t, since, sub)
|
||||
return s.sendOldMessages(t, since, sub)
|
||||
}
|
||||
subscriberID := t.Subscribe(sub)
|
||||
defer t.Unsubscribe(subscriberID)
|
||||
if err := sub(newOpenMessage(t.id)); err != nil { // Send out open message
|
||||
return err
|
||||
}
|
||||
if err := sendOldMessages(t, since, sub); err != nil {
|
||||
if err := s.sendOldMessages(t, since, sub); err != nil {
|
||||
return err
|
||||
}
|
||||
for {
|
||||
select {
|
||||
case <-t.ctx.Done():
|
||||
return nil
|
||||
case <-r.Context().Done():
|
||||
return nil
|
||||
case <-time.After(s.config.KeepaliveInterval):
|
||||
v.Keepalive()
|
||||
if err := sub(newKeepaliveMessage(t.id)); err != nil { // Send keepalive message
|
||||
return err
|
||||
}
|
||||
@@ -277,11 +298,15 @@ func (s *Server) handleSubscribe(w http.ResponseWriter, r *http.Request, format
|
||||
}
|
||||
}
|
||||
|
||||
func sendOldMessages(t *topic, since time.Time, sub subscriber) error {
|
||||
func (s *Server) sendOldMessages(t *topic, since time.Time, sub subscriber) error {
|
||||
if since.IsZero() {
|
||||
return nil
|
||||
}
|
||||
for _, m := range t.Messages(since) {
|
||||
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
|
||||
}
|
||||
@@ -308,16 +333,19 @@ func (s *Server) handleOptions(w http.ResponseWriter, r *http.Request) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Server) createTopic(id string) *topic {
|
||||
func (s *Server) topic(id string) (*topic, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if _, ok := s.topics[id]; !ok {
|
||||
s.topics[id] = newTopic(id)
|
||||
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)
|
||||
}
|
||||
}
|
||||
return s.topics[id]
|
||||
return s.topics[id], nil
|
||||
}
|
||||
|
||||
func (s *Server) updateStatsAndExpire() {
|
||||
@@ -326,27 +354,34 @@ func (s *Server) updateStatsAndExpire() {
|
||||
|
||||
// Expire visitors from rate visitors map
|
||||
for ip, v := range s.visitors {
|
||||
if time.Since(v.seen) > visitorExpungeAfter {
|
||||
if v.Stale() {
|
||||
delete(s.visitors, ip)
|
||||
}
|
||||
}
|
||||
|
||||
// Prune old messages, remove topics without subscribers
|
||||
for _, t := range s.topics {
|
||||
t.Prune(s.config.MessageBufferDuration)
|
||||
subs, msgs := t.Stats()
|
||||
if msgs == 0 && (subs == 0 || (s.firebase != nil && subs == 1)) {
|
||||
delete(s.topics, t.id)
|
||||
}
|
||||
// Prune cache
|
||||
if err := s.cache.Prune(s.config.MessageBufferDuration); err != nil {
|
||||
log.Printf("error pruning cache: %s", err.Error())
|
||||
}
|
||||
|
||||
// Print stats
|
||||
// Prune old messages, remove subscriptions without subscribers
|
||||
var subscribers, messages int
|
||||
for _, t := range s.topics {
|
||||
subs, msgs := t.Stats()
|
||||
subs := t.Subscribers()
|
||||
msgs, err := s.cache.MessageCount(t.id)
|
||||
if err != nil {
|
||||
log.Printf("cannot get stats for topic %s: %s", t.id, err.Error())
|
||||
continue
|
||||
}
|
||||
if msgs == 0 && (subs == 0 || (s.firebase != nil && subs == 1)) { // Firebase is a subscriber!
|
||||
delete(s.topics, t.id)
|
||||
continue
|
||||
}
|
||||
subscribers += subs
|
||||
messages += msgs
|
||||
}
|
||||
|
||||
// Print stats
|
||||
log.Printf("Stats: %d message(s) published, %d topic(s) active, %d subscriber(s), %d message(s) buffered, %d visitor(s)",
|
||||
s.messages, len(s.topics), subscribers, messages, len(s.visitors))
|
||||
}
|
||||
@@ -362,12 +397,8 @@ func (s *Server) visitor(remoteAddr string) *visitor {
|
||||
}
|
||||
v, exists := s.visitors[ip]
|
||||
if !exists {
|
||||
v = &visitor{
|
||||
rate.NewLimiter(s.config.Limit, s.config.LimitBurst),
|
||||
time.Now(),
|
||||
}
|
||||
s.visitors[ip] = v
|
||||
return v
|
||||
s.visitors[ip] = newVisitor(s.config)
|
||||
return s.visitors[ip]
|
||||
}
|
||||
v.seen = time.Now()
|
||||
return v
|
||||
|
||||
@@ -6,12 +6,13 @@ html, body {
|
||||
font-size: 1.1em;
|
||||
}
|
||||
|
||||
a {
|
||||
color: #39005a;
|
||||
a, a:visited {
|
||||
color: #3a9784;
|
||||
}
|
||||
|
||||
a:hover {
|
||||
text-decoration: none;
|
||||
color: #317f6f;
|
||||
}
|
||||
|
||||
h1 {
|
||||
@@ -20,7 +21,6 @@ h1 {
|
||||
font-size: 2.5em;
|
||||
}
|
||||
|
||||
|
||||
h2 {
|
||||
margin-top: 20px;
|
||||
margin-bottom: 5px;
|
||||
|
||||
Binary file not shown.
|
Before Width: | Height: | Size: 1.2 KiB After Width: | Height: | Size: 5.0 KiB |
BIN
server/static/img/ntfy.png
Normal file
BIN
server/static/img/ntfy.png
Normal file
Binary file not shown.
|
After Width: | Height: | Size: 5.2 KiB |
@@ -1,7 +1,6 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"log"
|
||||
"math/rand"
|
||||
"sync"
|
||||
@@ -12,11 +11,8 @@ import (
|
||||
// can publish a message
|
||||
type topic struct {
|
||||
id string
|
||||
subscribers map[int]subscriber
|
||||
messages []*message
|
||||
last time.Time
|
||||
ctx context.Context
|
||||
cancel context.CancelFunc
|
||||
subscribers map[int]subscriber
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
@@ -24,14 +20,11 @@ type topic struct {
|
||||
type subscriber func(msg *message) error
|
||||
|
||||
// newTopic creates a new topic
|
||||
func newTopic(id string) *topic {
|
||||
ctx, cancel := context.WithCancel(context.Background())
|
||||
func newTopic(id string, last time.Time) *topic {
|
||||
return &topic{
|
||||
id: id,
|
||||
last: last,
|
||||
subscribers: make(map[int]subscriber),
|
||||
last: time.Now(),
|
||||
ctx: ctx,
|
||||
cancel: cancel,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -54,7 +47,6 @@ func (t *topic) Publish(m *message) error {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
t.last = time.Now()
|
||||
t.messages = append(t.messages, m)
|
||||
for _, s := range t.subscribers {
|
||||
if err := s(m); err != nil {
|
||||
log.Printf("error publishing message to subscriber")
|
||||
@@ -63,38 +55,8 @@ func (t *topic) Publish(m *message) error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (t *topic) Messages(since time.Time) []*message {
|
||||
func (t *topic) Subscribers() int {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
messages := make([]*message, 0) // copy!
|
||||
for _, m := range t.messages {
|
||||
msgTime := time.Unix(m.Time, 0)
|
||||
if msgTime == since || msgTime.After(since) {
|
||||
messages = append(messages, m)
|
||||
}
|
||||
}
|
||||
return messages
|
||||
}
|
||||
|
||||
func (t *topic) Prune(keep time.Duration) {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
for i, m := range t.messages {
|
||||
msgTime := time.Unix(m.Time, 0)
|
||||
if time.Since(msgTime) < keep {
|
||||
t.messages = t.messages[i:]
|
||||
return
|
||||
}
|
||||
}
|
||||
t.messages = make([]*message, 0)
|
||||
}
|
||||
|
||||
func (t *topic) Stats() (subscribers int, messages int) {
|
||||
t.mu.Lock()
|
||||
defer t.mu.Unlock()
|
||||
return len(t.subscribers), len(t.messages)
|
||||
}
|
||||
|
||||
func (t *topic) Close() {
|
||||
t.cancel()
|
||||
return len(t.subscribers)
|
||||
}
|
||||
|
||||
65
server/visitor.go
Normal file
65
server/visitor.go
Normal file
@@ -0,0 +1,65 @@
|
||||
package server
|
||||
|
||||
import (
|
||||
"golang.org/x/time/rate"
|
||||
"heckel.io/ntfy/config"
|
||||
"heckel.io/ntfy/util"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
visitorExpungeAfter = 30 * time.Minute
|
||||
)
|
||||
|
||||
// visitor represents an API user, and its associated rate.Limiter used for rate limiting
|
||||
type visitor struct {
|
||||
config *config.Config
|
||||
limiter *rate.Limiter
|
||||
subscriptions *util.Limiter
|
||||
seen time.Time
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
func newVisitor(conf *config.Config) *visitor {
|
||||
return &visitor{
|
||||
config: conf,
|
||||
limiter: rate.NewLimiter(conf.VisitorRequestLimit, conf.VisitorRequestLimitBurst),
|
||||
subscriptions: util.NewLimiter(int64(conf.VisitorSubscriptionLimit)),
|
||||
seen: time.Now(),
|
||||
}
|
||||
}
|
||||
|
||||
func (v *visitor) RequestAllowed() error {
|
||||
if !v.limiter.Allow() {
|
||||
return errHTTPTooManyRequests
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (v *visitor) AddSubscription() error {
|
||||
v.mu.Lock()
|
||||
defer v.mu.Unlock()
|
||||
if err := v.subscriptions.Add(1); err != nil {
|
||||
return errHTTPTooManyRequests
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (v *visitor) RemoveSubscription() {
|
||||
v.mu.Lock()
|
||||
defer v.mu.Unlock()
|
||||
v.subscriptions.Sub(1)
|
||||
}
|
||||
|
||||
func (v *visitor) Keepalive() {
|
||||
v.mu.Lock()
|
||||
defer v.mu.Unlock()
|
||||
v.seen = time.Now()
|
||||
}
|
||||
|
||||
func (v *visitor) Stale() bool {
|
||||
v.mu.Lock()
|
||||
defer v.mu.Unlock()
|
||||
return time.Since(v.seen) > visitorExpungeAfter
|
||||
}
|
||||
65
util/limit.go
Normal file
65
util/limit.go
Normal file
@@ -0,0 +1,65 @@
|
||||
package util
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"sync"
|
||||
)
|
||||
|
||||
// ErrLimitReached is the error returned by the Limiter and LimitWriter when the predefined limit has been reached
|
||||
var ErrLimitReached = errors.New("limit reached")
|
||||
|
||||
// Limiter is a helper that allows adding values up to a well-defined limit. Once the limit is reached
|
||||
// ErrLimitReached will be returned. Limiter may be used by multiple goroutines.
|
||||
type Limiter struct {
|
||||
value int64
|
||||
limit int64
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
// NewLimiter creates a new Limiter
|
||||
func NewLimiter(limit int64) *Limiter {
|
||||
return &Limiter{
|
||||
limit: limit,
|
||||
}
|
||||
}
|
||||
|
||||
// Add adds n to the limiters internal value, but only if the limit has not been reached. If the limit would be
|
||||
// exceeded after adding n, ErrLimitReached is returned.
|
||||
func (l *Limiter) Add(n int64) error {
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
if l.limit == 0 {
|
||||
l.value += n
|
||||
return nil
|
||||
} else if l.value+n <= l.limit {
|
||||
l.value += n
|
||||
return nil
|
||||
} else {
|
||||
return ErrLimitReached
|
||||
}
|
||||
}
|
||||
|
||||
// Sub subtracts a value from the limiters internal value
|
||||
func (l *Limiter) Sub(n int64) {
|
||||
l.Add(-n)
|
||||
}
|
||||
|
||||
// Set sets the value of the limiter to n. This function ignores the limit. It is meant to set the value
|
||||
// based on reality.
|
||||
func (l *Limiter) Set(n int64) {
|
||||
l.mu.Lock()
|
||||
l.value = n
|
||||
l.mu.Unlock()
|
||||
}
|
||||
|
||||
// Value returns the internal value of the limiter
|
||||
func (l *Limiter) Value() int64 {
|
||||
l.mu.Lock()
|
||||
defer l.mu.Unlock()
|
||||
return l.value
|
||||
}
|
||||
|
||||
// Limit returns the defined limit
|
||||
func (l *Limiter) Limit() int64 {
|
||||
return l.limit
|
||||
}
|
||||
Reference in New Issue
Block a user