remove removecallback from interface (viamrobotics#3893)

components/board/client.go

@@ -204,9 +204,6 @@ func (dic *digitalInterruptClient) Name() string {
 	return dic.digitalInterruptName
 }
 
-func (dic *digitalInterruptClient) RemoveCallback(ch chan Tick) {
-}
-
 func (c *client) StreamTicks(ctx context.Context, interrupts []DigitalInterrupt, ch chan Tick,
 	extra map[string]interface{},
 ) error {

components/board/digital_interrupts.go

@@ -19,13 +19,10 @@ type Tick struct {
 // A DigitalInterrupt represents a configured interrupt on the board that
 // when interrupted, calls the added callbacks.
 type DigitalInterrupt interface {
+	// Name returns the name of the interrupt.
+	Name() string
+
 	// Value returns the current value of the interrupt which is
 	// based on the type of interrupt.
 	Value(ctx context.Context, extra map[string]interface{}) (int64, error)
-
-	// RemoveCallback removes a listener for interrupts.
-	RemoveCallback(c chan Tick)
-
-	// Name returns the name of the interrupt.
-	Name() string
 }

components/board/fake/board.go

@@ -399,10 +399,6 @@ func (s *DigitalInterrupt) reset(conf board.DigitalInterruptConfig) {
 	s.conf = conf
 }
 
-// RemoveCallback removes a listener for interrupts.
-func (s *DigitalInterrupt) RemoveCallback(c chan board.Tick) {
-}
-
 // Value returns the current value of the interrupt which is
 // based on the type of interrupt.
 func (s *DigitalInterrupt) Value(ctx context.Context, extra map[string]interface{}) (int64, error) {

components/board/genericlinux/board.go

@@ -15,7 +15,7 @@ import (
 	"github.com/pkg/errors"
 	"go.uber.org/multierr"
 	pb "go.viam.com/api/component/board/v1"
-	goutils "go.viam.com/utils"
+	"go.viam.com/utils"
 
 	"go.viam.com/rdk/components/board"
 	"go.viam.com/rdk/components/board/genericlinux/buses"
@@ -368,7 +368,7 @@ func (a *wrappedAnalogReader) reset(ctx context.Context, chipSelect string, read
 	a.mu.Lock()
 	defer a.mu.Unlock()
 	if a.reader != nil {
-		goutils.UncheckedError(a.reader.Close(ctx))
+		utils.UncheckedError(a.reader.Close(ctx))
 	}
 	a.reader = reader
 	a.chipSelect = chipSelect
@@ -499,6 +499,20 @@ func (b *Board) StreamTicks(ctx context.Context, interrupts []board.DigitalInter
 	for _, i := range interrupts {
 		pinwrappers.AddCallback(i.(*pinwrappers.BasicDigitalInterrupt), ch)
 	}
+
+	b.activeBackgroundWorkers.Add(1)
+
+	utils.ManagedGo(func() {
+		// Wait until it's time to shut down then remove callbacks.
+		select {
+		case <-ctx.Done():
+		case <-b.cancelCtx.Done():
+		}
+		for _, i := range interrupts {
+			pinwrappers.RemoveCallback(i.(*pinwrappers.BasicDigitalInterrupt), ch)
+		}
+	}, b.activeBackgroundWorkers.Done)
+
 	return nil
 }
 

components/board/pi/impl/digital_interrupts.go

@@ -85,16 +85,6 @@ func (i *BasicDigitalInterrupt) Value(ctx context.Context, extra map[string]inte
 	return count, nil
 }
 
-// Ticks is really just for testing.
-func (i *BasicDigitalInterrupt) Ticks(ctx context.Context, num int, now uint64) error {
-	for x := 0; x < num; x++ {
-		if err := Tick(ctx, i, true, now+uint64(x)); err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
 // Tick records an interrupt and notifies any interested callbacks. See comment on
 // the DigitalInterrupt interface for caveats.
 func Tick(ctx context.Context, i *BasicDigitalInterrupt, high bool, nanoseconds uint64) error {
@@ -136,21 +126,6 @@ func RemoveCallback(i *BasicDigitalInterrupt, c chan board.Tick) {
 	}
 }
 
-// RemoveCallback removes a listener for interrupts.
-func (i *BasicDigitalInterrupt) RemoveCallback(c chan board.Tick) {
-	i.mu.Lock()
-	defer i.mu.Unlock()
-	for id := range i.callbacks {
-		if i.callbacks[id] == c {
-			// To remove this item, we replace it with the last item in the list, then truncate the
-			// list by 1.
-			i.callbacks[id] = i.callbacks[len(i.callbacks)-1]
-			i.callbacks = i.callbacks[:len(i.callbacks)-1]
-			break
-		}
-	}
-}
-
 // Name returns the name of the interrupt.
 func (i *BasicDigitalInterrupt) Name() string {
 	i.mu.Lock()
@@ -207,13 +182,6 @@ func ServoTick(ctx context.Context, i *ServoDigitalInterrupt, high bool, now uin
 	return nil
 }
 
-// RemoveCallback currently panics.
-func (i *ServoDigitalInterrupt) RemoveCallback(c chan board.Tick) {
-	i.mu.Lock()
-	defer i.mu.Unlock()
-	panic("servos can't have callback")
-}
-
 // Name returns the name of the interrupt.
 func (i *ServoDigitalInterrupt) Name() string {
 	i.mu.Lock()

components/board/pi/impl/digital_interrupts_test.go

@@ -40,7 +40,7 @@ func TestBasicDigitalInterrupt1(t *testing.T) {
 	test.That(t, intVal, test.ShouldEqual, int64(1))
 
 	c := make(chan board.Tick)
-	AddCallback(i.(*BasicDigitalInterrupt), c)
+	AddCallback(basicInterrupt, c)
 
 	timeNanoSec := nowNanosecondsTest()
 	go func() { Tick(context.Background(), basicInterrupt, true, timeNanoSec) }()
@@ -55,10 +55,10 @@ func TestBasicDigitalInterrupt1(t *testing.T) {
 	test.That(t, v.High, test.ShouldBeTrue)
 	test.That(t, v.TimestampNanosec, test.ShouldEqual, timeNanoSec)
 
-	i.RemoveCallback(c)
+	RemoveCallback(basicInterrupt, c)
 
 	c = make(chan board.Tick, 2)
-	AddCallback(i.(*BasicDigitalInterrupt), c)
+	AddCallback(basicInterrupt, c)
 	go func() {
 		Tick(context.Background(), basicInterrupt, true, uint64(1))
 		Tick(context.Background(), basicInterrupt, true, uint64(4))
@@ -87,7 +87,7 @@ func TestRemoveCallbackDigitalInterrupt(t *testing.T) {
 
 	c1 := make(chan board.Tick)
 	test.That(t, c1, test.ShouldNotBeNil)
-	AddCallback(i.(*BasicDigitalInterrupt), c1)
+	AddCallback(basicInterrupt, c1)
 	var wg sync.WaitGroup
 	wg.Add(1)
 	ret := false
@@ -113,11 +113,11 @@ func TestRemoveCallbackDigitalInterrupt(t *testing.T) {
 	wg.Wait()
 	c2 := make(chan board.Tick)
 	test.That(t, c2, test.ShouldNotBeNil)
-	AddCallback(i.(*BasicDigitalInterrupt), c2)
+	AddCallback(basicInterrupt, c2)
 	test.That(t, ret, test.ShouldBeTrue)
 
-	i.RemoveCallback(c1)
-	i.RemoveCallback(c1)
+	RemoveCallback(basicInterrupt, c1)
+	RemoveCallback(basicInterrupt, c1)
 
 	ret2 := false
 	result := make(chan bool, 1)

components/board/pinwrappers/digital_interrupts.go

@@ -48,16 +48,6 @@ func (i *BasicDigitalInterrupt) Value(ctx context.Context, extra map[string]inte
 	return count, nil
 }
 
-// Ticks is really just for testing.
-func (i *BasicDigitalInterrupt) Ticks(ctx context.Context, num int, now uint64) error {
-	for x := 0; x < num; x++ {
-		if err := Tick(ctx, i, true, now+uint64(x)); err != nil {
-			return err
-		}
-	}
-	return nil
-}
-
 // Tick records an interrupt and notifies any interested callbacks. See comment on
 // the DigitalInterrupt interface for caveats.
 func Tick(ctx context.Context, i *BasicDigitalInterrupt, high bool, nanoseconds uint64) error {
@@ -84,9 +74,10 @@ func AddCallback(i *BasicDigitalInterrupt, c chan board.Tick) {
 }
 
 // RemoveCallback removes a listener for interrupts.
-func (i *BasicDigitalInterrupt) RemoveCallback(c chan board.Tick) {
+func RemoveCallback(i *BasicDigitalInterrupt, c chan board.Tick) {
 	i.mu.Lock()
 	defer i.mu.Unlock()
+
 	for id := range i.callbacks {
 		if i.callbacks[id] == c {
 			// To remove this item, we replace it with the last item in the list, then truncate the

components/board/pinwrappers/digital_interrupts_test.go

@@ -23,6 +23,8 @@ func TestBasicDigitalInterrupt1(t *testing.T) {
 	i, err := CreateDigitalInterrupt(config)
 	test.That(t, err, test.ShouldBeNil)
 
+	di := i.(*BasicDigitalInterrupt)
+
 	intVal, err := i.Value(context.Background(), nil)
 	test.That(t, err, test.ShouldBeNil)
 	test.That(t, intVal, test.ShouldEqual, int64(0))
@@ -36,28 +38,28 @@ func TestBasicDigitalInterrupt1(t *testing.T) {
 	test.That(t, intVal, test.ShouldEqual, int64(1))
 
 	c := make(chan board.Tick)
-	AddCallback(i.(*BasicDigitalInterrupt), c)
+	AddCallback(di, c)
 
 	timeNanoSec := nowNanosecondsTest()
-	go func() { Tick(context.Background(), i.(*BasicDigitalInterrupt), true, timeNanoSec) }()
+	go func() { Tick(context.Background(), di, true, timeNanoSec) }()
 	time.Sleep(1 * time.Microsecond)
 	v := <-c
 	test.That(t, v.High, test.ShouldBeTrue)
 	test.That(t, v.TimestampNanosec, test.ShouldEqual, timeNanoSec)
 
 	timeNanoSec = nowNanosecondsTest()
-	go func() { Tick(context.Background(), i.(*BasicDigitalInterrupt), true, timeNanoSec) }()
+	go func() { Tick(context.Background(), di, true, timeNanoSec) }()
 	v = <-c
 	test.That(t, v.High, test.ShouldBeTrue)
 	test.That(t, v.TimestampNanosec, test.ShouldEqual, timeNanoSec)
 
-	i.RemoveCallback(c)
+	RemoveCallback(di, c)
 
 	c = make(chan board.Tick, 2)
-	AddCallback(i.(*BasicDigitalInterrupt), c)
+	AddCallback(di, c)
 	go func() {
-		Tick(context.Background(), i.(*BasicDigitalInterrupt), true, uint64(1))
-		Tick(context.Background(), i.(*BasicDigitalInterrupt), true, uint64(4))
+		Tick(context.Background(), di, true, uint64(1))
+		Tick(context.Background(), di, true, uint64(4))
 	}()
 	v = <-c
 	v1 := <-c
@@ -72,17 +74,18 @@ func TestRemoveCallbackDigitalInterrupt(t *testing.T) {
 	}
 	i, err := CreateDigitalInterrupt(config)
 	test.That(t, err, test.ShouldBeNil)
+	di := i.(*BasicDigitalInterrupt)
 	intVal, err := i.Value(context.Background(), nil)
 	test.That(t, err, test.ShouldBeNil)
 	test.That(t, intVal, test.ShouldEqual, int64(0))
-	test.That(t, Tick(context.Background(), i.(*BasicDigitalInterrupt), true, nowNanosecondsTest()), test.ShouldBeNil)
+	test.That(t, Tick(context.Background(), di, true, nowNanosecondsTest()), test.ShouldBeNil)
 	intVal, err = i.Value(context.Background(), nil)
 	test.That(t, err, test.ShouldBeNil)
 	test.That(t, intVal, test.ShouldEqual, int64(1))
 
 	c1 := make(chan board.Tick)
 	test.That(t, c1, test.ShouldNotBeNil)
-	AddCallback(i.(*BasicDigitalInterrupt), c1)
+	AddCallback(di, c1)
 	var wg sync.WaitGroup
 	wg.Add(1)
 	ret := false
@@ -101,18 +104,17 @@ func TestRemoveCallbackDigitalInterrupt(t *testing.T) {
 			ret = tick.High
 		}
 	}()
-	test.That(t, Tick(context.Background(), i.(*BasicDigitalInterrupt), true, nowNanosecondsTest()), test.ShouldBeNil)
+	test.That(t, Tick(context.Background(), di, true, nowNanosecondsTest()), test.ShouldBeNil)
 	intVal, err = i.Value(context.Background(), nil)
 	test.That(t, err, test.ShouldBeNil)
 	test.That(t, intVal, test.ShouldEqual, int64(2))
 	wg.Wait()
 	c2 := make(chan board.Tick)
 	test.That(t, c2, test.ShouldNotBeNil)
-	AddCallback(i.(*BasicDigitalInterrupt), c2)
+	AddCallback(di, c2)
 	test.That(t, ret, test.ShouldBeTrue)
 
-	i.RemoveCallback(c1)
-	i.RemoveCallback(c1)
+	RemoveCallback(di, c1)
 
 	ret2 := false
 	result := make(chan bool, 1)
@@ -132,7 +134,7 @@ func TestRemoveCallbackDigitalInterrupt(t *testing.T) {
 	}()
 	wg.Add(1)
 	go func() {
-		err := Tick(context.Background(), i.(*BasicDigitalInterrupt), true, nowNanosecondsTest())
+		err := Tick(context.Background(), di, true, nowNanosecondsTest())
 		if err != nil {
 			result <- true
 		}

components/board/server.go

@@ -215,12 +215,6 @@ func (s *serviceServer) StreamTicks(
 		return err
 	}
 
-	defer func() {
-		for _, i := range interrupts {
-			i.RemoveCallback(ticksChan)
-		}
-	}()
-
 	// Send an empty response first so the client doesn't block while checking for errors.
 	err = server.Send(&pb.StreamTicksResponse{})
 	if err != nil {

components/board/server_test.go

@@ -825,21 +825,6 @@ func TestStreamTicks(t *testing.T) {
 				}
 				return nil, nil
 			}
-			if tc.injectDigitalInterrupts != nil {
-				for _, i := range tc.injectDigitalInterrupts {
-					i.RemoveCallbackFunc = func(c chan board.Tick) {
-						for id := range callbacks {
-							if callbacks[id] == c {
-								// To remove this item, we replace it with the last item in the list, then truncate the
-								// list by 1.
-								callbacks[id] = callbacks[len(callbacks)-1]
-								callbacks = callbacks[:len(callbacks)-1]
-								break
-							}
-						}
-					}
-				}
-			}
 
 			cancelCtx, cancel := context.WithCancel(context.Background())
 			defer cancel()

components/encoder/incremental/incremental_encoder.go

@@ -226,9 +226,6 @@ func (e *Encoder) Start(ctx context.Context, b board.Board) {
 	e.activeBackgroundWorkers.Add(1)
 
 	utils.ManagedGo(func() {
-		// Remove the callbacks added by the interrupt stream.
-		defer e.A.RemoveCallback(ch)
-		defer e.B.RemoveCallback(ch)
 		for {
 			// This looks redundant with the other select statement below, but it's not: if we're
 			// supposed to return, we need to do that even if chanA and chanB are full of data, and

components/encoder/incremental/incremental_test.go

@@ -195,12 +195,6 @@ func MakeBoard(t *testing.T) board.Board {
 	i2.ValueFunc = func(ctx context.Context, extra map[string]interface{}) (int64, error) {
 		return 0, nil
 	}
-	i1.RemoveCallbackFunc = func(c chan board.Tick) {
-		delete(callbacks, i1)
-	}
-	i2.RemoveCallbackFunc = func(c chan board.Tick) {
-		delete(callbacks, i2)
-	}
 
 	b.DigitalInterruptByNameFunc = func(name string) (board.DigitalInterrupt, error) {
 		if name == "11" {

components/encoder/single/single_encoder.go

@@ -191,8 +191,6 @@ func (e *Encoder) Start(ctx context.Context, b board.Board) {
 	e.activeBackgroundWorkers.Add(1)
 
 	utils.ManagedGo(func() {
-		defer e.I.RemoveCallback(encoderChannel)
-
 		for {
 			select {
 			case <-e.cancelCtx.Done():

components/encoder/single/single_encoder_test.go

@@ -253,7 +253,6 @@ func MakeBoard(t *testing.T, name, pinname string) board.Board {
 		ch <- board.Tick{Name: i.Name(), High: high, TimestampNanosec: nanoseconds}
 		return nil
 	}
-	i.RemoveCallbackFunc = func(c chan board.Tick) {}
 
 	b.DigitalInterruptByNameFunc = func(name string) (board.DigitalInterrupt, error) {
 		return i, nil