Merge pull request #45 from fern4lvarez/rotate-when-full

Add option to allow pushing on a Stack when this is full

config/value.go

@@ -42,6 +42,13 @@ func (c *Config) Port() int {
 	return t
 }
 
+// PushWhenFull returns the value of PUSH_WHEN_FULL.
+// Type: bool, Default: false
+func (c *Config) PushWhenFull() bool {
+	pushWhenFull := c.Get(vars.PushWhenFull)
+	return boolValue(pushWhenFull, vars.PushWhenFullDefault)
+}
+
 // intValue returns an Integer value given another value as an
 // interface. If conversion fails, a default value is used.
 func intValue(value interface{}, defaultValue int) int {
@@ -63,3 +70,22 @@ func intValue(value interface{}, defaultValue int) int {
 		return defaultValue
 	}
 }
+
+// boolValue returns a Boolean value given another value as an
+// interface. If conversion fails, a default value is used.
+func boolValue(value interface{}, defaultValue bool) bool {
+	switch value.(type) {
+	case bool:
+		return value.(bool)
+	case string:
+		if value == "true" {
+			return true
+		}
+		if value == "false" {
+			return false
+		}
+		return defaultValue
+	default:
+		return defaultValue
+	}
+}

config/value_test.go

@@ -104,3 +104,28 @@ func TestPort(t *testing.T) {
 		}
 	}
 }
+
+func TestPushWhenFull(t *testing.T) {
+	c := NewConfig()
+
+	inputOutput := []struct {
+		input  interface{}
+		output bool
+	}{
+		{true, true},
+		{false, false},
+		{"true", true},
+		{"false", false},
+		{"foo", vars.PushWhenFullDefault},
+		{42, vars.PushWhenFullDefault},
+		{[]byte("true"), vars.PushWhenFullDefault},
+	}
+
+	for _, io := range inputOutput {
+		c.Set(vars.PushWhenFull, io.input)
+
+		if s := c.PushWhenFull(); s != io.output {
+			t.Errorf("PushWhenFull is %v, expected %v", s, io.output)
+		}
+	}
+}

config/vars/vars.go

@@ -34,6 +34,14 @@ const (
 	// PortDefault represents the default value
 	// of Port.
 	PortDefault = 1205
+
+	// PushWhenFull enables to keep pushing elements
+	// into a Stack, even if this is full. If this is
+	// the case, the base element will be deleted.
+	PushWhenFull = "PUSH_WHEN_FULL"
+	// PushWhenFullDefault represents the default value
+	// of PushWhenFull.
+	PushWhenFullDefault = false
 )
 
 // Env returns the environment variable name
@@ -57,3 +65,13 @@ func DefaultInt(name string) int {
 	}
 	return -1
 }
+
+// DefaultBool returns the default value of a config
+// name of boolean type.
+func DefaultBool(name string) bool {
+	switch name {
+	case PushWhenFull:
+		return PushWhenFullDefault
+	}
+	return false
+}

config/vars/vars_test.go

@@ -29,3 +29,19 @@ func TestDefaultInt(t *testing.T) {
 		}
 	}
 }
+
+func TestDefaultBool(t *testing.T) {
+	inputOutput := []struct {
+		input  string
+		output bool
+	}{
+		{PushWhenFull, PushWhenFullDefault},
+		{"foo", false},
+	}
+
+	for _, io := range inputOutput {
+		if o := DefaultBool(io.input); o != io.output {
+			t.Errorf("DefaultBool is %v, expected %v", o, io.output)
+		}
+	}
+}

pila/stack.go

@@ -68,11 +68,24 @@ func (s *Stack) Push(element interface{}) {
 }
 
 // Pop removes and returns the element on top of the Stack.
-// If the Stack was empty, it returns false.
+// If the Stack is empty, it returns false.
 func (s *Stack) Pop() (interface{}, bool) {
 	return s.base.Pop()
 }
 
