153 lines
4.1 KiB
Go
153 lines
4.1 KiB
Go
package web
|
|
|
|
import (
|
|
"encoding/json"
|
|
"fmt"
|
|
"go/weather/internal/storage"
|
|
"go/weather/pkg/headers"
|
|
"go/weather/pkg/logger"
|
|
"io"
|
|
"net/http"
|
|
"time"
|
|
|
|
"github.com/gorilla/mux"
|
|
"go.elastic.co/apm"
|
|
"go.elastic.co/apm/module/apmzap"
|
|
"go.uber.org/zap"
|
|
)
|
|
|
|
const (
|
|
welcomeMessage = "Welcome to the weather API"
|
|
)
|
|
|
|
// WeatherHandler object grouping all handler for http web server
|
|
type WeatherHandler struct {
|
|
baseRouter *mux.Router
|
|
subRouters map[string]*mux.Router
|
|
wlogger *logger.WeatherLogger
|
|
storage *storage.S3Storage
|
|
}
|
|
|
|
func NewWeatherHandler(logger *logger.WeatherLogger, s3 *storage.S3Storage, baseRouter *mux.Router) *WeatherHandler {
|
|
wh := WeatherHandler{
|
|
baseRouter: baseRouter,
|
|
subRouters: make(map[string]*mux.Router, 1),
|
|
wlogger: logger,
|
|
storage: s3,
|
|
}
|
|
return &wh
|
|
}
|
|
|
|
//getOrCreateSubRouterByName return a router by name or create if it doesn't exist yet
|
|
func (wh *WeatherHandler) GetOrCreateSubRouterByName(name, prefix string) *mux.Router {
|
|
if api := wh.subRouters[name]; api == nil {
|
|
wh.subRouters[name] = wh.baseRouter.PathPrefix(prefix).Subrouter()
|
|
}
|
|
return wh.subRouters[name]
|
|
}
|
|
|
|
//RegisterBase base endpoint
|
|
func (wh *WeatherHandler) RegisterBase() {
|
|
wh.baseRouter.HandleFunc("/", func(rw http.ResponseWriter, r *http.Request) {
|
|
rw.Write([]byte(welcomeMessage))
|
|
})
|
|
}
|
|
|
|
//RegisterApi health endpoint
|
|
func (wh *WeatherHandler) RegisterApi(version string) {
|
|
api := wh.GetOrCreateSubRouterByName(apiRouterName, apiPrefix)
|
|
|
|
api.HandleFunc("/health", func(rw http.ResponseWriter, r *http.Request) {
|
|
rw.Header().Add(headers.ContentType, "application/json")
|
|
json.NewEncoder(rw).Encode(map[string]interface{}{
|
|
"ok": true,
|
|
"version": version,
|
|
})
|
|
})
|
|
|
|
api.HandleFunc("/all", func(rw http.ResponseWriter, r *http.Request) {
|
|
span, ctx := apm.StartSpan(r.Context(), "s3GetAllWeather", "custom")
|
|
defer span.End()
|
|
|
|
// application/stream+json
|
|
// application/x-ndjson
|
|
// text/event-stream
|
|
rw.Header().Set(headers.ContentType, "text/plain")
|
|
rw.Header().Set(headers.CacheControl, "no-cache")
|
|
rw.Header().Set(headers.Connection, "keep-alive")
|
|
|
|
// List all objects from a bucket-name with a matching prefix.
|
|
wh.ResponseStream(rw, r, wh.storage.GetAll(ctx), func(enc *json.Encoder, obj storage.Streamable) {
|
|
if o, ok := obj.(storage.WeatherFile); ok {
|
|
fmt.Printf("name : %s\n", o.Name)
|
|
}
|
|
|
|
err := enc.Encode(obj)
|
|
if err != nil {
|
|
wh.wlogger.Sugar().Errorf("All encoding error %v\n", obj)
|
|
}
|
|
})
|
|
})
|
|
|
|
api.HandleFunc("/at/{atDate}", func(rw http.ResponseWriter, r *http.Request) {
|
|
span, ctx := apm.StartSpan(r.Context(), "s3GetWeatherInfo", "custom")
|
|
defer span.End()
|
|
|
|
vars := mux.Vars(r)
|
|
atDate, err := time.Parse(time.RFC3339, vars["atDate"])
|
|
if err != nil {
|
|
wh.wlogger.Error(fmt.Sprintf("AtDate parse invalid format %s", vars["atDate"]), zap.Error(err))
|
|
rw.WriteHeader(http.StatusBadRequest)
|
|
} else {
|
|
wh.wlogger.Debug("AtDate GetObject", append(
|
|
apmzap.TraceContext(ctx),
|
|
zap.String("name", fmt.Sprintf("%s.json", atDate.Format(time.RFC3339))),
|
|
)...)
|
|
|
|
reader := wh.storage.GetAtDate(ctx, atDate)
|
|
if reader == nil {
|
|
rw.WriteHeader(http.StatusInternalServerError)
|
|
}
|
|
defer reader.Close()
|
|
|
|
rw.Header().Add(headers.ContentType, "application/json")
|
|
nb, err := io.Copy(rw, reader)
|
|
if err != nil {
|
|
wh.wlogger.Error("AtDate write response failed", zap.Error(err))
|
|
rw.WriteHeader(http.StatusInternalServerError)
|
|
} else {
|
|
wh.wlogger.Debugf("AtDate write %d", []interface{}{nb}, apmzap.TraceContext(ctx)...)
|
|
}
|
|
}
|
|
})
|
|
}
|
|
|
|
func (wh *WeatherHandler) ResponseStream(
|
|
w http.ResponseWriter, r *http.Request, stream <-chan storage.Streamable, f func(enc *json.Encoder, obj storage.Streamable),
|
|
) {
|
|
flusher, ok := w.(http.Flusher)
|
|
if !ok {
|
|
wh.wlogger.Debug("Flusher cast error")
|
|
http.NotFound(w, r)
|
|
return
|
|
}
|
|
|
|
w.Header().Set(headers.TransferEncoding, "chunked")
|
|
w.WriteHeader(http.StatusOK)
|
|
flusher.Flush()
|
|
|
|
enc := json.NewEncoder(w)
|
|
|
|
cn := r.Context()
|
|
for object := range stream {
|
|
select {
|
|
case <-cn.Done():
|
|
wh.wlogger.Debug("Client stopped listening", zap.Error(cn.Err()))
|
|
return
|
|
default:
|
|
f(enc, object)
|
|
flusher.Flush()
|
|
}
|
|
}
|
|
}
|