mirror of
https://github.com/minio/minio.git
synced 2025-02-03 09:55:59 -05:00
Merge pull request #688 from harshavardhana/pr_out_use_errorchannels_only_for_services_not_for_drivers_reduce_them_to_use_simple_functions
This commit is contained in:
commit
63f9647c80
@ -59,7 +59,7 @@ var _ = Suite(&MySuite{
|
||||
|
||||
var _ = Suite(&MySuite{
|
||||
initDriver: func() (drivers.Driver, string) {
|
||||
_, _, driver := memory.Start(10000, 3*time.Hour)
|
||||
driver, _ := memory.NewDriver(10000, 3*time.Hour)
|
||||
return driver, ""
|
||||
},
|
||||
})
|
||||
@ -69,7 +69,7 @@ var _ = Suite(&MySuite{
|
||||
root, _ := ioutil.TempDir(os.TempDir(), "minio-api")
|
||||
var roots []string
|
||||
roots = append(roots, root)
|
||||
_, _, driver := donut.Start(roots, 10000, 3*time.Hour)
|
||||
driver, _ := donut.NewDriver(roots, 10000, 3*time.Hour)
|
||||
return driver, root
|
||||
},
|
||||
})
|
||||
@ -77,7 +77,7 @@ var _ = Suite(&MySuite{
|
||||
var _ = Suite(&MySuite{
|
||||
initDriver: func() (drivers.Driver, string) {
|
||||
root, _ := ioutil.TempDir(os.TempDir(), "minio-fs-api")
|
||||
_, _, driver := filesystem.Start(root)
|
||||
driver, _ := filesystem.NewDriver(root)
|
||||
return driver, root
|
||||
},
|
||||
})
|
||||
|
@ -44,10 +44,10 @@ func Start(handler http.Handler, config Config) (chan<- string, <-chan error, *S
|
||||
return ctrlChannel, errorChannel, &server
|
||||
}
|
||||
|
||||
func start(ctrlChannel <-chan string, errorChannel chan<- error,
|
||||
router http.Handler, config Config, server *Server) {
|
||||
var err error
|
||||
func start(ctrlChannel <-chan string, errorChannel chan<- error, router http.Handler, config Config, server *Server) {
|
||||
defer close(errorChannel)
|
||||
|
||||
var err error
|
||||
// Minio server config
|
||||
httpServer := &http.Server{
|
||||
Addr: config.Address,
|
||||
@ -56,7 +56,10 @@ func start(ctrlChannel <-chan string, errorChannel chan<- error,
|
||||
}
|
||||
|
||||
host, port, err := net.SplitHostPort(config.Address)
|
||||
errorChannel <- err
|
||||
if err != nil {
|
||||
errorChannel <- err
|
||||
return
|
||||
}
|
||||
|
||||
var hosts []string
|
||||
switch {
|
||||
@ -67,9 +70,9 @@ func start(ctrlChannel <-chan string, errorChannel chan<- error,
|
||||
errorChannel <- err
|
||||
for _, addr := range addrs {
|
||||
if addr.Network() == "ip+net" {
|
||||
h := strings.Split(addr.String(), "/")[0]
|
||||
if ip := net.ParseIP(h); ip.To4() != nil {
|
||||
hosts = append(hosts, h)
|
||||
host := strings.Split(addr.String(), "/")[0]
|
||||
if ip := net.ParseIP(host); ip.To4() != nil {
|
||||
hosts = append(hosts, host)
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -86,6 +89,8 @@ func start(ctrlChannel <-chan string, errorChannel chan<- error,
|
||||
}
|
||||
err = httpServer.ListenAndServeTLS(config.CertFile, config.KeyFile)
|
||||
}
|
||||
errorChannel <- err
|
||||
close(errorChannel)
|
||||
if err != nil {
|
||||
errorChannel <- err
|
||||
}
|
||||
|
||||
}
|
||||
|
@ -42,7 +42,7 @@ type MemoryFactory struct {
|
||||
// GetStartServerFunc builds memory api server
|
||||
func (f MemoryFactory) GetStartServerFunc() StartServerFunc {
|
||||
return func() (chan<- string, <-chan error) {
|
||||
_, _, driver := memory.Start(f.MaxMemory, f.Expiration)
|
||||
driver, _ := memory.NewDriver(f.MaxMemory, f.Expiration)
|
||||
conf := api.Config{RateLimit: f.RateLimit}
|
||||
conf.SetDriver(driver)
|
||||
ctrl, status, _ := httpserver.Start(api.HTTPHandler(conf), f.Config)
|
||||
@ -59,7 +59,7 @@ type FilesystemFactory struct {
|
||||
// GetStartServerFunc builds memory api server
|
||||
func (f FilesystemFactory) GetStartServerFunc() StartServerFunc {
|
||||
return func() (chan<- string, <-chan error) {
|
||||
_, _, driver := fs.Start(f.Path)
|
||||
driver, _ := fs.NewDriver(f.Path)
|
||||
conf := api.Config{RateLimit: f.RateLimit}
|
||||
conf.SetDriver(driver)
|
||||
ctrl, status, _ := httpserver.Start(api.HTTPHandler(conf), f.Config)
|
||||
@ -91,7 +91,10 @@ type DonutFactory struct {
|
||||
// GetStartServerFunc DonutFactory builds donut api server
|
||||
func (f DonutFactory) GetStartServerFunc() StartServerFunc {
|
||||
return func() (chan<- string, <-chan error) {
|
||||
_, _, driver := donut.Start(f.Paths, f.MaxMemory, f.Expiration)
|
||||
driver, err := donut.NewDriver(f.Paths, f.MaxMemory, f.Expiration)
|
||||
if err != nil {
|
||||
log.Fatalln(err)
|
||||
}
|
||||
conf := api.Config{RateLimit: f.RateLimit}
|
||||
conf.SetDriver(driver)
|
||||
ctrl, status, _ := httpserver.Start(api.HTTPHandler(conf), f.Config)
|
||||
|
@ -101,18 +101,18 @@ func createNodeDiskMap(paths []string) map[string][]string {
|
||||
return nodes
|
||||
}
|
||||
|
||||
func initialize(d *donutDriver) {
|
||||
func initialize(d *donutDriver) error {
|
||||
// Soon to be user configurable, when Management API is available
|
||||
// we should remove "default" to something which is passed down
|
||||
// from configuration paramters
|
||||
var err error
|
||||
d.donut, err = donut.NewDonut("default", createNodeDiskMap(d.paths))
|
||||
if err != nil {
|
||||
panic(iodine.New(err, nil))
|
||||
return iodine.New(err, nil)
|
||||
}
|
||||
buckets, err := d.donut.ListBuckets()
|
||||
if err != nil {
|
||||
panic(iodine.New(err, nil))
|
||||
return iodine.New(err, nil)
|
||||
}
|
||||
for bucketName, metadata := range buckets {
|
||||
d.lock.RLock()
|
||||
@ -136,13 +136,11 @@ func initialize(d *donutDriver) {
|
||||
d.storedBuckets[bucketName] = storedBucket
|
||||
d.lock.Unlock()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
// Start a single disk subsystem
|
||||
func Start(paths []string, maxSize uint64, expiration time.Duration) (chan<- string, <-chan error, drivers.Driver) {
|
||||
ctrlChannel := make(chan string)
|
||||
errorChannel := make(chan error)
|
||||
|
||||
// NewDriver instantiate a donut driver
|
||||
func NewDriver(paths []string, maxSize uint64, expiration time.Duration) (drivers.Driver, error) {
|
||||
driver := new(donutDriver)
|
||||
driver.storedBuckets = make(map[string]storedBucket)
|
||||
driver.objects = trove.NewCache(maxSize, expiration)
|
||||
@ -160,14 +158,8 @@ func Start(paths []string, maxSize uint64, expiration time.Duration) (chan<- str
|
||||
driver.paths = paths
|
||||
driver.lock = new(sync.RWMutex)
|
||||
|
||||
initialize(driver)
|
||||
|
||||
go start(ctrlChannel, errorChannel, driver)
|
||||
return ctrlChannel, errorChannel, driver
|
||||
}
|
||||
|
||||
func start(ctrlChannel <-chan string, errorChannel chan<- error, driver *donutDriver) {
|
||||
defer close(errorChannel)
|
||||
err := initialize(driver)
|
||||
return driver, err
|
||||
}
|
||||
|
||||
func (d donutDriver) expiredObject(a ...interface{}) {
|
||||
|
@ -40,7 +40,8 @@ func (s *MySuite) TestAPISuite(c *C) {
|
||||
c.Check(err, IsNil)
|
||||
storageList = append(storageList, p)
|
||||
paths = append(paths, p)
|
||||
_, _, store := Start(paths, 1000000, 3*time.Hour)
|
||||
store, err := NewDriver(paths, 1000000, 3*time.Hour)
|
||||
c.Check(err, IsNil)
|
||||
return store
|
||||
}
|
||||
drivers.APITestSuite(c, create)
|
||||
|
@ -29,22 +29,14 @@ type fsDriver struct {
|
||||
multiparts *Multiparts
|
||||
}
|
||||
|
||||
// Start filesystem channel
|
||||
func Start(root string) (chan<- string, <-chan error, drivers.Driver) {
|
||||
ctrlChannel := make(chan string)
|
||||
errorChannel := make(chan error)
|
||||
// NewDriver instantiate a new filesystem driver
|
||||
func NewDriver(root string) (drivers.Driver, error) {
|
||||
fs := new(fsDriver)
|
||||
fs.root = root
|
||||
fs.lock = new(sync.Mutex)
|
||||
// internal related to multiparts
|
||||
fs.multiparts = new(Multiparts)
|
||||
fs.multiparts.ActiveSession = make(map[string]*MultipartSession)
|
||||
go start(ctrlChannel, errorChannel, fs)
|
||||
return ctrlChannel, errorChannel, fs
|
||||
}
|
||||
|
||||
func start(ctrlChannel <-chan string, errorChannel chan<- error, fs *fsDriver) {
|
||||
err := os.MkdirAll(fs.root, 0700)
|
||||
errorChannel <- err
|
||||
close(errorChannel)
|
||||
return fs, err
|
||||
}
|
||||
|
@ -38,7 +38,8 @@ func (s *MySuite) TestAPISuite(c *C) {
|
||||
path, err := ioutil.TempDir(os.TempDir(), "minio-fs-")
|
||||
c.Check(err, IsNil)
|
||||
storageList = append(storageList, path)
|
||||
_, _, store := Start(path)
|
||||
store, err := NewDriver(path)
|
||||
c.Check(err, IsNil)
|
||||
return store
|
||||
}
|
||||
drivers.APITestSuite(c, create)
|
||||
|
@ -64,13 +64,9 @@ const (
|
||||
totalBuckets = 100
|
||||
)
|
||||
|
||||
// Start memory object server
|
||||
func Start(maxSize uint64, expiration time.Duration) (chan<- string, <-chan error, drivers.Driver) {
|
||||
ctrlChannel := make(chan string)
|
||||
errorChannel := make(chan error)
|
||||
|
||||
var memory *memoryDriver
|
||||
memory = new(memoryDriver)
|
||||
// NewDriver instantiate a new memory driver
|
||||
func NewDriver(maxSize uint64, expiration time.Duration) (drivers.Driver, error) {
|
||||
memory := new(memoryDriver)
|
||||
memory.storedBuckets = make(map[string]storedBucket)
|
||||
memory.objects = trove.NewCache(maxSize, expiration)
|
||||
memory.maxSize = maxSize
|
||||
@ -83,13 +79,7 @@ func Start(maxSize uint64, expiration time.Duration) (chan<- string, <-chan erro
|
||||
|
||||
// set up memory expiration
|
||||
memory.objects.ExpireObjects(time.Second * 5)
|
||||
|
||||
go start(ctrlChannel, errorChannel)
|
||||
return ctrlChannel, errorChannel, memory
|
||||
}
|
||||
|
||||
func start(ctrlChannel <-chan string, errorChannel chan<- error) {
|
||||
close(errorChannel)
|
||||
return memory, nil
|
||||
}
|
||||
|
||||
// GetObject - GET object from memory buffer
|
||||
|
@ -32,7 +32,8 @@ var _ = Suite(&MySuite{})
|
||||
|
||||
func (s *MySuite) TestAPISuite(c *C) {
|
||||
create := func() drivers.Driver {
|
||||
_, _, store := Start(1000000, 3*time.Hour)
|
||||
store, err := NewDriver(1000000, 3*time.Hour)
|
||||
c.Check(err, IsNil)
|
||||
return store
|
||||
}
|
||||
drivers.APITestSuite(c, create)
|
||||
|
Loading…
x
Reference in New Issue
Block a user