Skip to content
This repository was archived by the owner on Dec 11, 2023. It is now read-only.

Commit 75d802c

Browse files
tzununbekovtzununbekov
committed
Transformation's CE type logic added
1 parent 26d24b5 commit 75d802c

16 files changed

Lines changed: 399 additions & 192 deletions

File tree

‎cmd/cmd.go‎

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -110,7 +110,6 @@ func initConfig() error {
110110
return err
111111
}
112112
viper.SetDefault("context", "default")
113-
viper.SetDefault("storage", "inmemory")
114113
viper.SetDefault("triggermesh.crd", "https://github.com/triggermesh/triggermesh/releases/download/${VERSION}/triggermesh-crds.yaml")
115114
viper.SetDefault("triggermesh.version", "latest")
116115
return viper.SafeWriteConfig()

‎cmd/create/broker.go‎

Lines changed: 4 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -25,6 +25,7 @@ import (
2525
"github.com/spf13/cobra"
2626
"github.com/spf13/viper"
2727

28+
"github.com/triggermesh/tmcli/pkg/output"
2829
"github.com/triggermesh/tmcli/pkg/triggermesh"
2930
tmbroker "github.com/triggermesh/tmcli/pkg/triggermesh/broker"
3031
)
@@ -53,28 +54,21 @@ func (o *CreateOptions) broker(name string) error {
5354
}
5455

5556
log.Println("Updating manifest")
56-
restart, err := triggermesh.Create(ctx, broker, path.Join(configDir, manifestFile))
57+
restart, err := triggermesh.WriteObject(ctx, broker, path.Join(configDir, manifestFile))
5758
if err != nil {
5859
return err
5960
}
6061

6162
log.Println("Starting container")
62-
container, err := triggermesh.Start(ctx, broker, restart)
63-
if err != nil {
63+
if _, err := triggermesh.Start(ctx, broker, restart); err != nil {
6464
return err
6565
}
6666

6767
viper.Set("context", name)
6868
if err := viper.WriteConfig(); err != nil {
6969
return fmt.Errorf("write config: %w", err)
7070
}
71-
fmt.Println("---")
72-
fmt.Printf("Broker is serving events at http://0.0.0.0:%s\n", container.HostPort())
73-
fmt.Printf("Current context is set to %q\n", name)
74-
fmt.Println("To change the context run \"tmcli config set context <context name>\"")
75-
fmt.Println("\nNext steps:")
76-
fmt.Println("\ttmcli create source\t - create source that will produce events")
77-
fmt.Println("---")
7871

72+
fmt.Println(output.ComponentStatus("broker", broker, "", []string{}))
7973
return nil
8074
}

‎cmd/create/cmd.go‎

Lines changed: 44 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,9 +18,16 @@ package create
1818

1919
import (
2020
"fmt"
21+
"strings"
2122

2223
"github.com/spf13/cobra"
2324
"github.com/spf13/viper"
25+
26+
"github.com/triggermesh/tmcli/pkg/manifest"
27+
"github.com/triggermesh/tmcli/pkg/triggermesh"
28+
"github.com/triggermesh/tmcli/pkg/triggermesh/source"
29+
"github.com/triggermesh/tmcli/pkg/triggermesh/target"
30+
"github.com/triggermesh/tmcli/pkg/triggermesh/transformation"
2431
)
2532

2633
const manifestFile = "manifest.yaml"
@@ -69,3 +76,40 @@ func parse(args []string) (string, []string, error) {
6976
}
7077
return args[0], args[1:], nil
7178
}
79+
80+
func (o *CreateOptions) getObject(name, manifestPath string) (triggermesh.Component, error) {
81+
manifest := manifest.New(manifestPath)
82+
if err := manifest.Read(); err != nil {
83+
return nil, fmt.Errorf("reading manifest: %w", err)
84+
}
85+
for _, object := range manifest.Objects {
86+
if object.Metadata.Name == name {
87+
switch object.APIVersion {
88+
case "sources.triggermesh.io/v1alpha1":
89+
return source.New(object.Metadata.Name, o.CRD, object.Kind, "", o.Version, object.Spec), nil
90+
case "targets.triggermesh.io/v1alpha1":
91+
return target.New(object.Metadata.Name, o.CRD, object.Kind, "", o.Version, object.Spec), nil
92+
case "flow.triggermesh.io/v1alpha1":
93+
return transformation.New(object.Metadata.Name, o.CRD, object.Kind, "", o.Version, object.Spec), nil
94+
}
95+
}
96+
}
97+
return nil, nil
98+
}
99+
100+
func parameterFromArgs(parameter string, args []string) (string, []string) {
101+
var value string
102+
for k := 0; k < len(args); k++ {
103+
if strings.HasPrefix(args[k], "--"+parameter) {
104+
if kv := strings.Split(args[k], "="); len(kv) == 2 {
105+
value = kv[1]
106+
} else if len(args) > k+1 && !strings.HasPrefix(args[k+1], "--") {
107+
value = args[k+1]
108+
k++
109+
}
110+
args = append(args[:k-1], args[k+1:]...)
111+
break
112+
}
113+
}
114+
return value, args
115+
}

