go/runtime/registry/notifier: Refactor ROFL notifier

Author: peternose

.changelog/6263.trivial.md

@@ -40,6 +40,9 @@ type RuntimeHostHandlerEnvironment interface {
 
 	// GetRuntimeRegistry returns the runtime registry.
 	GetRuntimeRegistry() Registry
+
+	// GetROFLNotifier returns the ROFL notifier.
+	GetROFLNotifier() (*ROFLNotifier, error)
 }
 
 // RuntimeHostHandler is a runtime host handler suitable for compute runtimes. It provides the
@@ -124,7 +127,11 @@ func (h *runtimeHostHandler) Handle(ctx context.Context, rq *protocol.Body) (*pr
 func (h *runtimeHostHandler) NewSubHandler(comp *bundle.ExplodedComponent) (host.RuntimeHandler, error) {
 	switch comp.Kind {
 	case component.ROFL:
-		return newSubHandlerROFL(comp, h)
+		roflNotifier, err := h.env.GetROFLNotifier()
+		if err != nil {
+			return nil, err
+		}
+		return newSubHandlerROFL(comp, h, roflNotifier)
 	default:
 		return nil, fmt.Errorf("cannot create sub-handler for component '%s'", comp.Kind)
 	}

go/runtime/registry/handler.go

@@ -36,13 +36,13 @@ type roflHostHandler struct {
 	comp  *bundle.ExplodedComponent
 	comps map[component.ID]host.Runtime
 
-	client        runtimeClient.RuntimeClient
-	eventNotifier *roflEventNotifier
+	client       runtimeClient.RuntimeClient
+	roflNotifier *ROFLNotifier
 
 	logger *logging.Logger
 }
 
-func newSubHandlerROFL(comp *bundle.ExplodedComponent, parent *runtimeHostHandler) (*roflHostHandler, error) {
+func newSubHandlerROFL(comp *bundle.ExplodedComponent, parent *runtimeHostHandler, roflNotifier *ROFLNotifier) (*roflHostHandler, error) {
 	client, err := parent.env.GetRuntimeRegistry().Client()
 	if err != nil {
 		return nil, err
@@ -53,13 +53,13 @@ func newSubHandlerROFL(comp *bundle.ExplodedComponent, parent *runtimeHostHandle
 		With("component_id", comp.ID())
 
 	return &roflHostHandler{
-		parent:        parent,
-		id:            comp.ID(),
-		comp:          comp,
-		comps:         make(map[component.ID]host.Runtime),
-		client:        client,
-		eventNotifier: newROFLEventNotifier(parent.runtime, client, logger),
-		logger:        logger,
+		parent:       parent,
+		id:           comp.ID(),
+		comp:         comp,
+		comps:        make(map[component.ID]host.Runtime),
+		client:       client,
+		roflNotifier: roflNotifier,
+		logger:       logger,
 	}, nil
 }
 
@@ -74,7 +74,7 @@ func (rh *roflHostHandler) AttachRuntime(id component.ID, rt host.Runtime) error
 	if id != rh.id {
 		return nil
 	}
-	return rh.eventNotifier.AttachRuntime(rt)
+	return nil
 }
 
 // getLocalStorageKey returns a properly namespaced version of the local storage key.
@@ -254,13 +254,15 @@ func (rh *roflHostHandler) handleHostSubmitTx(
 }
 
 func (rh *roflHostHandler) handleHostRegisterNotify(
-	ctx context.Context,
+	_ context.Context,
 	rq *protocol.HostRegisterNotifyRequest,
 ) (*protocol.Empty, error) {
 	// Subscribe to event notifications.
-	if err := rh.eventNotifier.RegisterNotify(ctx, rq); err != nil {
-		return nil, err
+	nfs := &rofl.Notifications{
+		Blocks: rq.RuntimeBlock,
+		Events: rq.RuntimeEvent.Tags,
 	}
+	rh.roflNotifier.register(rh.id, nfs)
 
 	return &protocol.Empty{}, nil
 }

go/runtime/registry/handler_rofl.go

@@ -30,6 +30,9 @@ const (
 	queueKeyManagerStatus      = "key-manager/status"
 	queueKeyManagerQuotePolicy = "key-manager/quote-policy"
 	queueConsensusSync         = "consensus-sync"
+	queueROFLBlock             = "rofl/block"
+	queueROFLTags              = "rofl/tags"
+	queueROFLConfig            = "rofl/config"
 )
 
 const (