localhots/burlesque
1
0
Fork 0
Code Releases Activity

Bump

Browse Source
This commit is contained in:
Gregory Eremin 2014-09-11 14:27:55 +04:00
parent 3adaa805eb
commit d417fa653f
3 changed files with 87 additions and 110 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
config.go
View File

@ -1,18 +0,0 @@
package main
import (
"flag"
)
var (
config struct {
storage string
port int
}
)
func setupConfig() {
flag.StringVar(&config.storage, "storage", "-", "Kyoto Cabinet storage path (e.g. burlesque.kch#dfunit=8#msiz=512M)")
flag.IntVar(&config.port, "port", 4401, "Server HTTP port")
flag.Parse()
}

43
main.go
View File

@ -1,39 +1,48 @@
package main
import (
"flag"
"os"
"os/signal"
"syscall"
"github.com/KosyanMedia/burlesque/hub"
"github.com/KosyanMedia/burlesque/storage"
)
const (
version = "0.1.3"
version = "0.2.0"
)
func handleShutdown() {
var (
theStorage *storage.Storage
theHub *hub.Hub
config struct {
storage string
port int
}
)
func main() {
flag.StringVar(&config.storage, "storage", "-", "Kyoto Cabinet storage path (e.g. burlesque.kch#dfunit=8#msiz=512M)")
flag.IntVar(&config.port, "port", 4401, "Server HTTP port")
flag.Parse()
theStorage, err := storage.New(config.storage)
if err != nil {
panic(err)
}
theHub = hub.New(theStorage)
ch := make(chan os.Signal)
signal.Notify(ch, os.Interrupt, os.Kill, syscall.SIGTERM, syscall.SIGINT)
go func() {
<-ch
saveState()
log("State successfully persisted")
closeStorage()
log("Stopped")
theStorage.Close()
os.Exit(0)
}()
}
func main() {
setupConfig()
setupLogging()
setupStorage()
setupServer()
handleShutdown()
loadState()
go keepStatePersisted()
startServer()
}

140
server.go
View File

@ -1,124 +1,110 @@
package main
import (
"encoding/json"
"fmt"
"io/ioutil"
"net/http"
"runtime"
"strconv"
"strings"
"github.com/KosyanMedia/burlesque/hub"
)
func startServer() {
port := fmt.Sprintf(":%d", config.port)
err := http.ListenAndServe(port, nil)
if err != nil {
http.HandleFunc("/status", statusHandler)
http.HandleFunc("/debug", debugHandler)
http.HandleFunc("/publish", pubHandler)
http.HandleFunc("/subscribe", subHandler)
if err := http.ListenAndServe(fmt.Sprintf(":%d", config.port), nil); err != nil {
alert(err, "Error starting server on port %d", config.port)
}
}
func statusHandler(w http.ResponseWriter, r *http.Request) {
info := make(map[string]map[string]uint)
// info := make(map[string]map[string]uint)
for _, q := range queues {
info[q.name] = map[string]uint{
"messages": q.counter.distance(),
"subscriptions": 0,
}
}
// for _, q := range queues {
// info[q.name] = map[string]uint{
// // "messages": q.counter.distance(),
// "subscriptions": 0,
// }
// }
for _, r := range pool.requests {
for _, q := range r.queues {
info[q]["subscriptions"]++
}
}
// for _, r := range pool.requests {
// for _, q := range r.queues {
// info[q]["subscriptions"]++
// }
// }
jsn, _ := json.Marshal(info)
w.Write(jsn)
// jsn, _ := json.Marshal(info)
// w.Write(jsn)
}
func debugHandler(w http.ResponseWriter, r *http.Request) {
info := make(map[string]interface{})
info["version"] = version
info["goroutines"] = runtime.NumGoroutine()
// info := make(map[string]interface{})
// info["version"] = version
// info["goroutines"] = runtime.NumGoroutine()
s, err := storage.Status()
if err != nil {
alert(err, "Failed to get Kyoto Cabinet status")
}
s = s[:len(s)-1] // Removing trailing new line
// s, err := storage.Status()
// if err != nil {
// alert(err, "Failed to get Kyoto Cabinet status")
// }
// s = s[:len(s)-1] // Removing trailing new line
ks := make(map[string]interface{})
tokens := strings.Split(s, "\n")
for _, t := range tokens {
tt := strings.Split(t, "\t")
num, err := strconv.Atoi(tt[1])
if err != nil {
ks[tt[0]] = tt[1]
} else {
ks[tt[0]] = num
}
}
info["kyoto_cabinet"] = ks
// ks := make(map[string]interface{})
// tokens := strings.Split(s, "\n")
// for _, t := range tokens {
// tt := strings.Split(t, "\t")
// num, err := strconv.Atoi(tt[1])
// if err != nil {
// ks[tt[0]] = tt[1]
// } else {
// ks[tt[0]] = num
// }
// }
// info["kyoto_cabinet"] = ks
jsn, _ := json.Marshal(info)
w.Write(jsn)
// jsn, _ := json.Marshal(info)
// w.Write(jsn)
}
func publishHandler(w http.ResponseWriter, r *http.Request) {
func pubHandler(w http.ResponseWriter, r *http.Request) {
defer r.Body.Close()
msg, _ := ioutil.ReadAll(r.Body)
if len(msg) == 0 {
msg = message(r.FormValue("msg"))
msg = []byte(r.FormValue("msg"))
}
queue := r.FormValue("queue")
qname := r.FormValue("queue")
ok := registerPublication(qname, msg)
if ok {
if ok := theHub.Pub(queue, msg); ok {
w.Write([]byte("OK"))
} else {
http.Error(w, "FAIL", 500)
}
}
func subscriptionHandler(w http.ResponseWriter, r *http.Request) {
rch := make(chan response)
abort := make(chan bool, 1)
req := &request{
queues: strings.Split(r.FormValue("queues"), ","),
responseCh: rch,
abort: abort,
}
go registerSubscription(req)
func subHandler(w http.ResponseWriter, r *http.Request) {
result := make(chan hub.Result)
queues := strings.Split(r.FormValue("queues"), ",")
sub := hub.NewSubscription(queues, result)
disconnected := w.(http.CloseNotifier).CloseNotify()
finished := make(chan bool)
finished := make(chan struct{})
go func() {
select {
case <-disconnected:
close(rch)
abort <- true
sub.Close()
close(finished)
case <-finished:
}
req.purge()
}()
defer sub.Close()
res, ok := <-rch
if !ok {
return
}
w.Header().Set("Queue", res.queue)
w.Write(res.message)
finished <- true
}
func setupServer() {
http.HandleFunc("/status", statusHandler)
http.HandleFunc("/debug", debugHandler)
http.HandleFunc("/publish", publishHandler)
http.HandleFunc("/subscribe", subscriptionHandler)
theHub.Sub(sub)
res := <-result
w.Header().Set("Queue", res.Queue)
w.Write(res.Message)
finished <- struct{}{}
}