Chris Lu
6 years ago
17 changed files with 70 additions and 236 deletions
-
2weed/command/filer.go
-
1weed/command/server.go
-
3weed/filer2/cassandra/cassandra_store.go
-
36weed/filer2/configuration.go
-
3weed/filer2/filerstore.go
-
2weed/filer2/leveldb/leveldb_store.go
-
3weed/filer2/memdb/memdb_store.go
-
3weed/filer2/mysql/mysql_store.go
-
3weed/filer2/postgres/postgres_store.go
-
3weed/filer2/redis/redis_cluster_store.go
-
3weed/filer2/redis/redis_store.go
-
61weed/msgqueue/configuration.go
-
3weed/msgqueue/kafka/kafka_queue.go
-
3weed/msgqueue/log/log_queue.go
-
7weed/msgqueue/message_queue.go
-
36weed/server/filer_server.go
-
134weed/util/config.go
@ -1,82 +1,29 @@ |
|||
package msgqueue |
|||
|
|||
import ( |
|||
"os" |
|||
|
|||
"github.com/chrislusf/seaweedfs/weed/glog" |
|||
"github.com/spf13/viper" |
|||
) |
|||
|
|||
const ( |
|||
MSG_QUEUE_TOML_EXAMPLE = ` |
|||
# A sample TOML config file for SeaweedFS message queue |
|||
|
|||
[log] |
|||
enabled = true |
|||
|
|||
[kafka] |
|||
enabled = false |
|||
hosts = [ |
|||
"localhost:9092" |
|||
] |
|||
topic = "seaweedfs_filer" |
|||
|
|||
` |
|||
) |
|||
|
|||
var ( |
|||
MessageQueues []MessageQueue |
|||
|
|||
Queue MessageQueue |
|||
) |
|||
|
|||
func LoadConfiguration() { |
|||
|
|||
// find a filer store
|
|||
viper.SetConfigName("message_queue") // name of config file (without extension)
|
|||
viper.AddConfigPath(".") // optionally look for config in the working directory
|
|||
viper.AddConfigPath("$HOME/.seaweedfs") // call multiple times to add many search paths
|
|||
viper.AddConfigPath("/etc/seaweedfs/") // path to look for the config file in
|
|||
if err := viper.ReadInConfig(); err != nil { // Handle errors reading the config file
|
|||
glog.Fatalf("Failed to load message_queue.toml file from current directory, or $HOME/.seaweedfs/, "+ |
|||
"or /etc/seaweedfs/"+ |
|||
"\n\nPlease follow this example and add a message_queue.toml file to "+ |
|||
"current directory, or $HOME/.seaweedfs/, or /etc/seaweedfs/:\n"+MSG_QUEUE_TOML_EXAMPLE) |
|||
} |
|||
func LoadConfiguration(config *viper.Viper) { |
|||
|
|||
glog.V(0).Infof("Reading message queue configuration from %s", viper.ConfigFileUsed()) |
|||
for _, store := range MessageQueues { |
|||
if viper.GetBool(store.GetName() + ".enabled") { |
|||
viperSub := viper.Sub(store.GetName()) |
|||
if config.GetBool(store.GetName() + ".enabled") { |
|||
viperSub := config.Sub(store.GetName()) |
|||
if err := store.Initialize(viperSub); err != nil { |
|||
glog.Fatalf("Failed to initialize store for %s: %+v", |
|||
store.GetName(), err) |
|||
} |
|||
Queue = store |
|||
glog.V(0).Infof("Configure message queue for %s from %s", store.GetName(), viper.ConfigFileUsed()) |
|||
glog.V(0).Infof("Configure message queue for %s", store.GetName()) |
|||
return |
|||
} |
|||
} |
|||
|
|||
println() |
|||
println("Supported message queues are:") |
|||
for _, store := range MessageQueues { |
|||
println(" " + store.GetName()) |
|||
} |
|||
|
|||
println() |
|||
println("Please configure a supported message queue in", viper.ConfigFileUsed()) |
|||
println() |
|||
|
|||
os.Exit(-1) |
|||
} |
|||
|
|||
// A simplified interface to decouple from Viper
|
|||
type Configuration interface { |
|||
GetString(key string) string |
|||
GetBool(key string) bool |
|||
GetInt(key string) int |
|||
GetInt64(key string) int64 |
|||
GetFloat64(key string) float64 |
|||
GetStringSlice(key string) []string |
|||
} |
|||
@ -1,11 +1,14 @@ |
|||
package msgqueue |
|||
|
|||
import "github.com/golang/protobuf/proto" |
|||
import ( |
|||
"github.com/golang/protobuf/proto" |
|||
"github.com/chrislusf/seaweedfs/weed/util" |
|||
) |
|||
|
|||
type MessageQueue interface { |
|||
// GetName gets the name to locate the configuration in message_queue.toml file
|
|||
GetName() string |
|||
// Initialize initializes the file store
|
|||
Initialize(configuration Configuration) error |
|||
Initialize(configuration util.Configuration) error |
|||
SendMessage(key string, message proto.Message) error |
|||
} |
|||
@ -1,130 +1,10 @@ |
|||
package util |
|||
|
|||
// Copyright 2011 Numerotron Inc.
|
|||
// Use of this source code is governed by an MIT-style license
|
|||
// that can be found in the LICENSE file.
|
|||
//
|
|||
// Developed at www.stathat.com by Patrick Crosby
|
|||
// Contact us on twitter with any questions: twitter.com/stat_hat
|
|||
|
|||
// The jconfig package provides a simple, basic configuration file parser using JSON.
|
|||
|
|||
import ( |
|||
"bytes" |
|||
"encoding/json" |
|||
"os" |
|||
|
|||
"github.com/chrislusf/seaweedfs/weed/glog" |
|||
) |
|||
|
|||
type Config struct { |
|||
data map[string]interface{} |
|||
filename string |
|||
} |
|||
|
|||
func newConfig() *Config { |
|||
result := new(Config) |
|||
result.data = make(map[string]interface{}) |
|||
return result |
|||
} |
|||
|
|||
// Loads config information from a JSON file
|
|||
func LoadConfig(filename string) *Config { |
|||
result := newConfig() |
|||
result.filename = filename |
|||
err := result.parse() |
|||
if err != nil { |
|||
glog.Fatalf("error loading config file %s: %s", filename, err) |
|||
} |
|||
return result |
|||
} |
|||
|
|||
// Loads config information from a JSON string
|
|||
func LoadConfigString(s string) *Config { |
|||
result := newConfig() |
|||
err := json.Unmarshal([]byte(s), &result.data) |
|||
if err != nil { |
|||
glog.Fatalf("error parsing config string %s: %s", s, err) |
|||
} |
|||
return result |
|||
} |
|||
|
|||
func (c *Config) StringMerge(s string) { |
|||
next := LoadConfigString(s) |
|||
c.merge(next.data) |
|||
} |
|||
|
|||
func (c *Config) LoadMerge(filename string) { |
|||
next := LoadConfig(filename) |
|||
c.merge(next.data) |
|||
} |
|||
|
|||
func (c *Config) merge(ndata map[string]interface{}) { |
|||
for k, v := range ndata { |
|||
c.data[k] = v |
|||
} |
|||
} |
|||
|
|||
func (c *Config) parse() error { |
|||
f, err := os.Open(c.filename) |
|||
if err != nil { |
|||
return err |
|||
} |
|||
defer f.Close() |
|||
b := new(bytes.Buffer) |
|||
_, err = b.ReadFrom(f) |
|||
if err != nil { |
|||
return err |
|||
} |
|||
err = json.Unmarshal(b.Bytes(), &c.data) |
|||
if err != nil { |
|||
return err |
|||
} |
|||
|
|||
return nil |
|||
} |
|||
|
|||
// Returns a string for the config variable key
|
|||
func (c *Config) GetString(key string) string { |
|||
result, present := c.data[key] |
|||
if !present { |
|||
return "" |
|||
} |
|||
return result.(string) |
|||
} |
|||
|
|||
// Returns an int for the config variable key
|
|||
func (c *Config) GetInt(key string) int { |
|||
x, ok := c.data[key] |
|||
if !ok { |
|||
return -1 |
|||
} |
|||
return int(x.(float64)) |
|||
} |
|||
|
|||
// Returns a float for the config variable key
|
|||
func (c *Config) GetFloat(key string) float64 { |
|||
x, ok := c.data[key] |
|||
if !ok { |
|||
return -1 |
|||
} |
|||
return x.(float64) |
|||
} |
|||
|
|||
// Returns a bool for the config variable key
|
|||
func (c *Config) GetBool(key string) bool { |
|||
x, ok := c.data[key] |
|||
if !ok { |
|||
return false |
|||
} |
|||
return x.(bool) |
|||
} |
|||
|
|||
// Returns an array for the config variable key
|
|||
func (c *Config) GetArray(key string) []interface{} { |
|||
result, present := c.data[key] |
|||
if !present { |
|||
return []interface{}(nil) |
|||
} |
|||
return result.([]interface{}) |
|||
type Configuration interface { |
|||
GetString(key string) string |
|||
GetBool(key string) bool |
|||
GetInt(key string) int |
|||
GetInt64(key string) int64 |
|||
GetFloat64(key string) float64 |
|||
GetStringSlice(key string) []string |
|||
} |
|||
Write
Preview
Loading…
Cancel
Save
Reference in new issue