‎cmd/create/source.go‎

Lines changed: 11 additions & 27 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ import (
2626
"github.com/spf13/cobra"
2727

2828
"github.com/triggermesh/tmcli/pkg/docker"
29+
"github.com/triggermesh/tmcli/pkg/output"
2930
"github.com/triggermesh/tmcli/pkg/triggermesh"
3031
tmbroker "github.com/triggermesh/tmcli/pkg/triggermesh/broker"
3132
"github.com/triggermesh/tmcli/pkg/triggermesh/crd"
@@ -53,12 +54,13 @@ func (o *CreateOptions) NewSourceCmd() *cobra.Command {
5354
if err != nil {
5455
return err
5556
}
56-
return o.source(kind, args)
57+
name, args := parameterFromArgs("name", args)
58+
return o.source(name, kind, args)
5759
},
5860
}
5961
}
6062

61-
func (o *CreateOptions) source(kind string, args []string) error {
63+
func (o *CreateOptions) source(name, kind string, args []string) error {
6264
ctx := context.Background()
6365
configDir := path.Join(o.ConfigBase, o.Context)
6466

@@ -70,42 +72,24 @@ func (o *CreateOptions) source(kind string, args []string) error {
7072
if err != nil {
7173
return fmt.Errorf("broker object: %v", err)
7274
}
73-
brontainer, err := broker.AsContainer()
75+
port, err := broker.GetPort(ctx, client)
7476
if err != nil {
75-
return fmt.Errorf("broker container: %v", err)
76-
}
77-
brontainer, err = brontainer.LookupHostConfig(ctx, client)
78-
if err != nil {
79-
return fmt.Errorf("broker config: %v", err)
77+
return fmt.Errorf("broker port: %v", err)
8078
}
8179

82-
s := source.New(o.CRD, kind, o.Context, o.Version,
83-
append(args, fmt.Sprintf("--sink.uri=http://host.docker.internal:%s", brontainer.HostPort())))
80+
spec := append(args, fmt.Sprintf("--sink.uri=http://host.docker.internal:%s", port))
81+
82+
s := source.New(name, o.CRD, kind, o.Context, o.Version, spec)
8483

8584
log.Println("Updating manifest")
86-
restart, err := triggermesh.Create(ctx, s, path.Join(configDir, manifestFile))
85+
restart, err := triggermesh.WriteObject(ctx, s, path.Join(configDir, manifestFile))
8786
if err != nil {
8887
return err
8988
}
9089
log.Println("Starting container")
9190
if _, err := triggermesh.Start(ctx, s, restart); err != nil {
9291
return err
9392
}
94-
95-
eventTypesMessage := "This event source does not announce its event types"
96-
eventTypes, err := s.GetEventTypes()
97-
if err != nil {
98-
return err
99-
}
100-
if len(eventTypes) != 0 {
101-
eventTypesMessage = fmt.Sprintf("Event types produced by this source:\n\t%s", strings.Join(eventTypes, ", "))
102-
}
103-
fmt.Println("---")
104-
fmt.Println(eventTypesMessage)
105-
fmt.Println("\nNext steps:")
106-
fmt.Printf("\ttmcli create target <kind> --source %s [--eventTypes <types>]\t - create target that will consume events from this source\n", kind)
107-
fmt.Println("\ttmcli watch\t\t\t\t\t\t\t\t - show events flowing through the broker in the real time")
108-
fmt.Println("---")
109-
93+
fmt.Println(output.ComponentStatus("producer", s, "", []string{}))
11094
return nil
11195
}

‎cmd/create/target.go‎

Lines changed: 26 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -25,10 +25,10 @@ import (
2525

2626
"github.com/spf13/cobra"
2727

28+
"github.com/triggermesh/tmcli/pkg/output"
2829
"github.com/triggermesh/tmcli/pkg/triggermesh"
2930
tmbroker "github.com/triggermesh/tmcli/pkg/triggermesh/broker"
3031
"github.com/triggermesh/tmcli/pkg/triggermesh/crd"
31-
"github.com/triggermesh/tmcli/pkg/triggermesh/source"
3232
"github.com/triggermesh/tmcli/pkg/triggermesh/target"
3333
)
3434

@@ -53,54 +53,48 @@ func (o *CreateOptions) NewTargetCmd() *cobra.Command {
5353
if err != nil {
5454
return err
5555
}
56-
sourceFilter, args := parameterFromArgs("source", args)
56+
name, args := parameterFromArgs("name", args)
57+
eventSourceFilter, args := parameterFromArgs("source", args)
5758
eventTypesFilter, args := parameterFromArgs("eventTypes", args)
58-
if sourceFilter == "" && eventTypesFilter == "" {
59+
if eventSourceFilter == "" && eventTypesFilter == "" {
5960
return fmt.Errorf("\"--source=<kind>\" or \"--eventTypes=<a,b,c>\" is required")
6061
}
6162
var eventFilter []string
6263
if eventTypesFilter != "" {
6364
eventFilter = strings.Split(eventTypesFilter, ",")
6465
}
65-
return o.target(kind, args, sourceFilter, eventFilter)
66+
return o.target(name, kind, args, eventSourceFilter, eventFilter)
6667
},
6768
}
6869
}
6970

70-
func parameterFromArgs(parameter string, args []string) (string, []string) {
71-
var value string
72-
for k := 0; k < len(args); k++ {
73-
if strings.HasPrefix(args[k], "--"+parameter) {
74-
if kv := strings.Split(args[k], "="); len(kv) == 2 {
75-
value = kv[1]
76-
} else if len(args) > k+1 && !strings.HasPrefix(args[k+1], "--") {
77-
value = args[k+1]
78-
k++
79-
}
80-
args = append(args[:k-1], args[k+1:]...)
81-
break
82-
}
83-
}
84-
return value, args
85-
}
86-
87-
func (o *CreateOptions) target(kind string, args []string, sourceKind string, eventTypes []string) error {
71+
func (o *CreateOptions) target(name, kind string, args []string, eventSourceFilter string, eventTypesFilter []string) error {
8872
ctx := context.Background()
8973
configDir := path.Join(o.ConfigBase, o.Context)
74+
manifest := path.Join(configDir, manifestFile)
9075

91-
var s triggermesh.Component
92-
if sourceKind != "" {
93-
s = source.New(o.CRD, sourceKind, o.Context, o.Version, nil)
94-
et, err := s.(*source.Source).GetEventTypes()
76+
if eventSourceFilter != "" {
77+
c, err := o.getObject(eventSourceFilter, manifest)
78+
if err != nil {
79+
return fmt.Errorf("%q does not exist", eventSourceFilter)
80+
}
81+
producer, ok := c.(triggermesh.Producer)
82+
if !ok {
83+
return fmt.Errorf("%q is not an event producer", eventSourceFilter)
84+
}
85+
et, err := producer.GetEventTypes()
9586
if err != nil {
96-
return fmt.Errorf("source event types: %w", err)
87+
return fmt.Errorf("%q event types: %w", eventSourceFilter, err)
88+
}
89+
if len(et) == 0 {
90+
return fmt.Errorf("%q does not expose its event types", eventSourceFilter)
9791
}
98-
eventTypes = append(eventTypes, et...)
92+
eventTypesFilter = append(eventTypesFilter, et...)
9993
}
10094

101-
t := target.New(o.CRD, kind, o.Context, o.Version, args)
95+
t := target.New(name, o.CRD, kind, o.Context, o.Version, args)
10296
log.Println("Updating manifest")
103-
restart, err := triggermesh.Create(ctx, t, path.Join(configDir, manifestFile))
97+
restart, err := triggermesh.WriteObject(ctx, t, manifest)
10498
if err != nil {
10599
return err
106100
}
@@ -110,25 +104,14 @@ func (o *CreateOptions) target(kind string, args []string, sourceKind string, ev
110104
return err
111105
}
112106

113-
tr := tmbroker.NewTrigger(fmt.Sprintf("%s-trigger", t.GetName()), o.Context, configDir, eventTypes)
107+
tr := tmbroker.NewTrigger(fmt.Sprintf("%s-trigger", t.GetName()), o.Context, configDir, eventTypesFilter)
114108
tr.SetTarget(container.Name, fmt.Sprintf("http://host.docker.internal:%s", container.HostPort()))
115109
if err := tr.UpdateBrokerConfig(); err != nil {
116110
return fmt.Errorf("broker config: %w", err)
117111
}
118112
if err := tr.UpdateManifest(); err != nil {
119113
return fmt.Errorf("broker manifest: %w", err)
120114
}
121-
122-
sourceMsg := strings.Join(eventTypes, ", ")
123-
if s != nil {
124-
sourceMsg = fmt.Sprintf("%s(%s)", s.GetKind(), sourceMsg)
125-
}
126-
fmt.Println("---")
127-
fmt.Printf("Target is subscribed to:\n\t%s\n", sourceMsg)
128-
fmt.Println("\nNext steps:")
129-
fmt.Println("\ttmcli watch\t - show events flowing through the broker in the real time")
130-
fmt.Println("\ttmcli dump\t - dump Kubernetes manifest")
131-
fmt.Println("---")
132-
115+
fmt.Println(output.ComponentStatus("consumer", t, eventSourceFilter, eventTypesFilter))
133116
return nil
134117
}

0 commit comments

Comments
 (0)