Played around with a retry queue

Author: kingledion

queue/README.md

@@ -4,4 +4,4 @@ A single ended queue with push and pop functionality implemented with channels f
 
 Because this queue is type specific, it should probably not be imported, and instead simply implemented in a project that uses this. Copy the implementation and retype as needed. For this reason, the struct is implemented as private.
 
-There are optional add-ons for this implementation such as a queue counter to prevent the queue from being cycled through excessively. 
+There are optional add-ons for this implementation such as a queue counter to prevent the queue from being cycled through excessively. An example implementation is included in `run.go`.

retry/README.md

@@ -0,0 +1,5 @@
+# retry
+
+Implements a retry queue with an input channel and internal buffered channel. The retry queue will continually attempt to retry a given action on the elements of that queue, until both no more input has been recieved for a number of seconds and the length of the queue is unchanged for at least one cycle through the queue. 
+
+Because this queue is type specific, it should probably not be imported, and instead simply implemented in a project that uses this. Copy the implementation and retype as needed. For this reason, the struct is implemented as private and typed as integer.

retry/retry.go

@@ -0,0 +1,119 @@
+package retry
+
+import (
+	"fmt"
+	"sync"
+	"time"
+)
+
+// Specifies an element of the retry queue. The element must have methods that process
+// and record failure to process. The processing method will return a boolean representing
+// whether processing should be tried again.
+type Element interface {
+	// Process this element. Return true if the process shoud be tried again.
+	Process() bool
+	// Record failure to process.
+	Fail()
+}
+
+// Retry is a queue with an automatic retry capability. Elements added to the queue are
+// processed and retried until the input is closed and
+type Retry struct {
+	input chan Element
+	retry chan Element
+	wait  sync.WaitGroup
+}
+
+// Build an empty retry queue
+func New(capacity uint) *Retry {
+	r := &Retry{
+		input: make(chan Element),
+		retry: make(chan Element, capacity),
+	}
+	r.run()
+	return r
+}
+
+// Add a new element to the retry queue
+func (r *Retry) Push(e Element) {
+	r.input <- e
+}
+
+// Close channel, indicating that no further input will be passed
+func (r *Retry) Close() {
+	close(r.input)
+}
+
+func (r *Retry) Wait() {
+	r.wait.Wait()
+}
+
+// run starts the processing of elements
+func (r *Retry) run() {
+
+	inputTimeout := make(chan bool, 1)
+
+	// input goroutine
+	go func() {
+		for elem := range r.input {
+			tryagain := elem.Process()
+			if tryagain {
+				r.retry <- elem
+			}
+		}
+		inputTimeout <- true
+	}()
+
+	// retry goroutine
+	go func() {
+
+		r.wait.Add(1)
+
+		cntr := 0
+		rlen := 0
+		inputDone := false
+		retryDone := false
+
+		for elem := range r.retry {
+
+			// if done, process all remaining elements
+			if retryDone {
+				elem.Fail()
+				continue
+			}
+
+			// process current element
+			tryagain := elem.Process()
+			if tryagain {
+				r.retry <- elem
+			}
+
+			// if the input channel is closed, start monitoring for shutdown conditions
+			select {
+			case <-inputTimeout:
+				inputDone = true
+				rlen = len(r.retry)
+			case <-time.After(1 * time.Second):
+				continue
+			default:
+			}
+
+			// if input is done, loop through all elements once then shutdown
+			if inputDone {
+				fmt.Printf("Input closed, loop %d\n", cntr)
+				cntr = cntr + 1
+
+				if cntr > rlen {
+					retryDone = true
+					close(r.retry)
+				}
+
+			}
+
+		}
+
+		r.wait.Done()
+
+	}()
+
+}

retry/retry_test.go

@@ -0,0 +1,59 @@
+package retry
+
+import (
+	"fmt"
+	"math/rand"
+	"testing"
+	"time"
+)
+
+type example struct {
+	i int
+}
+
+func (e *example) Process() bool {
+	if rand.Intn(3) < 1 {
+		fmt.Println("Fail on", e.i)
+		return true
+	}
+	fmt.Println("Success on", e.i)
+	return false
+}
+
+func (e *example) Fail() {
+	fmt.Println("Abandoning", e.i)
+}
+
+func TestRetry(t *testing.T) {
+
+	fmt.Println("Starting test")
+
+	rand.Seed(time.Now().UnixNano())
+
+	// setup
+
+	capacity := uint(10)
+
+	genelem := func() *example {
+		return &example{rand.Int()}
+	}
+
+	count := 100
+
+	// execution
+
+	r := New(capacity)
+
+	for i := 0; i < count; i++ {
+		e := genelem()
+		fmt.Println("New element", e.i)
+		r.Push(e)
+	}
+
+	r.Close()
+
+	fmt.Println("Waiting")
+
+	r.Wait()
+
+}