+// Sweep removes and returns the bottommost element of the Stack.
+// If the Stack is empty, it returns false.
+func (s *Stack) Sweep() (interface{}, bool) {
+	return s.base.Sweep()
+}
+
+// SweepPush removes and returns the bottommost element of the Stack,
+// and pushes an element on top of it, as an atomic operation.
+// If the Stack is empty, it returns false.
+func (s *Stack) SweepPush(element interface{}) (interface{}, bool) {
+	return s.base.SweepPush(element)
+}
+
 // Size returns the size of the Stack.
 func (s *Stack) Size() int {
 	return s.base.Size()

pila/stack_test.go

@@ -9,11 +9,13 @@ import (
 
 type TestBaseStack struct{}
 
-func (s *TestBaseStack) Push(element interface{}) { return }
-func (s *TestBaseStack) Pop() (interface{}, bool) { return nil, false }
-func (s *TestBaseStack) Size() int                { return 0 }
-func (s *TestBaseStack) Peek() interface{}        { return nil }
-func (s *TestBaseStack) Flush()                   { return }
+func (s *TestBaseStack) Push(element interface{})                          { return }
+func (s *TestBaseStack) Pop() (interface{}, bool)                          { return nil, false }
+func (s *TestBaseStack) Sweep() (interface{}, bool)                        { return nil, false }
+func (s *TestBaseStack) SweepPush(element interface{}) (interface{}, bool) { return nil, false }
+func (s *TestBaseStack) Size() int                                         { return 0 }
+func (s *TestBaseStack) Peek() interface{}                                 { return nil }
+func (s *TestBaseStack) Flush()                                            { return }
 
 func TestNewStack(t *testing.T) {
 	now := time.Now()
@@ -122,6 +124,62 @@ func TestStackPop_False(t *testing.T) {
 	}
 }
 
+func TestStackSweep(t *testing.T) {
+	stack := NewStack("test-stack", time.Now())
+	stack.Push("test")
+	stack.Push(8)
+
+	element, ok := stack.Sweep()
+	if !ok {
+		t.Errorf("stack.Sweep() not ok")
+	}
+	if element != "test" {
+		t.Errorf("element is %v, expected %v", element, "test")
+	}
+	if stack.Peek() != 8 {
+		t.Errorf("stack.Peek() is %v, expected %v", stack.Peek(), 8)
+	}
+	if stack.Size() != 1 {
+		t.Errorf("stack.Size() is %d, expected %d", stack.Size(), 1)
+	}
+}
+
+func TestStackSweep_False(t *testing.T) {
+	stack := NewStack("test-stack", time.Now())
+	_, ok := stack.Sweep()
+	if ok {
+		t.Error("stack.Sweep() is ok")
+	}
+}
+
+func TestStackSweepPush(t *testing.T) {
+	stack := NewStack("test-stack", time.Now())
+	stack.Push("test")
+	stack.Push(8)
+
+	element, ok := stack.SweepPush("foo")
+	if !ok {
+		t.Errorf("stack.Sweep() not ok")
+	}
+	if element != "test" {
+		t.Errorf("element is %v, expected %v", element, "test")
+	}
+	if stack.Peek() != "foo" {
+		t.Errorf("stack.Peek() is %v, expected %v", stack.Peek(), "foo")
+	}
+	if stack.Size() != 2 {
+		t.Errorf("stack.Size() is %d, expected %d", stack.Size(), 2)
+	}
+}
+
+func TestStackSweepPush_False(t *testing.T) {
+	stack := NewStack("test-stack", time.Now())
+	_, ok := stack.SweepPush(8)
+	if ok {
+		t.Error("stack.SweepPush(8) is ok")
+	}
+}
+
 func TestStackSize(t *testing.T) {
 	stack := NewStack("test-stack", time.Now())
 	if stack.Size() != 0 {

pilad/README.md

@@ -322,6 +322,8 @@ is used as default, the latter as fallback.
 }
 ```
 
+Returns `406 NOT ACCEPTABLE` if the stack is full.
+
 Returns `410 GONE` if the database or stack do not exist.
 
 Returns `400 BAD REQUEST` if there's an error serializing the element.

pilad/config.go

@@ -20,6 +20,7 @@ var (
 	maxStackSizeFlag                  int
 	readTimeoutFlag, writeTimeoutFlag int
 	portFlag                          int
+	pushWhenFullFlag                  bool
 	versionFlag                       bool
 )
 
@@ -28,6 +29,7 @@ func init() {
 	flag.IntVar(&readTimeoutFlag, "read-timeout", vars.ReadTimeoutDefault, "Read request timeout")
 	flag.IntVar(&writeTimeoutFlag, "write-timeout", vars.WriteTimeoutDefault, "Write response timeout")
 	flag.IntVar(&portFlag, "port", vars.PortDefault, "Port number")
+	flag.BoolVar(&pushWhenFullFlag, "push-when-full", vars.PushWhenFullDefault, "Allow push when Stack is full")
 	flag.BoolVar(&versionFlag, "v", false, "Version")
 }
 
@@ -44,15 +46,17 @@ func (c *Conn) buildConfig() {
 		{readTimeoutFlag, vars.ReadTimeout},
 		{writeTimeoutFlag, vars.WriteTimeout},
 		{portFlag, vars.Port},
+		{pushWhenFullFlag, vars.PushWhenFull},
 	}
 
 	for _, fk := range flagKeys {
 		if e := os.Getenv(vars.Env(fk.key)); e != "" {
-			if i, err := strconv.Atoi(e); err != nil {
+			if e == "true" || e == "false" {
+				c.Config.Set(fk.key, e)
+			} else if i, err := strconv.Atoi(e); err != nil {
 				c.Config.Set(fk.key, vars.DefaultInt(fk.key))
 			} else {
 				c.Config.Set(fk.key, i)
-
 			}
 			continue
 		}
@@ -133,11 +137,19 @@ func (c *Conn) configKeyHandler(configKey string) http.Handler {
 	})
 }
 
-// checkMaxStackSize checks config value for MaxStackSize and execute the
-// wrapped handler if check is validated.
+// checkMaxStackSize checks config value for MaxStackSize and PushWhenFull
+//and execute the wrapped handler if check is validated.
 func (c *Conn) checkMaxStackSize(handler stackHandlerFunc) stackHandlerFunc {
 	return func(w http.ResponseWriter, r *http.Request, stack *pila.Stack) {
 		if s := c.Config.MaxStackSize(); stack.Size() >= s && s != -1 {
+			if c.Config.PushWhenFull() {
+				// set internal config value to share with
+				// the next handler that we want to sweep
+				// before pushing
+				c.Config.Set("SWEEP_BEFORE_PUSH", true)
+				handler(w, r, stack)
+				return
+			}
 			log.Println(r.Method, r.URL, http.StatusNotAcceptable, vars.MaxStackSize, "value reached")
 			w.WriteHeader(http.StatusNotAcceptable)
 			return