add config options and graph.Handle

config/config.go

@@ -26,25 +26,29 @@ import (
 )
 
 type Config struct {
-	DatabaseType    string
+	DatabaseType       string
-	DatabasePath    string
+	DatabasePath       string
-	DatabaseOptions map[string]interface{}
+	DatabaseOptions    map[string]interface{}
-	ListenHost      string
+	ReplicationType    string
-	ListenPort      string
+	ReplicationOptions map[string]interface{}
-	ReadOnly        bool
+	ListenHost         string
-	Timeout         time.Duration
+	ListenPort         string
-	LoadSize        int
+	ReadOnly           bool
+	Timeout            time.Duration
+	LoadSize           int
 }
 
 type config struct {
-	DatabaseType    string                 `json:"database"`
+	DatabaseType       string                 `json:"database"`
-	DatabasePath    string                 `json:"db_path"`
+	DatabasePath       string                 `json:"db_path"`
-	DatabaseOptions map[string]interface{} `json:"db_options"`
+	DatabaseOptions    map[string]interface{} `json:"db_options"`
-	ListenHost      string                 `json:"listen_host"`
+	ReplicationType    string                 `json:"replication"`
-	ListenPort      string                 `json:"listen_port"`
+	ReplicationOptions map[string]interface{} `json:"replication_options"`
-	ReadOnly        bool                   `json:"read_only"`
+	ListenHost         string                 `json:"listen_host"`
-	Timeout         duration               `json:"timeout"`
+	ListenPort         string                 `json:"listen_port"`
-	LoadSize        int                    `json:"load_size"`
+	ReadOnly           bool                   `json:"read_only"`
+	Timeout            duration               `json:"timeout"`
+	LoadSize           int                    `json:"load_size"`
 }
 
 func (c *Config) UnmarshalJSON(data []byte) error {
@@ -54,28 +58,32 @@ func (c *Config) UnmarshalJSON(data []byte) error {
 		return err
 	}
 	*c = Config{
-		DatabaseType:    t.DatabaseType,
+		DatabaseType:       t.DatabaseType,
-		DatabasePath:    t.DatabasePath,
+		DatabasePath:       t.DatabasePath,
-		DatabaseOptions: t.DatabaseOptions,
+		DatabaseOptions:    t.DatabaseOptions,
-		ListenHost:      t.ListenHost,
+		ReplicationType:    t.ReplicationType,
-		ListenPort:      t.ListenPort,
+		ReplicationOptions: t.ReplicationOptions,
-		ReadOnly:        t.ReadOnly,
+		ListenHost:         t.ListenHost,
-		Timeout:         time.Duration(t.Timeout),
+		ListenPort:         t.ListenPort,
-		LoadSize:        t.LoadSize,
+		ReadOnly:           t.ReadOnly,
+		Timeout:            time.Duration(t.Timeout),
+		LoadSize:           t.LoadSize,
 	}
 	return nil
 }
 
 func (c *Config) MarshalJSON() ([]byte, error) {
 	return json.Marshal(config{
-		DatabaseType:    c.DatabaseType,
+		DatabaseType:       c.DatabaseType,
-		DatabasePath:    c.DatabasePath,
+		DatabasePath:       c.DatabasePath,
-		DatabaseOptions: c.DatabaseOptions,
+		DatabaseOptions:    c.DatabaseOptions,
-		ListenHost:      c.ListenHost,
+		ReplicationType:    c.ReplicationType,
-		ListenPort:      c.ListenPort,
+		ReplicationOptions: c.ReplicationOptions,
-		ReadOnly:        c.ReadOnly,
+		ListenHost:         c.ListenHost,
-		Timeout:         duration(c.Timeout),
+		ListenPort:         c.ListenPort,
-		LoadSize:        c.LoadSize,
+		ReadOnly:           c.ReadOnly,
+		Timeout:            duration(c.Timeout),
+		LoadSize:           c.LoadSize,
 	})
 }
 
