pkg/loop/internal/core/services/capability: use NewClientConn in place of Dial for reconnect

Author: jmank88

pkg/loop/internal/core/services/capability/capabilities.go

@@ -22,7 +22,7 @@ type TriggerCapabilityClient struct {
 	*baseCapabilityClient
 }
 
-func NewTriggerCapabilityClient(brokerExt *net.BrokerExt, conn *grpc.ClientConn) capabilities.TriggerCapability {
+func NewTriggerCapabilityClient(brokerExt *net.BrokerExt, conn grpc.ClientConnInterface) capabilities.TriggerCapability {
 	return &TriggerCapabilityClient{
 		triggerExecutableClient: newTriggerExecutableClient(brokerExt, conn),
 		baseCapabilityClient:    newBaseCapabilityClient(brokerExt, conn),
@@ -39,7 +39,7 @@ type ExecutableCapability interface {
 	capabilities.BaseCapability
 }
 
-func NewExecutableCapabilityClient(brokerExt *net.BrokerExt, conn *grpc.ClientConn) ExecutableCapability {
+func NewExecutableCapabilityClient(brokerExt *net.BrokerExt, conn grpc.ClientConnInterface) ExecutableCapability {
 	return &ExecutableCapabilityClient{
 		executableClient:     newExecutableClient(brokerExt, conn),
 		baseCapabilityClient: newBaseCapabilityClient(brokerExt, conn),
@@ -52,7 +52,7 @@ type CombinedCapabilityClient struct {
 	*triggerExecutableClient
 }
 
-func NewCombinedCapabilityClient(brokerExt *net.BrokerExt, conn *grpc.ClientConn) ExecutableCapability {
+func NewCombinedCapabilityClient(brokerExt *net.BrokerExt, conn grpc.ClientConnInterface) ExecutableCapability {
 	return &CombinedCapabilityClient{
 		executableClient:        newExecutableClient(brokerExt, conn),
 		baseCapabilityClient:    newBaseCapabilityClient(brokerExt, conn),
@@ -141,7 +141,7 @@ type baseCapabilityClient struct {
 
 var _ capabilities.BaseCapability = (*baseCapabilityClient)(nil)
 
-func newBaseCapabilityClient(brokerExt *net.BrokerExt, conn *grpc.ClientConn) *baseCapabilityClient {
+func newBaseCapabilityClient(brokerExt *net.BrokerExt, conn grpc.ClientConnInterface) *baseCapabilityClient {
 	return &baseCapabilityClient{grpc: capabilitiespb.NewBaseCapabilityClient(conn), BrokerExt: brokerExt}
 }
 
@@ -327,6 +327,7 @@ func (t *triggerExecutableClient) registerTrigger(ctx context.Context, req capab
 	if err != nil {
 		return nil, cancel, fmt.Errorf("failed to start forwarding messages from stream: %w", err)
 	}
+	//TODO remember these to re-run them on init
 
 	return ch, cancel, nil
 }
@@ -348,7 +349,7 @@ func (t *triggerExecutableClient) UnregisterTrigger(ctx context.Context, req cap
 	return nil
 }
 
-func newTriggerExecutableClient(brokerExt *net.BrokerExt, conn *grpc.ClientConn) *triggerExecutableClient {
+func newTriggerExecutableClient(brokerExt *net.BrokerExt, conn grpc.ClientConnInterface) *triggerExecutableClient {
 	return &triggerExecutableClient{
 		grpc:        capabilitiespb.NewTriggerExecutableClient(conn),
 		BrokerExt:   brokerExt,
@@ -439,7 +440,7 @@ type executableClient struct {
 	*net.BrokerExt
 }
 
-func newExecutableClient(brokerExt *net.BrokerExt, conn *grpc.ClientConn) *executableClient {
+func newExecutableClient(brokerExt *net.BrokerExt, conn grpc.ClientConnInterface) *executableClient {
 	return &executableClient{
 		grpc:      capabilitiespb.NewExecutableClient(conn),
 		BrokerExt: brokerExt,

pkg/loop/internal/core/services/capability/capabilities_registry.go

@@ -208,15 +208,13 @@ func (cr *capabilitiesRegistryClient) Get(ctx context.Context, ID string) (capab
 		Id: ID,
 	}
 
-	res, err := cr.grpc.Get(ctx, req)
-	if err != nil {
-		return nil, err
-	}
-
-	conn, err := cr.Dial(res.CapabilityID)
-	if err != nil {
-		return nil, net.ErrConnDial{Name: "Capability", ID: res.CapabilityID, Err: err}
-	}
+	conn := cr.NewClientConn("Capability", func(ctx context.Context) (id uint32, deps net.Resources, err error) {
+		res, err := cr.grpc.Get(ctx, req)
+		if err != nil {
+			return 0, nil, err
+		}
+		return res.CapabilityID, nil, nil
+	})
 	client := newBaseCapabilityClient(cr.BrokerExt, conn)
 	return client, nil
 }
@@ -226,15 +224,13 @@ func (cr *capabilitiesRegistryClient) GetTrigger(ctx context.Context, ID string)
 		Id: ID,
 	}
 
-	res, err := cr.grpc.GetTrigger(ctx, req)
-	if err != nil {
-		return nil, err
-	}
-
-	conn, err := cr.Dial(res.CapabilityID)
-	if err != nil {
-		return nil, net.ErrConnDial{Name: "GetTrigger", ID: res.CapabilityID, Err: err}
-	}
+	conn := cr.NewClientConn("Trigger", func(ctx context.Context) (id uint32, deps net.Resources, err error) {
+		res, err := cr.grpc.GetTrigger(ctx, req)
+		if err != nil {
+			return 0, nil, err
+		}
+		return res.CapabilityID, nil, nil
+	})
 	client := NewTriggerCapabilityClient(cr.BrokerExt, conn)
 	return client, nil
 }
@@ -244,14 +240,13 @@ func (cr *capabilitiesRegistryClient) GetExecutable(ctx context.Context, ID stri
 		Id: ID,
 	}
 
-	res, err := cr.grpc.GetExecutable(ctx, req)
-	if err != nil {
-		return nil, err
-	}
-	conn, err := cr.Dial(res.CapabilityID)
-	if err != nil {
-		return nil, net.ErrConnDial{Name: "GetExecutable", ID: res.CapabilityID, Err: err}
-	}
+	conn := cr.NewClientConn("Executable", func(ctx context.Context) (id uint32, deps net.Resources, err error) {
+		res, err := cr.grpc.GetExecutable(ctx, req)
+		if err != nil {
+			return 0, nil, err
+		}
+		return res.CapabilityID, nil, nil
+	})
 	client := NewExecutableCapabilityClient(cr.BrokerExt, conn)
 	return client, nil
 }