2014-11-29 23:42:22 +01:00
|
|
|
package gateway
|
2014-11-03 00:56:33 +01:00
|
|
|
|
|
|
|
import (
|
2014-11-07 03:34:46 +01:00
|
|
|
"errors"
|
2014-11-03 00:56:33 +01:00
|
|
|
"fmt"
|
2014-11-07 04:32:35 +01:00
|
|
|
"io/ioutil"
|
2014-11-03 00:56:33 +01:00
|
|
|
"net/http"
|
2014-11-15 02:22:50 +01:00
|
|
|
"path"
|
2014-11-29 23:42:22 +01:00
|
|
|
|
|
|
|
"github.com/gorilla/mux"
|
2014-12-08 23:25:57 +01:00
|
|
|
"github.com/minio-io/minio/pkgs/storage/fsstorage"
|
2014-11-29 23:42:22 +01:00
|
|
|
"github.com/tchap/go-patricia/patricia"
|
2014-11-03 00:56:33 +01:00
|
|
|
)
|
|
|
|
|
2014-11-07 07:45:27 +01:00
|
|
|
// Stores system configuration, populated from CLI or test runner
|
|
|
|
type GatewayConfig struct {
|
2014-11-08 04:07:16 +01:00
|
|
|
StorageDriver StorageDriver
|
2014-11-14 07:18:44 +01:00
|
|
|
BucketDriver BucketDriver
|
2014-11-08 04:07:16 +01:00
|
|
|
requestBucketChan chan BucketRequest
|
2014-11-15 02:22:50 +01:00
|
|
|
dataDir string
|
2014-11-07 07:45:27 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
// Message for requesting a bucket
|
2014-11-07 03:34:46 +01:00
|
|
|
type BucketRequest struct {
|
|
|
|
name string
|
|
|
|
context Context
|
|
|
|
callback chan Bucket
|
|
|
|
}
|
|
|
|
|
2014-11-07 07:45:27 +01:00
|
|
|
// Context interface for security and session information
|
2014-11-07 03:34:46 +01:00
|
|
|
type Context interface{}
|
|
|
|
|
2014-11-07 07:45:27 +01:00
|
|
|
// Bucket definition
|
2014-11-07 03:34:46 +01:00
|
|
|
type Bucket interface {
|
|
|
|
GetName(Context) string
|
|
|
|
Get(Context, string) ([]byte, error)
|
|
|
|
Put(Context, string, []byte) error
|
|
|
|
}
|
|
|
|
|
2014-11-14 07:18:44 +01:00
|
|
|
// Bucket driver function, should read from a channel and respond through callback channels
|
|
|
|
type BucketDriver func(config GatewayConfig)
|
|
|
|
|
2014-11-07 07:45:27 +01:00
|
|
|
// Storage driver function, should read from a channel and respond through callback channels
|
2014-11-15 02:22:50 +01:00
|
|
|
type StorageDriver func(bucket string, input chan ObjectRequest, config GatewayConfig)
|
2014-11-07 04:32:35 +01:00
|
|
|
|
2014-11-07 07:45:27 +01:00
|
|
|
// TODO remove when building real context
|
|
|
|
type fakeContext struct{}
|
2014-11-03 00:56:33 +01:00
|
|
|
|
2014-11-07 04:32:35 +01:00
|
|
|
type GatewayGetHandler struct {
|
2014-11-08 04:07:16 +01:00
|
|
|
config GatewayConfig
|
2014-11-07 04:32:35 +01:00
|
|
|
}
|
|
|
|
|
2014-11-07 07:45:27 +01:00
|
|
|
// GET requests server
|
2014-11-07 04:32:35 +01:00
|
|
|
func (handler GatewayGetHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
|
2014-11-03 04:55:01 +01:00
|
|
|
vars := mux.Vars(req)
|
2014-11-07 04:32:35 +01:00
|
|
|
bucketName := vars["bucket"]
|
|
|
|
path := vars["path"]
|
|
|
|
context := fakeContext{}
|
|
|
|
callback := make(chan Bucket)
|
2014-11-08 04:07:16 +01:00
|
|
|
handler.config.requestBucketChan <- BucketRequest{
|
2014-11-07 04:32:35 +01:00
|
|
|
name: bucketName,
|
|
|
|
context: context,
|
|
|
|
callback: callback,
|
|
|
|
}
|
|
|
|
bucket := <-callback
|
|
|
|
object, err := bucket.Get(context, string(path))
|
|
|
|
if err != nil {
|
|
|
|
http.Error(w, err.Error(), 404)
|
2014-11-07 07:45:27 +01:00
|
|
|
} else if object == nil {
|
|
|
|
http.Error(w, errors.New("Object not found").Error(), 404)
|
2014-11-07 04:32:35 +01:00
|
|
|
} else {
|
|
|
|
fmt.Fprintf(w, string(object))
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
type GatewayPutHandler struct {
|
2014-11-08 04:07:16 +01:00
|
|
|
config GatewayConfig
|
2014-11-07 04:32:35 +01:00
|
|
|
}
|
|
|
|
|
|
|
|
func (handler GatewayPutHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
|
|
|
|
vars := mux.Vars(req)
|
|
|
|
bucketName := vars["bucket"]
|
|
|
|
path := vars["path"]
|
|
|
|
object, _ := ioutil.ReadAll(req.Body)
|
|
|
|
context := fakeContext{}
|
|
|
|
callback := make(chan Bucket)
|
2014-11-08 04:07:16 +01:00
|
|
|
handler.config.requestBucketChan <- BucketRequest{
|
2014-11-07 04:32:35 +01:00
|
|
|
name: bucketName,
|
|
|
|
context: context,
|
|
|
|
callback: callback,
|
|
|
|
}
|
|
|
|
bucket := <-callback
|
|
|
|
bucket.Put(context, path, object)
|
2014-11-03 04:55:01 +01:00
|
|
|
}
|
|
|
|
|
2014-11-07 07:45:27 +01:00
|
|
|
func RegisterGatewayHandlers(router *mux.Router, config GatewayConfig) {
|
2014-11-08 04:07:16 +01:00
|
|
|
config.requestBucketChan = make(chan BucketRequest)
|
2014-11-14 07:18:44 +01:00
|
|
|
go config.BucketDriver(config)
|
2014-11-08 04:07:16 +01:00
|
|
|
getHandler := GatewayGetHandler{config}
|
|
|
|
putHandler := GatewayPutHandler{config}
|
2014-11-07 04:32:35 +01:00
|
|
|
router.Handle("/{bucket}/{path:.*}", getHandler).Methods("GET")
|
|
|
|
router.Handle("/{bucket}/{path:.*}", putHandler).Methods("PUT")
|
2014-11-03 00:56:33 +01:00
|
|
|
}
|
2014-11-07 03:34:46 +01:00
|
|
|
|
2014-11-14 07:18:44 +01:00
|
|
|
func SynchronizedBucketDriver(config GatewayConfig) {
|
2014-11-07 03:34:46 +01:00
|
|
|
buckets := make(map[string]*SynchronizedBucket)
|
2014-11-08 04:07:16 +01:00
|
|
|
for request := range config.requestBucketChan {
|
2014-11-07 03:34:46 +01:00
|
|
|
if buckets[request.name] == nil {
|
|
|
|
bucketChannel := make(chan ObjectRequest)
|
2014-11-15 02:22:50 +01:00
|
|
|
go config.StorageDriver(request.name, bucketChannel, config)
|
2014-11-07 03:34:46 +01:00
|
|
|
buckets[request.name] = &SynchronizedBucket{
|
|
|
|
name: request.name,
|
|
|
|
channel: bucketChannel,
|
|
|
|
}
|
|
|
|
}
|
|
|
|
request.callback <- buckets[request.name]
|
|
|
|
}
|
|
|
|
for key := range buckets {
|
|
|
|
buckets[key].closeChannel()
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
type SynchronizedBucket struct {
|
|
|
|
name string
|
|
|
|
channel chan ObjectRequest
|
|
|
|
objects map[string][]byte
|
|
|
|
}
|
|
|
|
|
|
|
|
type ObjectRequest struct {
|
|
|
|
requestType string
|
|
|
|
path string
|
|
|
|
object []byte
|
|
|
|
callback chan interface{}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (bucket SynchronizedBucket) GetName(context Context) string {
|
|
|
|
return bucket.name
|
|
|
|
}
|
|
|
|
|
|
|
|
func (bucket SynchronizedBucket) Get(context Context, path string) ([]byte, error) {
|
|
|
|
callback := make(chan interface{})
|
|
|
|
bucket.channel <- ObjectRequest{
|
|
|
|
requestType: "GET",
|
|
|
|
path: path,
|
|
|
|
callback: callback,
|
|
|
|
}
|
2014-11-08 08:32:01 +01:00
|
|
|
response := <-callback
|
|
|
|
|
|
|
|
switch response.(type) {
|
2014-11-07 03:34:46 +01:00
|
|
|
case error:
|
|
|
|
return nil, response.(error)
|
2014-11-08 08:32:01 +01:00
|
|
|
case nil:
|
|
|
|
return nil, errors.New("Object not found")
|
|
|
|
case interface{}:
|
|
|
|
return response.([]byte), nil
|
2014-11-07 03:34:46 +01:00
|
|
|
default:
|
|
|
|
return nil, errors.New("Unexpected error, service failed")
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (bucket SynchronizedBucket) Put(context Context, path string, object []byte) error {
|
|
|
|
callback := make(chan interface{})
|
|
|
|
bucket.channel <- ObjectRequest{
|
|
|
|
requestType: "PUT",
|
|
|
|
path: path,
|
|
|
|
object: object,
|
|
|
|
callback: callback,
|
|
|
|
}
|
|
|
|
switch response := <-callback; response.(type) {
|
|
|
|
case error:
|
|
|
|
return response.(error)
|
|
|
|
case nil:
|
|
|
|
return nil
|
|
|
|
default:
|
|
|
|
return errors.New("Unexpected error, service failed")
|
|
|
|
}
|
|
|
|
}
|
|
|
|
|
|
|
|
func (bucket *SynchronizedBucket) closeChannel() {
|
|
|
|
close(bucket.channel)
|
|
|
|
}
|
|
|
|
|
2014-11-15 02:22:50 +01:00
|
|
|
func InMemoryStorageDriver(bucket string, input chan ObjectRequest, config GatewayConfig) {
|
2014-11-08 08:32:01 +01:00
|
|
|
objects := patricia.NewTrie()
|
2014-11-07 03:34:46 +01:00
|
|
|
for request := range input {
|
2014-11-08 08:32:01 +01:00
|
|
|
prefix := patricia.Prefix(request.path)
|
2014-11-07 03:34:46 +01:00
|
|
|
switch request.requestType {
|
|
|
|
case "GET":
|
2014-11-08 08:32:01 +01:00
|
|
|
request.callback <- objects.Get(prefix)
|
2014-11-07 03:34:46 +01:00
|
|
|
case "PUT":
|
2014-11-08 08:32:01 +01:00
|
|
|
objects.Insert(prefix, request.object)
|
2014-11-07 03:34:46 +01:00
|
|
|
request.callback <- nil
|
|
|
|
default:
|
|
|
|
request.callback <- errors.New("Unexpected message")
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|
2014-11-15 02:22:50 +01:00
|
|
|
|
|
|
|
func SimpleFileStorageDriver(bucket string, input chan ObjectRequest, config GatewayConfig) {
|
2014-12-08 23:25:57 +01:00
|
|
|
fileStorage := fsstorage.FileSystemStorage{
|
2014-11-15 02:22:50 +01:00
|
|
|
RootDir: config.dataDir,
|
|
|
|
}
|
|
|
|
for request := range input {
|
|
|
|
switch request.requestType {
|
|
|
|
case "GET":
|
|
|
|
objectPath := path.Join(bucket, request.path)
|
2014-11-29 23:42:22 +01:00
|
|
|
object, err := fileStorage.Get(objectPath)
|
2014-11-15 02:22:50 +01:00
|
|
|
if err != nil {
|
|
|
|
request.callback <- nil
|
|
|
|
} else {
|
|
|
|
request.callback <- object
|
|
|
|
}
|
|
|
|
case "PUT":
|
|
|
|
objectPath := path.Join(bucket, request.path)
|
2014-11-29 23:42:22 +01:00
|
|
|
fileStorage.Put(objectPath, request.object)
|
2014-11-15 02:22:50 +01:00
|
|
|
request.callback <- nil
|
|
|
|
default:
|
|
|
|
request.callback <- errors.New("Unexpected message")
|
|
|
|
}
|
|
|
|
}
|
|
|
|
}
|