extension point for pub/sub support

cmds/ocm/app/app.go

@@ -24,6 +24,7 @@ import (
 	"github.com/open-component-model/ocm/cmds/ocm/commands/ocmcmds/componentarchive"
 	"github.com/open-component-model/ocm/cmds/ocm/commands/ocmcmds/components"
 	"github.com/open-component-model/ocm/cmds/ocm/commands/ocmcmds/plugins"
+	"github.com/open-component-model/ocm/cmds/ocm/commands/ocmcmds/pubsub"
 	"github.com/open-component-model/ocm/cmds/ocm/commands/ocmcmds/references"
 	"github.com/open-component-model/ocm/cmds/ocm/commands/ocmcmds/resources"
 	"github.com/open-component-model/ocm/cmds/ocm/commands/ocmcmds/routingslips"
@@ -44,6 +45,7 @@ import (
 	"github.com/open-component-model/ocm/cmds/ocm/commands/verbs/hash"
 	"github.com/open-component-model/ocm/cmds/ocm/commands/verbs/install"
 	"github.com/open-component-model/ocm/cmds/ocm/commands/verbs/list"
+	"github.com/open-component-model/ocm/cmds/ocm/commands/verbs/set"
 	"github.com/open-component-model/ocm/cmds/ocm/commands/verbs/show"
 	"github.com/open-component-model/ocm/cmds/ocm/commands/verbs/sign"
 	"github.com/open-component-model/ocm/cmds/ocm/commands/verbs/transfer"
@@ -57,6 +59,7 @@ import (
 	topicocmaccessmethods "github.com/open-component-model/ocm/cmds/ocm/topics/ocm/accessmethods"
 	topicocmdownloaders "github.com/open-component-model/ocm/cmds/ocm/topics/ocm/downloadhandlers"
 	topicocmlabels "github.com/open-component-model/ocm/cmds/ocm/topics/ocm/labels"
+	topicocmpubsub "github.com/open-component-model/ocm/cmds/ocm/topics/ocm/pubsub"
 	topicocmrefs "github.com/open-component-model/ocm/cmds/ocm/topics/ocm/refs"
 	topicocmuploaders "github.com/open-component-model/ocm/cmds/ocm/topics/ocm/uploadhandlers"
 	topicbootstrap "github.com/open-component-model/ocm/cmds/ocm/topics/toi/bootstrapping"
@@ -222,6 +225,7 @@ func newCliCommand(opts *CLIOptions, mod ...func(clictx.Context, *cobra.Command)
 
 	cmd.AddCommand(check.NewCommand(opts.Context))
 	cmd.AddCommand(get.NewCommand(opts.Context))
+	cmd.AddCommand(set.NewCommand(opts.Context))
 	cmd.AddCommand(list.NewCommand(opts.Context))
 	cmd.AddCommand(create.NewCommand(opts.Context))
 	cmd.AddCommand(add.NewCommand(opts.Context))
@@ -246,6 +250,7 @@ func newCliCommand(opts *CLIOptions, mod ...func(clictx.Context, *cobra.Command)
 	cmd.AddCommand(cmdutils.HideCommand(plugins.NewCommand(opts.Context)))
 	cmd.AddCommand(cmdutils.HideCommand(action.NewCommand(opts.Context)))
 	cmd.AddCommand(cmdutils.HideCommand(routingslips.NewCommand(opts.Context)))
+	cmd.AddCommand(cmdutils.HideCommand(pubsub.NewCommand(opts.Context)))
 
 	cmd.AddCommand(cmdutils.OverviewCommand(cachecmds.NewCommand(opts.Context)))
 	cmd.AddCommand(cmdutils.OverviewCommand(ocicmds.NewCommand(opts.Context)))
@@ -268,6 +273,7 @@ func newCliCommand(opts *CLIOptions, mod ...func(clictx.Context, *cobra.Command)
 	cmd.AddCommand(topicocmuploaders.New(ctx))
 	cmd.AddCommand(topicocmdownloaders.New(ctx))
 	cmd.AddCommand(topicocmlabels.New(ctx))
+	cmd.AddCommand(topicocmpubsub.New(ctx))
 	cmd.AddCommand(attributes.New(ctx))
 	cmd.AddCommand(topicbootstrap.New(ctx, "toi-bootstrapping"))
 
@@ -280,6 +286,7 @@ func newCliCommand(opts *CLIOptions, mod ...func(clictx.Context, *cobra.Command)
 	help.AddCommand(topicocmuploaders.New(ctx))
 	help.AddCommand(topicocmdownloaders.New(ctx))
 	help.AddCommand(topicocmlabels.New(ctx))
+	help.AddCommand(topicocmpubsub.New(ctx))
 	help.AddCommand(attributes.New(ctx))
 	help.AddCommand(topicbootstrap.New(ctx, "toi-bootstrapping"))
 

cmds/ocm/commands/ocmcmds/cmd.go

@@ -7,6 +7,7 @@ import (
 	"github.com/open-component-model/ocm/cmds/ocm/commands/ocmcmds/components"
 	"github.com/open-component-model/ocm/cmds/ocm/commands/ocmcmds/ctf"
 	"github.com/open-component-model/ocm/cmds/ocm/commands/ocmcmds/plugins"
+	"github.com/open-component-model/ocm/cmds/ocm/commands/ocmcmds/pubsub"
 	"github.com/open-component-model/ocm/cmds/ocm/commands/ocmcmds/references"
 	"github.com/open-component-model/ocm/cmds/ocm/commands/ocmcmds/resourceconfig"
 	"github.com/open-component-model/ocm/cmds/ocm/commands/ocmcmds/resources"
@@ -38,6 +39,7 @@ func NewCommand(ctx clictx.Context) *cobra.Command {
 	cmd.AddCommand(versions.NewCommand(ctx))
 	cmd.AddCommand(plugins.NewCommand(ctx))
 	cmd.AddCommand(routingslips.NewCommand(ctx))
+	cmd.AddCommand(pubsub.NewCommand(ctx))
 
 	cmd.AddCommand(topicocmrefs.New(ctx))
 	cmd.AddCommand(topicocmaccessmethods.New(ctx))

cmds/ocm/commands/ocmcmds/names/names.go

@@ -15,4 +15,5 @@ var (
 	Plugins                = []string{"plugins", "plugin", "p"}
 	Action                 = []string{"action"}
 	RoutingSlips           = []string{"routingslips", "routingslip", "rs"}
+	PubSub                 = []string{"pubsub", "ps"}
 )

cmds/ocm/commands/ocmcmds/pubsub/cmd.go

@@ -0,0 +1,23 @@
+package pubsub
+
+import (
+	"github.com/spf13/cobra"
+
+	"github.com/open-component-model/ocm/cmds/ocm/commands/ocmcmds/names"
+	"github.com/open-component-model/ocm/cmds/ocm/commands/ocmcmds/pubsub/get"
+	"github.com/open-component-model/ocm/cmds/ocm/commands/ocmcmds/pubsub/set"
+	"github.com/open-component-model/ocm/cmds/ocm/pkg/utils"
+	"github.com/open-component-model/ocm/pkg/contexts/clictx"
+)
+
+var Names = names.PubSub
+
+// NewCommand creates a new pubsub command.
+func NewCommand(ctx clictx.Context) *cobra.Command {
+	cmd := utils.MassageCommand(&cobra.Command{
+		Short: "Commands acting on sub/sub specifications",
+	}, Names...)
+	cmd.AddCommand(get.NewCommand(ctx, get.Verb))
+	cmd.AddCommand(set.NewCommand(ctx, set.Verb))
+	return cmd
+}

cmds/ocm/commands/ocmcmds/pubsub/get/cmd.go

@@ -0,0 +1,132 @@
+package get
+
+import (
+	"strings"
+
+	"github.com/mandelsoft/goutils/sliceutils"
+	"github.com/spf13/cobra"
+
+	"github.com/open-component-model/ocm/cmds/ocm/commands/ocmcmds/names"
+	"github.com/open-component-model/ocm/cmds/ocm/commands/verbs"
+	"github.com/open-component-model/ocm/cmds/ocm/pkg/output"
+	"github.com/open-component-model/ocm/cmds/ocm/pkg/processing"
+	"github.com/open-component-model/ocm/cmds/ocm/pkg/utils"
+	"github.com/open-component-model/ocm/pkg/contexts/clictx"
+	"github.com/open-component-model/ocm/pkg/contexts/ocm"
+	"github.com/open-component-model/ocm/pkg/contexts/ocm/cpi"
+	"github.com/open-component-model/ocm/pkg/contexts/ocm/pubsub"
+)
+
+var (
+	Names = names.PubSub
+	Verb  = verbs.Get
+)
+
+type Command struct {
+	utils.BaseCommand
+
+	RepoSpecs []string
+}
+
+var _ utils.OCMCommand = (*Command)(nil)
+
+// NewCommand creates a new pubsub command.
+func NewCommand(ctx clictx.Context, names ...string) *cobra.Command {
+	return utils.SetupCommand(&Command{BaseCommand: utils.NewBaseCommand(ctx, output.OutputOptions(outputs))}, utils.Names(Names, names...)...)
+}
+
+func (o *Command) ForName(name string) *cobra.Command {
+	return &cobra.Command{
+		Use:   "{<ocm repository>}",
+		Short: "Get the pubsub spec for an ocm repository",
+		Long: `
+A repository may be able to store a publish/subscribe specification
+to propagate the creation or update of component version.
+If such an implementation is available and a specification is
+assigned to the repository, it is shown. The specification
+can be set with the <CMD>ocm set pubsub</CMD>.
+`,
+	}
+}
+
+func (o *Command) Complete(args []string) error {
+	o.RepoSpecs = args
+	return nil
+}
+
+func (o *Command) Run() error {
+	return utils.HandleOutputsFor("repository spec", output.From(o), o.transform, o.RepoSpecs...)
+}
+
+func (o *Command) transform(in string) *Repo {
+	var spec cpi.RepositorySpec
+	rs := &Repo{RepoSpec: in}
+	u, err := ocm.ParseRepo(in)
+	if err == nil {
+		spec, err = o.OCMContext().MapUniformRepositorySpec(&u)
+	}
+	if err == nil {
+		rs.Repo, err = o.OCMContext().RepositoryForSpec(spec)
+	}
+	if err == nil {
+		rs.Spec, err = pubsub.SpecForRepo(rs.Repo)
+	}
+	if err != nil {
+		rs.Error = err.Error()
+	}
+	return rs
+}
+
+type Repo struct {
+	RepoSpec string            `json:"repository"`
+	Repo     cpi.Repository    `json:"-"`
+	Spec     pubsub.PubSubSpec `json:"pubsub,omitempty"`
+	Error    string            `json:"error,omitempty"`
+}
+
+var _ output.Manifest = (*Repo)(nil)
+
+func (r *Repo) AsManifest() interface{} {
+	return r
+}
+
+var outputs = output.NewOutputs(getRegular).AddManifestOutputs()
+
+func TableOutput(opts *output.Options, mapping processing.MappingFunction) *output.TableOutput {
+	return &output.TableOutput{
+		Headers: output.Fields("REPOSITORY", "PUBSUBTYPE", "ERROR"),
+		Options: opts,
+		Mapping: mapping,
+	}
+}
+
+func getRegular(opts *output.Options) output.Output {
+	return TableOutput(opts, mapGetRegularOutput).New()
+}
+
+func mapGetRegularOutput(e interface{}) interface{} {
+	r := e.(*Repo)
+	if r.Error != "" {
+		return output.Fields(r.RepoSpec, "", r.Error)
+	}
+	if r.Spec == nil {
+		return output.Fields(r.RepoSpec, "-", "")
+	}
+	list := sliceutils.Slice[string]{}
+	Add(r.Repo.GetContext(), r.Spec, &list)
+	strings.Join(list, ", ")
+
+	return output.Fields(r.RepoSpec, strings.Join(list, ", "), "")
+}
+
+func Add(ctx cpi.Context, s pubsub.PubSubSpec, slice *sliceutils.Slice[string]) {
+	if s == nil {
+		return
+	}
+	slice.Add(s.GetKind())
+	if u, ok := s.(pubsub.Unwrapable); ok {
+		for _, n := range u.Unwrap(ctx) {
+			Add(ctx, n, slice)
+		}
+	}
+}

cmds/ocm/commands/ocmcmds/pubsub/get/cmd_test.go

@@ -0,0 +1,106 @@
+package get_test
+
+import (
+	"bytes"
+	"fmt"
+
+	. "github.com/mandelsoft/goutils/testutils"
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+	. "github.com/open-component-model/ocm/cmds/ocm/testhelper"
+
+	"github.com/open-component-model/ocm/pkg/common/accessio"
+	"github.com/open-component-model/ocm/pkg/contexts/ocm"
+	"github.com/open-component-model/ocm/pkg/contexts/ocm/pubsub"
+	"github.com/open-component-model/ocm/pkg/contexts/ocm/pubsub/providers/ocireg"
+	"github.com/open-component-model/ocm/pkg/contexts/ocm/repositories/ctf"
+	"github.com/open-component-model/ocm/pkg/runtime"
+)
+
+const ARCH = "ctf"
+const ARCH2 = "ctf2"
+
+var _ = Describe("Test Environment", func() {
+	var env *TestEnv
+
+	BeforeEach(func() {
+		env = NewTestEnv()
+		env.OCMCommonTransport(ARCH, accessio.FormatDirectory)
+		env.OCMCommonTransport(ARCH2, accessio.FormatDirectory)
+		attr := pubsub.For(env)
+		attr.ProviderRegistry.Register(ctf.Type, &ocireg.Provider{})
+		attr.TypeScheme.Register(pubsub.NewPubSubType[*Spec](Type))
+		attr.TypeScheme.Register(pubsub.NewPubSubType[*Spec](TypeV1))
+
+		repo := Must(ctf.Open(env, ctf.ACC_WRITABLE, ARCH, 0o600, env))
+		defer repo.Close()
+		MustBeSuccessful(pubsub.SetForRepo(repo, NewSpec("testtarget")))
+	})
+
+	AfterEach(func() {
+		env.Cleanup()
+	})
+
+	It("get pubsub", func() {
+		var buf bytes.Buffer
+
+		MustBeSuccessful(env.CatchOutput(&buf).Execute("get", "pubsub", ARCH))
+		Expect(buf.String()).To(StringEqualTrimmedWithContext(`
+REPOSITORY PUBSUBTYPE ERROR
+ctf        test       
+`))
+	})
+
+	It("get pubsub list", func() {
+		var buf bytes.Buffer
+
+		MustBeSuccessful(env.CatchOutput(&buf).Execute("get", "pubsub", ARCH, ARCH2, "ARCH2"))
+		Expect(buf.String()).To(StringEqualTrimmedWithContext(`
+REPOSITORY PUBSUBTYPE ERROR
+ctf        test       
+ctf2       -          
+ARCH2                 repository "ARCH2" is unknown
+`))
+	})
+
+	It("get pubsub yaml", func() {
+		var buf bytes.Buffer
+
+		MustBeSuccessful(env.CatchOutput(&buf).Execute("get", "pubsub", ARCH, "-o", "yaml"))
+		Expect(buf.String()).To(StringEqualTrimmedWithContext(`
+---
+pubsub:
+  target: testtarget
+  type: test
+repository: ctf
+`))
+	})
+})
+
+////////////////////////////////////////////////////////////////////////////////
+
+const (
+	Type   = "test"
+	TypeV1 = Type + runtime.VersionSeparator + "v1"
+)
+
+type Spec struct {
+	runtime.ObjectVersionedType
+	Target string `json:"target"`
+}
+
+var (
+	_ pubsub.PubSubSpec = (*Spec)(nil)
+)
+
+func NewSpec(target string) *Spec {
+	return &Spec{runtime.NewVersionedObjectType(Type), target}
+}
+
+func (s *Spec) PubSubMethod(repo ocm.Repository) (pubsub.PubSubMethod, error) {
+	return nil, nil
+}
+
+func (s *Spec) Describe(_ ocm.Context) string {
+	return fmt.Sprintf("test pubsub")
+}

cmds/ocm/commands/ocmcmds/pubsub/get/suite_test.go

@@ -0,0 +1,13 @@
+package get_test
+
+import (
+	"testing"
+
+	. "github.com/onsi/ginkgo/v2"
+	. "github.com/onsi/gomega"
+)
+
+func TestConfig(t *testing.T) {
+	RegisterFailHandler(Fail)
+	RunSpecs(t, "get pubsub Test Suite")
+}