circolog/cmd/circologd/http_log.go

package main

import (
	"errors"
	"fmt"
	"log"
	"net/http"
	"strconv"
	"time"

	"git.lattuga.net/boyska/circolog"
	"github.com/gorilla/websocket"
	"gopkg.in/mcuadros/go-syslog.v2/format"
)

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 renderFormat
}

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")
		}
		format, err := parseRenderFormat(val[0])
		if err != nil {
			return opts, err
		}
		opts.Format = format
	}
	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 format.LogParts, 20),
			Options:  opts,
		}
		hub.Register <- client

		for x := range client.Messages {
			writeFormatted(w, render_opts.Format, x)
			if render_opts.Format != 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 format.LogParts, 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()
			}()
			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
					}
					writeFormatted(w, render_opts.Format, message)

					if err := w.Close(); err != nil {
						return
					}
					// TODO: ticker/ping
				}
			}
		}(conn, client)
	}
}