add subscription data source

SubscriptionDataSource allows querying of objects directly from the subscription results array. It is beneficial to use this data source when it is useful to query by name, possibly allowing earlier data sources to pick up the query first (e.g. local mocking).

I've also updated the local fixed data source to be a prepend action instead of append, to ensure that any local mocks are always handled first. With this, the order of options registered should not impact query behaviour.

policy/data/subscription.go

@@ -0,0 +1,40 @@
+package data
+
+import (
+	"context"
+	"fmt"
+	"olympos.io/encoding/edn"
+)
+
+// SubscriptionDataSource allows querying of objects directly from the subscription results array.
+// It is beneficial to use this data source when it is useful to query by name,
+// possibly allowing earlier data sources to pick up the query first.
+type SubscriptionDataSource struct {
+	queryIndexes        map[string]int
+	subscriptionResults [][]edn.RawMessage
+}
+
+func NewSubscriptionDataSource(queryIndexes map[string]int, subscriptionResults [][]edn.RawMessage) SubscriptionDataSource {
+	return SubscriptionDataSource{
+		queryIndexes:        queryIndexes,
+		subscriptionResults: subscriptionResults,
+	}
+}
+
+func (ds SubscriptionDataSource) Query(_ context.Context, queryName string, _ string, _ map[string]interface{}, output interface{}) (*QueryResponse, error) {
+	ix, ok := ds.queryIndexes[queryName]
+	if !ok {
+		return nil, nil
+	}
+
+	if ix >= len(ds.subscriptionResults) {
+		return nil, fmt.Errorf("can't get subscription query %s (index %d) from result length %d", queryName, ix, len(ds.subscriptionResults))
+	}
+
+	err := edn.Unmarshal(ds.subscriptionResults[0][ix], output)
+	if err != nil {
+		return nil, err
+	}
+
+	return &QueryResponse{}, nil
+}

policy/policy_handler/local.go

@@ -22,12 +22,12 @@ func WithLocal() Opt {
 	return func(h *EventHandler) {
 		h.subscriptionNames = append(h.subscriptionNames, eventNameLocalEval)
 		h.subscriptionDataProviders = append(h.subscriptionDataProviders, getLocalSubscriptionData)
-		h.dataSourceProviders = append(h.dataSourceProviders, buildLocalDataSources)
+		h.dataSourceProviders = append([]dataSourceProvider{buildLocalDataSources}, h.dataSourceProviders...)
 		h.transactFilters = append(h.transactFilters, shouldTransactLocal)
 	}
 }
 
-func getLocalSubscriptionData(ctx context.Context, req skill.RequestContext) (*goals.EvaluationMetadata, skill.Configuration, error) {
+func getLocalSubscriptionData(_ context.Context, req skill.RequestContext) (*goals.EvaluationMetadata, skill.Configuration, error) {
 	if req.Event.Context.SyncRequest.Name != eventNameLocalEval {
 		return nil, skill.Configuration{}, nil
 	}
@@ -45,7 +45,7 @@ func getLocalSubscriptionData(ctx context.Context, req skill.RequestContext) (*g
 	}, req.Event.Context.SyncRequest.Configuration, nil
 }
 
-func buildLocalDataSources(ctx context.Context, req skill.RequestContext, evalMeta goals.EvaluationMetadata) ([]data.DataSource, error) {
+func buildLocalDataSources(ctx context.Context, req skill.RequestContext, _ goals.EvaluationMetadata) ([]data.DataSource, error) {
 	if req.Event.Context.SyncRequest.Name != eventNameLocalEval {
 		return []data.DataSource{}, nil
 	}
@@ -80,6 +80,6 @@ func buildLocalDataSources(ctx context.Context, req skill.RequestContext, evalMe
 	}, nil
 }
 
-func shouldTransactLocal(ctx context.Context, req skill.RequestContext) bool {
+func shouldTransactLocal(_ context.Context, req skill.RequestContext) bool {
 	return req.Event.Context.SyncRequest.Name != eventNameLocalEval
 }

policy/policy_handler/subscription.go

@@ -2,6 +2,7 @@ package policy_handler
 
 import (
 	"context"
+	"github.com/atomist-skills/go-skill/policy/data"
 	"github.com/atomist-skills/go-skill/policy/goals"
 
 	"github.com/atomist-skills/go-skill"
@@ -13,7 +14,7 @@ func withSubscription() Opt {
 	}
 }
 
-func getSubscriptionData(ctx context.Context, req skill.RequestContext) (*goals.EvaluationMetadata, skill.Configuration, error) {
+func getSubscriptionData(_ context.Context, req skill.RequestContext) (*goals.EvaluationMetadata, skill.Configuration, error) {
 	if req.Event.Context.Subscription.Name == "" {
 		return nil, skill.Configuration{}, nil
 	}
@@ -24,3 +25,17 @@ func getSubscriptionData(ctx context.Context, req skill.RequestContext) (*goals.
 	}
 	return evalMeta, req.Event.Context.Subscription.Configuration, nil
 }
+
+func WithSubscriptionDataSource(queryIndexes map[string]int) Opt {
+	return func(h *EventHandler) {
+		h.dataSourceProviders = append(h.dataSourceProviders, buildSubscriptionDataSource(queryIndexes))
+	}
+}
+
+func buildSubscriptionDataSource(queryIndexes map[string]int) dataSourceProvider {
+	return func(ctx context.Context, req skill.RequestContext, evalMeta goals.EvaluationMetadata) ([]data.DataSource, error) {
+		return []data.DataSource{
+			data.NewSubscriptionDataSource(queryIndexes, evalMeta.SubscriptionResult),
+		}, nil
+	}
+}