influxdb/messaging/handler.go

package messaging

import (
	"io/ioutil"
	"net/http"
	"strconv"
	"strings"

	"github.com/influxdb/influxdb/raft"
)

// Handler represents an HTTP handler by the broker.
type Handler struct {
	raftHandler *raft.HTTPHandler
	broker      *Broker
}

// NewHandler returns a new instance of Handler.
func NewHandler(b *Broker) *Handler {
	return &Handler{
		raftHandler: raft.NewHTTPHandler(b.log),
		broker:      b,
	}
}

// Broker returns the broker on the handler.
func (h *Handler) Broker() *Broker { return h.broker }

// ServeHTTP serves an HTTP request.
func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	// Delegate raft requests to its own handler.
	if strings.HasPrefix(r.URL.Path, "/raft") {
		h.raftHandler.ServeHTTP(w, r)
		return
	}

	// Route all InfluxDB broker requests.
	switch r.URL.Path {
	case "/messages":
		if r.Method == "GET" {
			h.stream(w, r)
		} else if r.Method == "POST" {
			h.publish(w, r)
		} else {
			http.Error(w, "Method Not Allowed", http.StatusMethodNotAllowed)
		}
	default:
		http.NotFound(w, r)
	}
}

// connects the requestor as the replica's writer.
func (h *Handler) stream(w http.ResponseWriter, r *http.Request) {
	// Retrieve the replica name.
	name := r.URL.Query().Get("name")
	if name == "" {
		h.error(w, ErrReplicaNameRequired, http.StatusBadRequest)
		return
	}

	// Find the replica on the broker.
	replica := h.broker.Replica(name)
	if replica == nil {
		h.error(w, ErrReplicaNotFound, http.StatusNotFound)
		return
	}

	// Connect the response writer to the replica.
	// This will block until the replica is closed or a new writer connects.
	_, _ = replica.WriteTo(w)
}

// publishes a message to the broker.
func (h *Handler) publish(w http.ResponseWriter, r *http.Request) {
	m := &Message{}

	// Read the message type.
	if n, err := strconv.ParseUint(r.URL.Query().Get("type"), 10, 16); err != nil {
		h.error(w, ErrMessageTypeRequired, http.StatusBadRequest)
		return
	} else {
		m.Type = MessageType(n)
	}

	// Read the topic ID.
	if n, err := strconv.ParseUint(r.URL.Query().Get("topicID"), 10, 32); err != nil {
		h.error(w, ErrTopicRequired, http.StatusBadRequest)
		return
	} else {
		m.TopicID = uint64(n)
	}

	// Read the request body.
	data, err := ioutil.ReadAll(r.Body)
	if err != nil {
		h.error(w, err, http.StatusInternalServerError)
		return
	}
	m.Data = data

	// Publish message to the broker.
	index, err := h.broker.Publish(m)
	if err != nil {
		h.error(w, err, http.StatusInternalServerError)
		return
	}

	// Return index.
	w.Header().Set("X-Broker-Index", strconv.FormatUint(index, 10))
}

// error writes an error to the client and sets the status code.
func (h *Handler) error(w http.ResponseWriter, err error, code int) {
	s := err.Error()
	w.Header().Set("X-Broker-Error", s)
	http.Error(w, s, code)
}