peter/websub-hub
1
0
Fork 0
Code Issues 1 Pull Requests Releases Activity

Compare commits

base: peter:7aa64716964292d86b98286c135e684a8c710f70
Branches Tags
peter:master
..
compare: peter:280853054060f611f0bcf30abef5b7340c261f3f
Branches Tags
peter:master

No commits in common. "7aa64716964292d86b98286c135e684a8c710f70" and "280853054060f611f0bcf30abef5b7340c261f3f" have entirely different histories.
7aa6471696 ... 2808530540

1 changed files with 36 additions and 57 deletions
Show Stats Expand all files Collapse all files

93
cmd/hubserver/main.go
View File

@ -13,7 +13,6 @@ import (
"os"
"strconv"
"strings"
"sync"
"time"
)
@ -44,11 +43,8 @@ type Stat struct {
}
type subscriptionHandler struct {
LockSubs sync.Mutex
Subscribers map[string][]Subscriber
LockStats sync.Mutex
Stats map[string]Stat
Stats map[string]Stat
}
func (handler *subscriptionHandler) handlePublish(w http.ResponseWriter, r *http.Request) error {
@ -70,12 +66,12 @@ func (handler *subscriptionHandler) handlePublish(w http.ResponseWriter, r *http
return err
}
handler.incStat("published")
// handler.incStat(fmt.Sprintf("publish.%s", topic))
if subs, e := handler.Subscribers[topic]; e {
for _, sub := range subs {
handler.incStat("publish.post")
// handler.incStat(fmt.Sprintf("publish.post.%s.%s", topic, sub.Callback))
log.Printf("publish: creating post to %s\n", sub.Callback)
postReq, err := http.NewRequest("POST", sub.Callback, strings.NewReader(string(feedContent)))
if err != nil {
@ -151,7 +147,6 @@ func (handler *subscriptionHandler) handleUnsubscription(w http.ResponseWriter,
func (handler *subscriptionHandler) handleSubscription(w http.ResponseWriter, r *http.Request) error {
log.Printf("subscription request received: %s %#v\n", r.URL.String(), r.Form)
callback := r.Form.Get("hub.callback")
topic := r.Form.Get("hub.topic")
secret := r.Form.Get("hub.secret")
@ -257,27 +252,23 @@ func (handler *subscriptionHandler) addSubscriberCallback(topic string, subscrib
}
func (handler *subscriptionHandler) incStat(name string) {
handler.LockStats.Lock()
defer handler.LockStats.Unlock()
if v, e := handler.Stats[name]; e {
handler.Stats[name] = Stat{LastUpdate: time.Now(), Updates: v.Updates + 1}
} else {
handler.Stats[name] = Stat{LastUpdate: time.Now(), Updates: 1}
}
handler.saveStats()
// if v, e := handler.Stats[name]; e {
// handler.Stats[name] = Stat{LastUpdate: time.Now(), Updates: v.Updates + 1}
// } else {
// handler.Stats[name] = Stat{LastUpdate: time.Now(), Updates: 1}
// }
// handler.saveStats()
}
func (handler *subscriptionHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
if r.Method == http.MethodGet {
_, _ = fmt.Fprintln(w, "WebSub hub")
fmt.Fprintln(w, "WebSub hub")
if r.URL.Query().Get("debug") == "1" {
handler.incStat("http.index.debug")
// handler.incStat("http.index.debug")
enc := json.NewEncoder(w)
enc.SetIndent("", " ")
_ = enc.Encode(handler.Subscribers)
_ = enc.Encode(handler.Stats)
enc.Encode(handler.Subscribers)
//enc.Encode(handler.Stats)
}
return
}
@ -296,14 +287,14 @@ func (handler *subscriptionHandler) ServeHTTP(w http.ResponseWriter, r *http.Req
mode := r.Form.Get("hub.mode")
if mode == "subscribe" {
err = handler.handleSubscription(w, r)
handler.handleSubscription(w, r)
return
} else if mode == "unsubscribe" {
err = handler.handleUnsubscription(w, r)
handler.handleUnsubscription(w, r)
return
} else if mode == "publish" {
log.Println("hub.mode=publish received")
err = handler.handlePublish(w, r)
handler.handlePublish(w, r)
return
} else {
http.Error(w, "Unknown hub.mode", 400)
@ -312,9 +303,6 @@ func (handler *subscriptionHandler) ServeHTTP(w http.ResponseWriter, r *http.Req
}
func (handler *subscriptionHandler) loadStats() error {
handler.LockStats.Lock()
defer handler.LockStats.Unlock()
file, err := os.Open("./stats.json")
if err != nil {
if os.IsNotExist(err) {
@ -328,25 +316,7 @@ func (handler *subscriptionHandler) loadStats() error {
return err
}
func (handler *subscriptionHandler) saveStats() error {
handler.LockStats.Lock()
defer handler.LockStats.Unlock()
file, err := os.Create("./stats.json")
if err != nil {
return err
}
defer file.Close()
dec := json.NewEncoder(file)
dec.SetIndent("", " ")
err = dec.Encode(&handler.Stats)
return err
}
func (handler *subscriptionHandler) loadSubscriptions() error {
handler.LockSubs.Lock()
defer handler.LockSubs.Unlock()
file, err := os.Open("./subscription.json")
if err != nil {
if os.IsNotExist(err) {
@ -360,11 +330,28 @@ func (handler *subscriptionHandler) loadSubscriptions() error {
return err
}
func (handler *subscriptionHandler) load() error {
err := handler.loadSubscriptions()
if err != nil {
return err
}
return nil //handler.loadStats()
}
func (handler *subscriptionHandler) saveStats() error {
// file, err := os.Create("./stats.json")
// if err != nil {
// return err
// }
// defer file.Close()
// dec := json.NewEncoder(file)
// dec.SetIndent("", " ")
// err = dec.Encode(&handler.Stats)
// return err
return nil
}
func (handler *subscriptionHandler) saveSubscriptions() error {
handler.LockSubs.Lock()
defer handler.LockSubs.Unlock()
file, err := os.Create("./subscription.json")
if err != nil {
return err
@ -376,14 +363,6 @@ func (handler *subscriptionHandler) saveSubscriptions() error {
return err
}
func (handler *subscriptionHandler) load() error {
err := handler.loadSubscriptions()
if err != nil {
return err
}
return handler.loadStats()
}
func (handler *subscriptionHandler) save() error {
handler.saveSubscriptions()
return handler.saveStats()