cleanup and callbacks

Author: [email protected]

README.md

@@ -19,24 +19,28 @@ import (
 
 func main() {
 
-	// the update function
-	f := func(x interface{}) interface{} {
-		return fmt.Sprintf("%s-", x)
+	// the function to call
+	var update = func(x interface{}) (interface{}, error) {
+		y := x
+		if y == nil { y = "" }
+		return fmt.Sprintf("%s-", y), nil
 	}
 
-	// creating the new resource
-	r := updatingresource.NewUpdatingResource("-", f, 500 * time.Millisecond)
-
-	// query the resource 8 times
-	for i := 0; i <8; i++{
-		time.Sleep(200 * time.Millisecond)
-		x := r.Get().(string)
-		fmt.Printf("%s\n", x)
-		
-		// stop updating after the 6th time 
-		if i == 6 {
-			r.Done()
-		}
+	// the resource config
+	config := updatingresource.Config{
+		Name:     "dashes",
+		Update:   update,
+		Interval: 20 * time.Millisecond,
 	}
+	resource := config.NewResource()
+
+	fmt.Printf("%s\n", resource.Get().(string))
+
+	time.Sleep(30 * time.Millisecond)
+	fmt.Printf("%s\n", resource.Get().(string))
+
+	time.Sleep(20 * time.Millisecond)
+	fmt.Printf("%s\n", resource.Get().(string))
+
 }
 ~~~~

example_basic_test.go

@@ -9,33 +9,34 @@ import (
 
 func Example() {
 
-	// the update function
-	f := func(x interface{}) interface{} {
-		return fmt.Sprintf("%s-", x)
+	// the function to call
+	var update = func(x interface{}) (interface{}, error) {
+		y := x
+		if y == nil { y = "" }
+		return fmt.Sprintf("%s-", y), nil
 	}
 
-	// creating the new resource
-	r := updatingresource.NewUpdatingResource("-", f, 500 * time.Millisecond)
+	// the resource config
+	config := updatingresource.Config{
+		Name:     "dashes",
+		Update:   update,
+		Interval: 300 * time.Millisecond,
+	}
+	resource := config.NewResource()
 
-	// query the resource 8 times
-	for i := 0; i <8; i++{
-		time.Sleep(200 * time.Millisecond)
-		x := r.Get().(string)
-		fmt.Printf("%s\n", x)
+	fmt.Printf("%s\n", resource.Get().(string))
 
-		// stop updating after the 6th time
-		if i == 6 {
-			r.Done()
-		}
-	}
+	resource.Tick()
+	time.Sleep(50 * time.Millisecond)
+
+	fmt.Printf("%s\n", resource.Get().(string))
+
+	time.Sleep(450 * time.Millisecond)
+
+	fmt.Printf("%s\n", resource.Get().(string))
 
 	// Output:
 	// -
-	// -
-	// --
 	// --
 	// ---
-	// ---
-	// ---
-	// ---
 }

go.mod

@@ -1,3 +1,4 @@
 module github.com/sebogh/updating-resource
 
 go 1.15
+

updatingresource.go

@@ -5,46 +5,122 @@ import (
 	"time"
 )
 
-// UpdatingResource is a structure to wrap an object x which is regularly and
-// asynchronously computed / updated.
-type UpdatingResource struct {
+// Config is the configuration for a Resource.
+type Config struct {
+
+	// Name of this resource.
+	Name string
+
+	// Update is the function to update the wrapped object. Update must handle nil
+	// values.
+	Update func(x interface{}) (interface{}, error)
+
+	// Interval is the time to wait between between automatic updates. If 0 (or not
+	// set), updates will never trigger automatically.
+	Interval time.Duration
+
+	// Success is the function (if any) to call, if the object was successfully
+	// updated. Success is called with the new value the name of the resource.
+	Success func(x interface{}, name string)
+
+	// Success is the function (if any) to call, if the object was successfully
+	// updated. Error is called with the the error that occurred the name of the
+	// resource.
+	Error func(e error, name string)
+}
+
+// Resource regularly, asynchronously updates a wrapped object.
+type Resource struct {
+
+	// resource configuration
+	*Config
+
+	// mutex for protecting object updates
 	mu *sync.RWMutex
-	x  *interface{}
+
+	// the wrapped object
+	x *interface{}
+
+	// a channel for terminating the updates
 	done chan bool
+
+	// channel to manually trigger an update
+	tick chan bool
+
 }
 
-// NewUpdatingResource creates a new UpdatingResource. Thereby, f is the function
-// that will be called every ttl to compute a new value for x (i.e. x=f(x)).
-func NewUpdatingResource(x interface{}, f func(x interface{}) interface{}, ttl time.Duration) *UpdatingResource {
+// NewResource creates a new Resource.
+func (c *Config) NewResource() *Resource {
 	var mu sync.RWMutex
+
 	done := make(chan bool)
-	go func(f func(x interface{}) interface{}) {
-		ticker := time.NewTicker(ttl)
+	tick := make(chan bool)
+
+	var startValue interface{} = nil
+	x := &startValue
+
+	var timeTicker <-chan time.Time
+	if c.Interval > 0 {
+		timeTicker = time.NewTicker(c.Interval).C
+	}
+
+	go func(f func(x interface{}) (interface{}, error)) {
+
+		// wrap f to be used for inputs on multiple channels
+		var fWrapper = func() {
+			y, err := f(*x)
+			if err != nil && c.Error != nil {
+				c.Error(err, c.Name)
+			} else {
+				if c.Success != nil {
+					c.Success(y, c.Name)
+				}
+				mu.Lock()
+				*x = y
+				mu.Unlock()
+			}
+		}
+
+		// wait (forever) on channels
 		for {
 			select {
 			case <-done:
 				return
-			case <-ticker.C:
-				y := f(x)
-				mu.Lock()
-				x = y
-				mu.Unlock()
+			case <-timeTicker:
+				fWrapper()
+			case <-tick:
+				fWrapper()
 			}
 		}
-	}(f)
-	resource := UpdatingResource{x: &x, mu: &mu, done: done}
+	}(c.Update)
+
+	// compute step 0
+	tick <- true
+
+	resource := Resource{
+		Config: c,
+		mu:     &mu,
+		x:      x,
+		done:   done,
+		tick:   tick,
+	}
 	return &resource
 }
 
 // Get returns the current value of the wrapped object. Get is thread-safe
 // wrt. to the function updating the encapsulated object.
-func (r *UpdatingResource) Get() interface{} {
+func (r *Resource) Get() interface{} {
 	r.mu.RLock()
 	defer r.mu.RUnlock()
 	return *r.x
 }
 
 // Done stops the updating of the wrapped object.
-func (r *UpdatingResource) Done() {
+func (r *Resource) Done() {
 	r.done <- true
-}
+}
+
+// Tick manually trigger an update.
+func (r *Resource) Tick() {
+	r.tick <- true
+}