fix: local bucket notification filtering (#715)

Co-authored-by: Tim Holm <[email protected]>
nitrictech · Apr 12, 2024 · 8b6a950 · 8b6a950
1 parent c42df82
commit 8b6a950
Show file tree

Hide file tree

Showing 3 changed files with 23 additions and 6 deletions.
diff --git a/cmd/start.go b/cmd/start.go
@@ -34,6 +34,7 @@ import (
 	"github.com/nitrictech/cli/pkg/view/tui/commands/services"
 	"github.com/nitrictech/cli/pkg/view/tui/fragments"
 	"github.com/nitrictech/cli/pkg/view/tui/teax"
+	"github.com/nitrictech/nitric/core/pkg/logger"
 )
 
 var startNoBrowser bool
@@ -89,7 +90,8 @@ var startCmd = &cobra.Command{
 		go func() {
 			err := proj.RunServicesWithCommand(localCloud, stopChan, updatesChan)
 			if err != nil {
-				panic(err)
+				// typically these are just exit statuses
+				logger.Debugf("Services exited with: %s", err.Error())
 			}
 		}()
 

diff --git a/pkg/cloud/storage/storage.go b/pkg/cloud/storage/storage.go
@@ -24,6 +24,7 @@ import (
 	"net/http"
 	"os"
 	"path/filepath"
+	"strings"
 	"sync"
 
 	"github.com/asaskevich/EventBus"
@@ -136,6 +137,11 @@ func (r *LocalStorageService) Listen(stream storagepb.StorageListener_ListenServ
 		return fmt.Errorf("expected registration request on first request")
 	}
 
+	// Added for compatibility, will be deprecated and removed in a future release
+	if firstRequest.GetRegistrationRequest().KeyPrefixFilter == "*" {
+		firstRequest.GetRegistrationRequest().KeyPrefixFilter = ""
+	}
+
 	err = stream.Send(&storagepb.ServerMessage{
 		Id: firstRequest.Id,
 		Content: &storagepb.ServerMessage_RegistrationResponse{
@@ -154,16 +160,24 @@ func (r *LocalStorageService) Listen(stream storagepb.StorageListener_ListenServ
 	r.registerListener(serviceName, firstRequest.GetRegistrationRequest())
 	defer r.unregisterListener(serviceName, firstRequest.GetRegistrationRequest())
 
-	err = eventbus.StorageBus().SubscribeAsync(listenTopicName, func(req *storagepb.ServerMessage) {
-		err := stream.Send(req)
-		if err != nil {
-			fmt.Println("problem sending the event")
+	fun := func(req *storagepb.ServerMessage) {
+		if strings.HasPrefix(req.GetBlobEventRequest().GetBlobEvent().Key, firstRequest.GetRegistrationRequest().KeyPrefixFilter) {
+			err := stream.Send(req)
+			if err != nil {
+				fmt.Println("problem sending the event")
+			}
 		}
-	}, false)
+	}
+
+	err = eventbus.StorageBus().SubscribeAsync(listenTopicName, fun, false)
 	if err != nil {
 		return fmt.Errorf("error subscribing to topic: %s", err.Error())
 	}
 
+	defer func() {
+		_ = eventbus.StorageBus().Unsubscribe(listenTopicName, fun)
+	}()
+
 	// block here...
 	for {
 		_, err := stream.Recv()

diff --git a/pkg/project/project.go b/pkg/project/project.go
@@ -241,6 +241,7 @@ func (p *Project) RunServicesWithCommand(localCloud *cloud.LocalCloud, stop <-ch
 			}
 
 			return svc.Run(stopChannels[idx], updates, map[string]string{
+				"PYTHONUNBUFFERED":   "TRUE", // ensure all print statements print immediately for python
 				"NITRIC_ENVIRONMENT": "run",
 				"SERVICE_ADDRESS":    "localhost:" + strconv.Itoa(port),
 				// TODO: add .env variables.