Implement atomicTrigger AckEvent

Author: DylanTinianov

pkg/capabilities/registry/base.go

@@ -244,6 +244,14 @@ func (a *atomicTriggerCapability) GetState() connectivity.State {
 	return connectivity.State(-1) // unknown
 }
 
+func (a *atomicTriggerCapability) AckEvent(ctx context.Context, eventId string) error {
+	c := a.Load()
+	if c == nil {
+		return errors.New("capability unavailable")
+	}
+	return (*c).AckEvent(ctx, eventId)
+}
+
 func (a *atomicTriggerCapability) RegisterTrigger(ctx context.Context, request capabilities.TriggerRegistrationRequest) (<-chan capabilities.TriggerResponse, error) {
 	c := a.Load()
 	if c == nil {
@@ -360,6 +368,14 @@ func (a *atomicExecuteAndTriggerCapability) GetState() connectivity.State {
 	return connectivity.State(-1) // unknown
 }
 
+func (a *atomicExecuteAndTriggerCapability) AckEvent(ctx context.Context, eventId string) error {
+	c := a.Load()
+	if c == nil {
+		return errors.New("capability unavailable")
+	}
+	return (*c).AckEvent(ctx, eventId)
+}
+
 func (a *atomicExecuteAndTriggerCapability) RegisterTrigger(ctx context.Context, request capabilities.TriggerRegistrationRequest) (<-chan capabilities.TriggerResponse, error) {
 	c := a.Load()
 	if c == nil {

pkg/capabilities/triggers/mercury_trigger.go

@@ -37,11 +37,6 @@ type MercuryTriggerService struct {
 	metaOverride       datastreams.Metadata // usually empty, but set to a value in mock trigger
 }
 
-func (o *MercuryTriggerService) AckEvent(ctx context.Context, eventId string) error {
-	//TODO Use BaseTriggerCapability?
-	return nil
-}
-
 var _ capabilities.TriggerCapability = (*MercuryTriggerService)(nil)
 var _ services.Service = &MercuryTriggerService{}
 
@@ -96,6 +91,11 @@ func (o *MercuryTriggerService) ProcessReport(reports []datastreams.FeedReport)
 	return nil
 }
 
+func (o *MercuryTriggerService) AckEvent(ctx context.Context, eventId string) error {
+	// TODO Implement?
+	return nil
+}
+
 func (o *MercuryTriggerService) RegisterTrigger(ctx context.Context, req capabilities.TriggerRegistrationRequest) (<-chan capabilities.TriggerResponse, error) {
 	wid := req.Metadata.WorkflowID
 

pkg/capabilities/triggers/on_demand_trigger.go

@@ -73,7 +73,7 @@ func (o *OnDemand) SendEvent(ctx context.Context, wid string, event capabilities
 }
 
 func (o *OnDemand) AckEvent(ctx context.Context, eventId string) error {
-	//TODO Should this trigger type use BaseTriggerCapability?
+	//TODO Implement?
 	return nil
 }