83 lines
1.7 KiB
Go
83 lines
1.7 KiB
Go
|
package server
|
||
|
|
|
||
|
|
import (
|
||
|
|
"encoding/json"
|
||
|
|
"net/http"
|
||
|
|
|
||
|
|
"github.com/gorilla/mux"
|
||
|
|
)
|
||
|
|
|
||
|
|
func NewHTTPServer(addr string) *http.Server {
|
||
|
|
httpsrv := newHttpServer()
|
||
|
|
r := mux.NewRouter()
|
||
|
|
r.HandleFunc("/", httpsrv.handleProduce).Methods("POST")
|
||
|
|
r.HandleFunc("/", httpsrv.handleConsume).Methods("GET")
|
||
|
|
return &http.Server{
|
||
|
|
Addr: addr,
|
||
|
|
Handler: r,
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
type httpServer struct {
|
||
|
|
Log *Log
|
||
|
|
}
|
||
|
|
|
||
|
|
func newHttpServer() *httpServer {
|
||
|
|
return &httpServer{
|
||
|
|
Log: NewLog(),
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
type ProduceRequest struct {
|
||
|
|
Record Record `json:"record"`
|
||
|
|
}
|
||
|
|
type ProduceResponse struct {
|
||
|
|
Offset uint64 `json:"offset"`
|
||
|
|
}
|
||
|
|
type ConsumeRequest struct {
|
||
|
|
Offset uint64 `json:"offset"`
|
||
|
|
}
|
||
|
|
type ConsumeResponse struct {
|
||
|
|
Record Record `json:"record"`
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *httpServer) handleProduce(w http.ResponseWriter, r *http.Request) {
|
||
|
|
var req ProduceRequest
|
||
|
|
err := json.NewDecoder(r.Body).Decode(&req)
|
||
|
|
if err != nil {
|
||
|
|
http.Error(w, err.Error(), http.StatusBadRequest)
|
||
|
|
return
|
||
|
|
}
|
||
|
|
off, err := s.Log.Append(req.Record)
|
||
|
|
if err != nil {
|
||
|
|
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||
|
|
return
|
||
|
|
}
|
||
|
|
res := ProduceResponse{Offset: off}
|
||
|
|
err = json.NewEncoder(w).Encode(res)
|
||
|
|
if err != nil {
|
||
|
|
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||
|
|
return
|
||
|
|
}
|
||
|
|
}
|
||
|
|
|
||
|
|
func (s *httpServer) handleConsume(w http.ResponseWriter, r *http.Request) {
|
||
|
|
var req ConsumeRequest
|
||
|
|
err := json.NewDecoder(r.Body).Decode(&req)
|
||
|
|
if err != nil {
|
||
|
|
http.Error(w, err.Error(), http.StatusBadRequest)
|
||
|
|
return
|
||
|
|
}
|
||
|
|
record, err := s.Log.Read(req.Offset)
|
||
|
|
if err == ErrOffsetNotFound {
|
||
|
|
http.Error(w, err.Error(), http.StatusNotFound)
|
||
|
|
return
|
||
|
|
}
|
||
|
|
res := ConsumeResponse{Record: record}
|
||
|
|
err = json.NewEncoder(w).Encode(res)
|
||
|
|
if err != nil {
|
||
|
|
http.Error(w, err.Error(), http.StatusInternalServerError)
|
||
|
|
return
|
||
|
|
}
|
||
|
|
}
|