Observer refactoring

Author: lukasz-zimnoch

pkg/metrics/observer.go

@@ -5,19 +5,23 @@ import (
 	"time"
 )
 
-// Source defines a source of metric data.
-type Source func() float64
+// ObserverInput defines a source of metric data.
+type ObserverInput func() float64
 
-// Sink defines a destination of collected metric data.
-type Sink interface {
+// ObserverOutput defines a destination of collected metric data.
+type ObserverOutput interface {
 	Set(value float64)
 }
 
-// Observe triggers a cyclic metric observation goroutine.
-func Observe(
+// Observer represent a definition of a cyclic metric observation process.
+type Observer struct {
+	input  ObserverInput
+	output ObserverOutput
+}
+
+// Observe triggers a cyclic metric observation process.
+func (o *Observer) Observe(
 	ctx context.Context,
-	source Source,
-	sink Sink,
 	tick time.Duration,
 ) {
 	go func() {
@@ -27,7 +31,7 @@ func Observe(
 		for {
 			select {
 			case <-ticker.C:
-				sink.Set(source())
+				o.output.Set(o.input())
 			case <-ctx.Done():
 				return
 			}

pkg/metrics/observer_test.go

@@ -7,17 +7,15 @@ import (
 )
 
 func TestObserve(t *testing.T) {
+	input := func() float64 {
+		return 5000
+	}
 	gauge := &Gauge{}
 	ctx, _ := context.WithTimeout(context.Background(), 5*time.Millisecond)
 
-	Observe(
-		ctx,
-		func() float64 {
-			return 5000
-		},
-		gauge,
-		1*time.Millisecond,
-	)
+	observer := &Observer{input, gauge}
+
+	observer.Observe(ctx, 1*time.Millisecond)
 
 	<-ctx.Done()
 

pkg/metrics/registry.go

@@ -100,3 +100,18 @@ func (r *Registry) NewGauge(name string) (*Gauge, error) {
 	r.metrics[name] = gauge
 	return gauge, nil
 }
+
+func (r *Registry) NewGaugeObserver(
+	name string,
+	input ObserverInput,
+) (*Observer, error) {
+	gauge, err := r.NewGauge(name)
+	if err != nil {
+		return nil, err
+	}
+
+	return &Observer{
+		input:  input,
+		output: gauge,
+	}, nil
+}