package main import ( "errors" "fmt" "log" "net/http" "strconv" "time" "git.lattuga.net/boyska/circolog" "git.lattuga.net/boyska/circolog/data" "git.lattuga.net/boyska/circolog/formatter" "github.com/gorilla/websocket" ) func setupHTTP(hub circolog.Hub) *http.ServeMux { mux := http.NewServeMux() mux.HandleFunc("/", getHTTPHandler(hub)) mux.HandleFunc("/ws", getWSHandler(hub)) return mux } func parseParameterL(r *http.Request) (int, error) { var requestMessageLen int = -1 var err error if reqL, ok := r.Form["l"]; ok { if len(reqL) == 1 { requestMessageLen, err = strconv.Atoi(reqL[0]) if err != nil { return 0, err } if requestMessageLen <= 0 { return 0, errors.New("malformed request") } } else { return 0, errors.New("malformed request") } } return requestMessageLen, nil } func parseParameters(r *http.Request) (circolog.ClientOptions, error) { var opts circolog.ClientOptions err := r.ParseForm() if err != nil { log.Println("error parsing http request", err) return opts, err } l, err := parseParameterL(r) if err != nil { return opts, err } opts.BacklogLength = l return opts, err } type renderOptions struct { // those are options relevant to the rendered (that is, the HTTP side of circologd) Format formatter.Format } func parseRenderParameters(r *http.Request) (renderOptions, error) { opts := renderOptions{} err := r.ParseForm() if err != nil { log.Println("error parsing http request", err) return opts, err } if val, ok := r.Form["fmt"]; ok { if len(val) != 1 { return opts, errors.New("Format repeated multiple times") } err := opts.Format.Set(val[0]) if err != nil { return opts, err } } return opts, nil } func getHTTPHandler(hub circolog.Hub) http.HandlerFunc { return func(w http.ResponseWriter, r *http.Request) { // Looking for known parameter in the request render_opts, err := parseRenderParameters(r) if err != nil { log.Println("Error parsing:", err) w.WriteHeader(400) fmt.Fprintln(w, err) return } opts, err := parseParameters(r) if err != nil { log.Println("Error on request parameter \"l\":", err) w.WriteHeader(400) fmt.Fprintln(w, err) return } opts.Nofollow = true client := circolog.Client{ Messages: make(chan data.Message, 20), Options: opts, } hub.Register <- client for x := range client.Messages { if err := render_opts.Format.WriteFormatted(w, x); err == nil { if render_opts.Format != formatter.FormatJSON { // bleah w.Write([]byte("\n")) } } } } } func getWSHandler(hub circolog.Hub) http.HandlerFunc { var upgrader = websocket.Upgrader{ ReadBufferSize: 1024, WriteBufferSize: 1024, } return func(w http.ResponseWriter, r *http.Request) { render_opts, err := parseRenderParameters(r) if err != nil { log.Println("Error parsing:", err) w.WriteHeader(400) fmt.Fprintln(w, err) return } opts, err := parseParameters(r) if err != nil { log.Println("Error on request parameter \"l\":", err) w.WriteHeader(400) fmt.Fprintln(w, err) return } opts.Nofollow = false conn, err := upgrader.Upgrade(w, r, nil) if err != nil { return } client := circolog.Client{ Messages: make(chan data.Message, 20), Options: opts, } hub.Register <- client // Allow collection of memory referenced by the caller by doing all work in // new goroutines. go func(conn *websocket.Conn, c circolog.Client) { defer func() { hub.Unregister <- c conn.Close() }() go func() { for { _, _, err := conn.ReadMessage() if err != nil { conn.Close() return } } }() for { select { case message, ok := <-c.Messages: conn.SetWriteDeadline(time.Now().Add(10 * time.Second)) if !ok { // The hub closed the channel. conn.WriteMessage(websocket.CloseMessage, []byte{}) return } w, err := conn.NextWriter(websocket.TextMessage) if err != nil { return } render_opts.Format.WriteFormatted(w, message) if err := w.Close(); err != nil { return } // TODO: ticker/ping } } }(conn, client) } }