@@ -115,13 +123,14 @@ func (d *duration) MarshalJSON() ([]byte, error) {
 }
 
 var (
-	databasePath    = flag.String("dbpath", "/tmp/testdb", "Path to the database.")
+	databasePath       = flag.String("dbpath", "/tmp/testdb", "Path to the database.")
-	databaseBackend = flag.String("db", "memstore", "Database Backend.")
+	databaseBackend    = flag.String("db", "memstore", "Database Backend.")
-	host            = flag.String("host", "0.0.0.0", "Host to listen on (defaults to all).")
+	replicationBackend = flag.String("replication", "single", "Replication method.")
-	loadSize        = flag.Int("load_size", 10000, "Size of triplesets to load")
+	host               = flag.String("host", "0.0.0.0", "Host to listen on (defaults to all).")
-	port            = flag.String("port", "64210", "Port to listen on.")
+	loadSize           = flag.Int("load_size", 10000, "Size of triplesets to load")
-	readOnly        = flag.Bool("read_only", false, "Disable writing via HTTP.")
+	port               = flag.String("port", "64210", "Port to listen on.")
-	timeout         = flag.Duration("timeout", 30*time.Second, "Elapsed time until an individual query times out.")
+	readOnly           = flag.Bool("read_only", false, "Disable writing via HTTP.")
+	timeout            = flag.Duration("timeout", 30*time.Second, "Elapsed time until an individual query times out.")
 )
 
 func ParseConfigFromFile(filename string) *Config {
@@ -175,6 +184,10 @@ func ParseConfigFromFlagsAndFile(fileFlag string) *Config {
 		config.DatabaseType = *databaseBackend
 	}
 
+	if config.ReplicationType == "" {
+		config.ReplicationType = *replicationBackend
+	}
+
 	if config.ListenHost == "" {
 		config.ListenHost = *host
 	}

db/db.go

@@ -36,8 +36,20 @@ func Init(cfg *config.Config) error {
 	return graph.InitTripleStore(cfg.DatabaseType, cfg.DatabasePath, cfg.DatabaseOptions)
 }
 
-func Open(cfg *config.Config) (graph.TripleStore, error) {
+func Open(cfg *config.Config) (*graph.Handle, error) {
-	glog.Infof("Opening database %q at %s", cfg.DatabaseType, cfg.DatabasePath)
+	qs, err := OpenQuadStore(cfg)
+	if err != nil {
+		return nil, err
+	}
+	qw, err := OpenQuadWriter(qs, cfg)
+	if err != nil {
+		return nil, err
+	}
+	return &graph.Handle{QuadStore: qs, QuadWriter: qw}, nil
+}
+
+func OpenQuadStore(cfg *config.Config) (graph.TripleStore, error) {
+	glog.Infof("Opening quad store %q at %s", cfg.DatabaseType, cfg.DatabasePath)
 	ts, err := graph.NewTripleStore(cfg.DatabaseType, cfg.DatabasePath, cfg.DatabaseOptions)
 	if err != nil {
 		return nil, err
@@ -46,6 +58,16 @@ func Open(cfg *config.Config) (graph.TripleStore, error) {
 	return ts, nil
 }
 
+func OpenQuadWriter(qs graph.TripleStore, cfg *config.Config) (graph.QuadWriter, error) {
+	glog.Infof("Opening replication method %q", cfg.ReplicationType)
+	w, err := graph.NewQuadWriter(cfg.ReplicationType, qs, cfg.ReplicationOptions)
+	if err != nil {
+		return nil, err
+	}
+
+	return w, nil
+}
+
 func Load(ts graph.TripleStore, cfg *config.Config, dec quad.Unmarshaler) error {
 	bulker, canBulk := ts.(graph.BulkLoader)
 	if canBulk {

graph/quadwriter.go

@@ -43,6 +43,11 @@ type Delta struct {
 	Timestamp time.Time
 }
 
+type Handle struct {
+	QuadStore  TripleStore
+	QuadWriter QuadWriter
+}
+
 var ErrQuadExists = errors.New("Quad exists")
 var ErrQuadNotExist = errors.New("Quad doesn't exist")