Add support for proper usage of Context (#239)

* Update interfaces & impls to pass context

Author: CDRayn

.golangci.yml

@@ -0,0 +1,45 @@
+version: "2"
+run:
+  modules-download-mode: readonly
+linters:
+  default: none
+  enable:
+    - errcheck
+    - goheader
+    - govet
+    - ineffassign
+    - staticcheck
+    - unused
+  settings:
+    goheader:
+      values:
+        regexp:
+          decade: '[0-9]'
+          year: '[0-9]'
+      template: Copyright (C) 20{{ decade }}{{ year }} by Posit Software, PBC
+  exclusions:
+    generated: lax
+    presets:
+      - comments
+      - common-false-positives
+      - legacy
+      - std-error-handling
+    paths:
+      - mock_.*\.go$
+    rules:
+      - linters:
+          - staticcheck
+        text: "QF1008|ST1001|ST1005|ST1019|SA1019|QF1003|ST1012|SA1029"
+formatters:
+  enable:
+    - gofmt
+    - goimports
+  settings:
+    goimports:
+      local-prefixes:
+        - github.com/rstudio
+        - rspm
+  exclusions:
+    generated: lax
+    paths:
+      - mock_.*\.go$

examples/cmd/markdownRenderer/main.go

@@ -73,8 +73,8 @@ type cacheQueueWrapper struct {
 // AddressedPush is part of the cache's queue interface. Since we're using the
 // rsqueue/Queue package for message queuing, we can simply wrap the queue's
 // AddressedPush method and hand off the wrapped queue to the cache.
-func (q *cacheQueueWrapper) AddressedPush(priority uint64, groupId int64, address string, work rscache.QueueWork) error {
-	return q.Queue.AddressedPush(priority, groupId, address, work)
+func (q *cacheQueueWrapper) AddressedPush(ctx context.Context, priority uint64, groupId int64, address string, work rscache.QueueWork) error {
+	return q.Queue.AddressedPush(ctx, priority, groupId, address, work)
 }
 
 func init() {
@@ -90,6 +90,8 @@ func (l *leveler) Level() slog.Level {
 }
 
 func main() {
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt)
+	defer stop()
 
 	logger := slog.New(slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{
 		Level: &leveler{level: slog.LevelDebug},
@@ -286,12 +288,12 @@ func main() {
 		JobLifecycleWrapper:    &metrics.EmptyJobLifecycleWrapper{},
 	}
 	ag := agent.NewAgent(agentCfg)
-	go ag.Run(func(n listener.Notification) {
+	go ag.Run(ctx, func(n listener.Notification) {
 		// Since we don't enforce how notifications are sent, the agent sends a
 		// notification to this callback method when work is completed. Here, we
 		// simply pass those messages on to the store, which knows how to send
 		// notifications.
-		err = exampleStore.Notify(notifytypes.ChannelMessages, n)
+		err = exampleStore.Notify(ctx, notifytypes.ChannelMessages, n)
 		if err != nil {
 			log.Printf("Error notifying of queue work complete: %s", err)
 		}
@@ -300,7 +302,7 @@ func main() {
 	// Start HTTP services and listen until the application exits.
 	router := mux.NewRouter()
 	handler := handlers.NewHttpHandler(address, router, cache)
-	ctx, cancel := context.WithCancel(context.Background())
+	ctx, cancel := context.WithCancel(ctx)
 	go handler.Start(ctx)
 	// Cancel the handler's context when the application exits for graceful
 	// shutdown.

examples/cmd/markdownRenderer/storage/chunks.go

@@ -3,6 +3,7 @@ package storage
 // Copyright (C) 2022 by RStudio, PBC
 
 import (
+	"context"
 	"time"
 
 	"github.com/rstudio/platform-lib/v2/examples/cmd/markdownRenderer/notifytypes"
@@ -15,7 +16,7 @@ type ExampleChunkWaiter struct {
 	awb broadcaster.Broadcaster
 }
 
-func (cw *ExampleChunkWaiter) WaitForChunk(c *types.ChunkNotification) {
+func (cw *ExampleChunkWaiter) WaitForChunk(ctx context.Context, c *types.ChunkNotification) {
 	timeout := time.NewTimer(c.Timeout)
 	defer timeout.Stop()
 
@@ -46,15 +47,15 @@ func NewExampleChunkWaiter(awb broadcaster.Broadcaster) *ExampleChunkWaiter {
 }
 
 type ChunkNotifierStore interface {
-	Notify(channelName string, n interface{}) error
+	Notify(ctx context.Context, channelName string, n interface{}) error
 }
 
 type ExampleChunkNotifier struct {
 	store ChunkNotifierStore
 }
 
-func (cn *ExampleChunkNotifier) Notify(c *types.ChunkNotification) error {
-	return cn.store.Notify(notifytypes.ChannelMessages, notifytypes.NewChunkNotification(c.Address, c.Chunk))
+func (cn *ExampleChunkNotifier) Notify(ctx context.Context, c *types.ChunkNotification) error {
+	return cn.store.Notify(ctx, notifytypes.ChannelMessages, notifytypes.NewChunkNotification(c.Address, c.Chunk))
 }
 
 func NewExampleChunkNotifier(cstore ChunkNotifierStore) *ExampleChunkNotifier {

examples/cmd/markdownRenderer/storage/chunks_test.go

@@ -3,6 +3,7 @@ package storage
 // Copyright (C) 2022 by RStudio, PBC
 
 import (
+	"context"
 	"errors"
 	"testing"
 	"time"
@@ -39,7 +40,7 @@ type dummyStore struct {
 	notify error
 }
 
-func (d *dummyStore) Notify(channel string, n interface{}) error {
+func (d *dummyStore) Notify(ctx context.Context, channel string, n interface{}) error {
 	return d.notify
 }
 
@@ -49,6 +50,7 @@ type ChunksSuite struct {
 var _ = check.Suite(&ChunksSuite{})
 
 func (s *ChunksSuite) TestWait(c *check.C) {
+	ctx := context.Background()
 	awb := &fakeBroadcaster{
 		l: make(chan listener.Notification),
 	}
@@ -63,28 +65,29 @@ func (s *ChunksSuite) TestWait(c *check.C) {
 	go func() {
 		awb.l <- notifytypes.NewChunkNotification("", 4)
 	}()
-	w.WaitForChunk(cn)
+	w.WaitForChunk(ctx, cn)
 
 	// Test waiting for chunk (timeout)
 	cn = &storagetypes.ChunkNotification{
 		Timeout: time.Millisecond,
 		Chunk:   3,
 	}
-	w.WaitForChunk(cn)
+	w.WaitForChunk(ctx, cn)
 }
 
 func (s *ChunksSuite) TestNotify(c *check.C) {
+	ctx := context.Background()
 	cstore := &dummyStore{
 		notify: errors.New("some error"),
 	}
 	cn := NewExampleChunkNotifier(cstore)
 	c.Assert(cn, check.DeepEquals, &ExampleChunkNotifier{store: cstore})
 
-	err := cn.Notify(&storagetypes.ChunkNotification{})
+	err := cn.Notify(ctx, &storagetypes.ChunkNotification{})
 	c.Assert(err, check.ErrorMatches, "some error")
 
 	cstore.notify = nil
-	err = cn.Notify(&storagetypes.ChunkNotification{})
+	err = cn.Notify(ctx, &storagetypes.ChunkNotification{})
 	c.Assert(err, check.IsNil)
 }