scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-10-08 20:07:51 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1413 from codysoyland/command-server-opts

Browse source 
Add CommandOptions and use OptCommandServerOptions to inject mock TranslateStore
This commit is contained in:
Matthew Jaffee 2018-06-26 11:44:49 -05:00 • committed by GitHub
commit 6e89abb8ed
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
6 changed files with 78 additions and 48 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

33
http/translator_test.go
View file

@ -4,13 +4,13 @@ import (
"context"
"io"
"io/ioutil"
"net/http/httptest"
"testing"
"time"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/http"
"github.com/pilosa/pilosa/mock"
"github.com/pilosa/pilosa/server"
"github.com/pilosa/pilosa/test"
)
@ -52,13 +52,14 @@ func TestTranslateStore_Reader(t *testing.T) {
}
return &mrc, nil
}
h := test.MustNewHandler()
h.API.TranslateStore = &translateStore
s := httptest.NewServer(h)
defer s.Close()
opts := server.OptCommandServerOptions(pilosa.OptServerPrimaryTranslateStore(translateStore))
main := test.MustRunMainWithCluster(t, 1, []server.CommandOption{opts})[0]
defer main.Close()
// Connect to server and stream all available data.
store := http.NewTranslateStore(s.URL)
store := http.NewTranslateStore(main.Server.URI.String())
rc, err := store.Reader(context.Background(), 100)
if err != nil {
t.Fatal(err)
@ -95,15 +96,16 @@ func TestTranslateStore_Reader(t *testing.T) {
translateStore.ReaderFunc = func(ctx context.Context, off int64) (io.ReadCloser, error) {
return &mrc, nil
}
h := test.MustNewHandler()
h.API.TranslateStore = &translateStore
s := httptest.NewServer(h)
defer s.Close()
opts := server.OptCommandServerOptions(pilosa.OptServerPrimaryTranslateStore(translateStore))
main := test.MustRunMainWithCluster(t, 1, []server.CommandOption{opts})[0]
defer main.Close()
defer close(done)
// Connect to server and begin streaming.
ctx, cancel := context.WithCancel(context.Background())
store := http.NewTranslateStore(s.URL)
store := http.NewTranslateStore(main.Server.URI.String())
if _, err := store.Reader(ctx, 0); err != nil {
t.Fatal(err)
}
@ -123,12 +125,11 @@ func TestTranslateStore_Reader(t *testing.T) {
translateStore.ReaderFunc = func(ctx context.Context, off int64) (io.ReadCloser, error) {
return nil, pilosa.ErrNotImplemented
}
h := test.MustNewHandler()
h.API.TranslateStore = &translateStore
s := httptest.NewServer(h)
defer s.Close()
_, err := http.NewTranslateStore(s.URL).Reader(context.Background(), 0)
opts := server.OptCommandServerOptions(pilosa.OptServerPrimaryTranslateStore(translateStore))
main := test.MustRunMainWithCluster(t, 1, []server.CommandOption{opts})[0]
_, err := http.NewTranslateStore(main.Server.URI.String()).Reader(context.Background(), 0)
if err != pilosa.ErrNotImplemented {
t.Fatalf("unexpected error: %s", err)
}

10
mock/translator.go
View file

@ -17,22 +17,22 @@ type TranslateStore struct {
ReaderFunc func(ctx context.Context, off int64) (io.ReadCloser, error)
}
func (s *TranslateStore) TranslateColumnsToUint64(index string, values []string) ([]uint64, error) {
func (s TranslateStore) TranslateColumnsToUint64(index string, values []string) ([]uint64, error) {
return s.TranslateColumnsToUint64Func(index, values)
}
func (s *TranslateStore) TranslateColumnToString(index string, values uint64) (string, error) {
func (s TranslateStore) TranslateColumnToString(index string, values uint64) (string, error) {
return s.TranslateColumnToStringFunc(index, values)
}
func (s *TranslateStore) TranslateRowsToUint64(index, frame string, values []string) ([]uint64, error) {
func (s TranslateStore) TranslateRowsToUint64(index, frame string, values []string) ([]uint64, error) {
return s.TranslateRowsToUint64Func(index, frame, values)
}
func (s *TranslateStore) TranslateRowToString(index, frame string, value uint64) (string, error) {
func (s TranslateStore) TranslateRowToString(index, frame string, value uint64) (string, error) {
return s.TranslateRowToStringFunc(index, frame, value)
}
func (s *TranslateStore) Reader(ctx context.Context, off int64) (io.ReadCloser, error) {
func (s TranslateStore) Reader(ctx context.Context, off int64) (io.ReadCloser, error) {
return s.ReaderFunc(ctx, off)
}

3
server/handler_test.go
View file

@ -31,6 +31,7 @@ import (
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/http"
"github.com/pilosa/pilosa/internal"
"github.com/pilosa/pilosa/server"
"github.com/pilosa/pilosa/test"
)
@ -565,7 +566,7 @@ func TestHandler_Endpoints(t *testing.T) {
t.Fatalf("CORS preflight status should be 405, but is %v", result.StatusCode)
}
clus := test.MustRunMainWithCluster(t, 1, test.OptAllowedOrigins([]string{"http://test/"}))
clus := test.MustRunMainWithCluster(t, 1, []server.CommandOption{test.OptAllowedOrigins([]string{"http://test/"})})
w = httptest.NewRecorder()
h := clus[0].Handler.(*http.Handler).Handler
h.ServeHTTP(w, req)

34
server/server.go
View file

@ -76,11 +76,22 @@ type Command struct {
Handler pilosa.Handler
ln net.Listener
serverOptions []pilosa.ServerOption
}
type CommandOption func(c *Command) error
func OptCommandServerOptions(opts ...pilosa.ServerOption) CommandOption {
return func(c *Command) error {
c.serverOptions = append(c.serverOptions, opts...)
return nil
}
}
// NewCommand returns a new instance of Main.
func NewCommand(stdin io.Reader, stdout, stderr io.Writer) *Command {
return &Command{
func NewCommand(stdin io.Reader, stdout, stderr io.Writer, opts ...CommandOption) *Command {
c := &Command{
Config: NewConfig(),
CmdIO: pilosa.NewCmdIO(stdin, stdout, stderr),
@ -88,6 +99,16 @@ func NewCommand(stdin io.Reader, stdout, stderr io.Writer) *Command {
Started: make(chan struct{}),
done: make(chan struct{}),
}
for _, opt := range opts {
err := opt(c)
if err != nil {
panic(err)
// TODO: Return error instead of panic?
}
}
return c
}
// Start starts the pilosa server - it returns once the server is running.
@ -225,7 +246,7 @@ func (m *Command) SetupServer() error {
primaryTranslateStore = http.NewTranslateStore(m.Config.Translation.PrimaryURL)
}
m.Server, err = pilosa.NewServer(
serverOptions := []pilosa.ServerOption{
pilosa.OptServerAntiEntropyInterval(time.Duration(m.Config.AntiEntropy.Interval)),
pilosa.OptServerLongQueryTime(time.Duration(m.Config.Cluster.LongQueryTime)),
pilosa.OptServerDataDir(m.Config.DataDir),
@ -243,7 +264,12 @@ func (m *Command) SetupServer() error {
pilosa.OptServerInternalClient(http.NewInternalClientFromURI(uri, c)),
pilosa.OptServerPrimaryTranslateStore(primaryTranslateStore),
pilosa.OptServerClusterDisabled(m.Config.Cluster.Disabled, m.Config.Cluster.Hosts),
)
}
serverOptions = append(serverOptions, m.serverOptions...)
m.Server, err = pilosa.NewServer(serverOptions...)
if err != nil {
return errors.Wrap(err, "new server")
}

3
server_test.go
View file

@ -20,6 +20,7 @@ import (
"time"
"github.com/pilosa/pilosa"
"github.com/pilosa/pilosa/server"
"github.com/pilosa/pilosa/test"
)
@ -27,7 +28,7 @@ import (
// pilosa.Server was not having its remoteClient field set by an option and so
// it was using a nil client in monitorAntiEntropy.
func TestMonitorAntiEntropy(t *testing.T) {
cluster := test.MustRunMainWithCluster(t, 3, test.OptAntiEntropyInterval(time.Millisecond*20))
cluster := test.MustRunMainWithCluster(t, 3, []server.CommandOption{test.OptAntiEntropyInterval(time.Millisecond * 20)})
client := cluster[1].Client()
err := client.CreateIndex(context.Background(), "balh", pilosa.IndexOptions{})
if err != nil {

43
test/pilosa.go
View file

@ -34,52 +34,46 @@ import (
)
////////////////////////////////////////////////////////////////////////////////////
// Main represents a test wrapper for main.Main.
// Main represents a test wrapper for server.Command.
type Main struct {
*server.Command
commandOptions []server.CommandOption
Stdin bytes.Buffer
Stdout bytes.Buffer
Stderr bytes.Buffer
}
type MainOpt func(m *Main) error
func OptAntiEntropyInterval(dur time.Duration) MainOpt {
return func(m *Main) error {
m.Command.Config.AntiEntropy.Interval = toml.Duration(dur)
func OptAntiEntropyInterval(dur time.Duration) server.CommandOption {
return func(m *server.Command) error {
m.Config.AntiEntropy.Interval = toml.Duration(dur)
return nil
}
}
func OptAllowedOrigins(origins []string) MainOpt {
return func(m *Main) error {
func OptAllowedOrigins(origins []string) server.CommandOption {
return func(m *server.Command) error {
m.Config.Handler.AllowedOrigins = origins
return nil
}
}
// NewMain returns a new instance of Main with a temporary data directory and random port.
func NewMain(opts ...MainOpt) *Main {
func NewMain(opts ...server.CommandOption) *Main {
path, err := ioutil.TempDir("", "pilosa-")
if err != nil {
panic(err)
}
m := &Main{Command: server.NewCommand(os.Stdin, os.Stdout, os.Stderr)}
m := &Main{Command: server.NewCommand(os.Stdin, os.Stdout, os.Stderr, opts...), commandOptions: opts}
m.Config.DataDir = path
m.Config.Bind = "http://localhost:0"
m.Config.Cluster.Disabled = true
m.Command.Stdin = &m.Stdin
m.Command.Stdout = &m.Stdout
m.Command.Stderr = &m.Stderr
for _, opt := range opts {
err := opt(m)
if err != nil {
panic(err)
}
}
err = m.SetupServer()
if err != nil {
panic(err)
@ -94,7 +88,7 @@ func NewMain(opts ...MainOpt) *Main {
}
// NewMainWithCluster returns a new instance of Main with clustering enabled.
func NewMainWithCluster(isCoordinator bool, opts ...MainOpt) *Main {
func NewMainWithCluster(isCoordinator bool, opts ...server.CommandOption) *Main {
m := NewMain(opts...)
m.Config.Cluster.Disabled = false
m.Config.Cluster.Coordinator = isCoordinator
@ -103,7 +97,7 @@ func NewMainWithCluster(isCoordinator bool, opts ...MainOpt) *Main {
// MustRunMainWithCluster ruturns a running array of *Main where
// all nodes are joined via memberlist (i.e. clustering enabled).
func MustRunMainWithCluster(t *testing.T, size int, opts ...MainOpt) []*Main {
func MustRunMainWithCluster(t *testing.T, size int, opts ...[]server.CommandOption) []*Main {
ma, err := runMainWithCluster(size, opts...)
if err != nil {
t.Fatalf("new main array with cluster: %v", err)
@ -113,10 +107,13 @@ func MustRunMainWithCluster(t *testing.T, size int, opts ...MainOpt) []*Main {
// runMainWithCluster runs an array of *Main where all nodes are
// joined via memberlist (i.e. clustering enabled).
func runMainWithCluster(size int, opts ...MainOpt) ([]*Main, error) {
func runMainWithCluster(size int, opts ...[]server.CommandOption) ([]*Main, error) {
if size == 0 {
return nil, errors.New("cluster must contain at least one node")
}
if len(opts) != size && len(opts) != 0 && len(opts) != 1 {
return nil, errors.New("Slice of CommandOptions must be of length 0, 1, or equal to the number of cluster nodes")
}
mains := make([]*Main, size)
@ -126,7 +123,11 @@ func runMainWithCluster(size int, opts ...MainOpt) ([]*Main, error) {
var gossipSeeds = make([]string, size)
for i := 0; i < size; i++ {
m := NewMainWithCluster(i == 0, opts...)
var commandOpts []server.CommandOption
if len(opts) > 0 {
commandOpts = opts[i%len(opts)]
}
m := NewMainWithCluster(i == 0, commandOpts...)
m.Config.Cluster.Disabled = false
gossipSeeds[i], err = m.RunWithTransport(gossipHost, gossipPort, gossipSeeds[:i])
@ -164,7 +165,7 @@ func (m *Main) Reopen() error {
// Create new main with the same config.
config := m.Command.Config
m.Command = server.NewCommand(os.Stdin, os.Stdout, os.Stderr)
m.Command = server.NewCommand(os.Stdin, os.Stdout, os.Stderr, m.commandOptions...)
m.Command.Config = config
err := m.SetupServer()
if err != nil {