Micro-service for file storage and processing written in Go
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 

149 lines
3.7 KiB

package main
import (
"bytes"
"encoding/json"
"github.com/geplauder/lithium/auth"
"github.com/geplauder/lithium/pipelines"
"github.com/geplauder/lithium/settings"
"github.com/geplauder/lithium/storage"
"github.com/gorilla/mux"
"github.com/spf13/afero"
"io"
"net/http"
"strings"
)
const Name string = "Lithium"
const Version string = "0.1.0"
type Metadata struct {
Name string `json:"name"`
Version string `json:"version"`
}
func PipelineHandler(pipeline pipelines.IPipeline, storageProvider storage.IStorageProvider, w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
err := json.NewEncoder(w).Encode(pipeline)
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
}
}
func IndexHandler(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
err := json.NewEncoder(w).Encode(Metadata{Name, Version})
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
}
}
func writeError(w http.ResponseWriter, status int, errStr string) {
w.WriteHeader(status)
json.NewEncoder(w).Encode(struct {
Error string `json:"error"`
}{errStr})
}
func UploadHandler(w http.ResponseWriter, r *http.Request, pipes []pipelines.IPipeline, storageProvider storage.IStorageProvider) {
// open file handler
formFile, handler, err := r.FormFile("file")
if err != nil {
writeError(w, http.StatusUnprocessableEntity, err.Error())
return
}
defer formFile.Close()
// check pipelines form param
formPipelines := strings.Split(r.FormValue("pipelines"), ",")
if len(formPipelines) == 0 {
writeError(w, http.StatusUnprocessableEntity, "pipeline parameter missing")
return
}
bucket := r.FormValue("bucket")
if bucket == "" {
writeError(w, http.StatusUnprocessableEntity, "bucket parameter missing")
return
}
// open file
file, err := handler.Open()
if err != nil {
writeError(w, http.StatusInternalServerError, "error reading uploaded file")
return
}
defer file.Close()
// read file to buffer
buf := bytes.NewBuffer(nil)
_, err = io.Copy(buf, file)
if err != nil {
writeError(w, http.StatusInternalServerError, "error reading file from buffer")
return
}
// store uploaded file
_, err = storageProvider.StoreRaw(bucket, "source.jpg", buf.Bytes())
if err != nil {
return
}
w.Header().Set("Content-Type", "application/json")
err = json.NewEncoder(w).Encode(struct {
Message string `json:"message"`
}{"ok"})
if err != nil {
w.WriteHeader(http.StatusInternalServerError)
return
}
}
func RegisterRoutes(r *mux.Router, pipelines []pipelines.IPipeline, storageProvider storage.IStorageProvider) {
r.HandleFunc("/", IndexHandler).Methods("GET")
r.HandleFunc("/upload", func(w http.ResponseWriter, r *http.Request) {
UploadHandler(w, r, pipelines, storageProvider)
}).Methods("POST")
r.HandleFunc("/pipelines/{pipeline}", func(w http.ResponseWriter, r *http.Request) {
for _, pipeline := range pipelines {
if pipeline.GetSlug() == mux.Vars(r)["pipeline"] {
PipelineHandler(pipeline, storageProvider, w, r)
return
}
}
w.WriteHeader(404)
}).Methods("GET")
}
func main() {
appSettings, err := settings.LoadSettings(afero.NewOsFs())
if err != nil {
panic(err)
}
var storageProvider storage.IStorageProvider
if appSettings.StorageProvider.Type == 0 {
storageProvider = storage.GetFileSystemStorageProvider(appSettings.StorageProvider.BasePath, "")
} else {
panic("Invalid file system provided!")
}
pipes := pipelines.LoadPipelines()
authMiddleware := auth.CreateAuthenticationMiddleware(appSettings.Token)
r := mux.NewRouter()
r.Use(authMiddleware.Middleware)
RegisterRoutes(r, pipes, storageProvider)
err = http.ListenAndServe(appSettings.Endpoint, r)
if err != nil {
panic(err)
}
}