minio/pkgs/gateway/gateway.go

/*
 * Mini Object Storage, (C) 2014 Minio, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package gateway

import (
	"bytes"
	"errors"
	"fmt"
	"io/ioutil"
	"net/http"
	"path"

	"github.com/gorilla/mux"
	"github.com/minio-io/minio/pkgs/storage/encodedstorage"
	"github.com/tchap/go-patricia/patricia"
)

// Stores system configuration, populated from CLI or test runner
type GatewayConfig struct {
	StorageDriver     StorageDriver
	BucketDriver      BucketDriver
	requestBucketChan chan BucketRequest
	DataDir           string
	K,
	M int
	BlockSize uint64
}

// Message for requesting a bucket
type BucketRequest struct {
	name     string
	context  Context
	callback chan Bucket
}

// Context interface for security and session information
type Context interface{}

// Bucket definition
type Bucket interface {
	GetName(Context) string
	Get(Context, string) ([]byte, error)
	Put(Context, string, []byte) error
}

// Bucket driver function, should read from a channel and respond through callback channels
type BucketDriver func(config GatewayConfig)

// Storage driver function, should read from a channel and respond through callback channels
type StorageDriver func(bucket string, input chan ObjectRequest, config GatewayConfig)

// TODO remove when building real context
type fakeContext struct{}

type GatewayGetHandler struct {
	config GatewayConfig
}

// GET requests server
func (handler GatewayGetHandler) ServeHTTP(w http.ResponseWriter, req *http.Request) {
	vars := mux.Vars(req)
	bucketName := vars["bucket"]
	path := vars["path"]
	context := fakeContext{}
	callback := make(chan Bucket)
	handler.config.requestBucketChan <- BucketRequest{
		name:     bucketName,
		context:  context,
		callback: callback,
	}
	bucket := <-callback
	object, err := bucket.Get(context, string(path))
	if err != nil {
		http.Error(w, err.Error(), 404)
	} else if object == nil {
		http.Error(w, errors.New("Object not found").Error(), 404)
	} else {
		fmt.Fprintf(w, string(object))
	}
}

type GatewayPutHandler struct {
	config GatewayConfig
}

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)
	handler.config.requestBucketChan <- BucketRequest{
		name:     bucketName,
		context:  context,
		callback: callback,
	}
	bucket := <-callback
	bucket.Put(context, path, object)
}

func RegisterGatewayHandlers(router *mux.Router, config GatewayConfig) {
	config.requestBucketChan = make(chan BucketRequest)
	go config.BucketDriver(config)
	getHandler := GatewayGetHandler{config}
	putHandler := GatewayPutHandler{config}
	router.Handle("/{bucket}/{path:.*}", getHandler).Methods("GET")
	router.Handle("/{bucket}/{path:.*}", putHandler).Methods("PUT")
}

func SynchronizedBucketDriver(config GatewayConfig) {
	buckets := make(map[string]*SynchronizedBucket)
	for request := range config.requestBucketChan {
		if buckets[request.name] == nil {
			bucketChannel := make(chan ObjectRequest)
			go config.StorageDriver(request.name, bucketChannel, config)
			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,
	}
	response := <-callback

	switch response.(type) {
	case error:
		return nil, response.(error)
	case nil:
		return nil, errors.New("Object not found")
	case interface{}:
		return response.([]byte), nil
	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)
}

func InMemoryStorageDriver(bucket string, input chan ObjectRequest, config GatewayConfig) {
	objects := patricia.NewTrie()
	for request := range input {
		prefix := patricia.Prefix(request.path)
		switch request.requestType {
		case "GET":
			request.callback <- objects.Get(prefix)
		case "PUT":
			objects.Insert(prefix, request.object)
			request.callback <- nil
		default:
			request.callback <- errors.New("Unexpected message")
		}
	}
}

func SimpleEncodedStorageDriver(bucket string, input chan ObjectRequest, config GatewayConfig) {
	eStorage, _ := encodedstorage.NewStorage(config.DataDir, config.K, config.M, config.BlockSize)
	for request := range input {
		switch request.requestType {
		case "GET":
			objectPath := path.Join(bucket, request.path)
			object, err := eStorage.Get(objectPath)
			if err != nil {
				request.callback <- err
			} else {
				request.callback <- object
			}
		case "PUT":
			objectPath := path.Join(bucket, request.path)
			err := eStorage.Put(objectPath, bytes.NewBuffer(request.object))
			if err != nil {
				request.callback <- err
			} else {
				request.callback <- nil
			}
		default:
			request.callback <- errors.New("Unexpected message")
		}
	}
}

//func SimpleFileStorageDriver(bucket string, input chan ObjectRequest, config GatewayConfig) {
//	fileStorage, _ := fsstorage.NewStorage(config.DataDir, config.BlockSize)
//	for request := range input {
//		switch request.requestType {
//		case "GET":
//			objectPath := path.Join(bucket, request.path)
//			object, err := fileStorage.Get(objectPath)
//			if err != nil {
//				request.callback <- nil
//			} else {
//				request.callback <- object
//			}
//		case "PUT":
//			objectPath := path.Join(bucket, request.path)
//			fileStorage.Put(objectPath, bytes.NewBuffer(request.object))
//			request.callback <- nil
//		default:
//			request.callback <- errors.New("Unexpected message")
//		}
//	}
//}