diff --git a/pkg/fission-cli/cli.go b/pkg/fission-cli/cli.go index cb7432f5..cddcf8a8 100644 --- a/pkg/fission-cli/cli.go +++ b/pkg/fission-cli/cli.go @@ -35,6 +35,7 @@ import ( "github.com/fission/fission/pkg/fission-cli/cmd/function" "github.com/fission/fission/pkg/fission-cli/cmd/httptrigger" "github.com/fission/fission/pkg/fission-cli/cmd/kubewatch" + "github.com/fission/fission/pkg/fission-cli/cmd/mqtrigger" _package "github.com/fission/fission/pkg/fission-cli/cmd/package" plugincmd "github.com/fission/fission/pkg/fission-cli/cmd/plugin" "github.com/fission/fission/pkg/fission-cli/cmd/spec" @@ -189,11 +190,10 @@ func NewCliApp() *cli.App { mqtMaxRetries := cli.IntFlag{Name: "maxretries", Value: 0, Usage: "Maximum number of times the function will be retried upon failure (optional; default is 0)"} mqtMsgContentType := cli.StringFlag{Name: "contenttype, c", Value: "application/json", Usage: "Content type of messages that publish to the topic (optional)"} mqtSubcommands := []cli.Command{ - {Name: "create", Aliases: []string{"add"}, Usage: "Create Message queue trigger", Flags: []cli.Flag{mqtNameFlag, mqtFnNameFlag, fnNamespaceFlag, mqtMQTypeFlag, mqtTopicFlag, mqtRespTopicFlag, mqtErrorTopicFlag, mqtMaxRetries, mqtMsgContentType, specSaveFlag}, Action: mqtCreate}, - {Name: "get", Usage: "Get message queue trigger", Flags: []cli.Flag{triggerNamespaceFlag}, Action: mqtGet}, - {Name: "update", Usage: "Update message queue trigger", Flags: []cli.Flag{mqtNameFlag, triggerNamespaceFlag, mqtTopicFlag, mqtRespTopicFlag, mqtErrorTopicFlag, mqtMaxRetries, mqtFnNameFlag, mqtMsgContentType}, Action: mqtUpdate}, - {Name: "delete", Usage: "Delete message queue trigger", Flags: []cli.Flag{mqtNameFlag, triggerNamespaceFlag}, Action: mqtDelete}, - {Name: "list", Usage: "List message queue triggers", Flags: []cli.Flag{mqtMQTypeFlag, triggerNamespaceFlag}, Action: mqtList}, + {Name: "create", Aliases: []string{"add"}, Usage: "Create Message queue trigger", Flags: []cli.Flag{mqtNameFlag, mqtFnNameFlag, fnNamespaceFlag, mqtMQTypeFlag, mqtTopicFlag, mqtRespTopicFlag, mqtErrorTopicFlag, mqtMaxRetries, mqtMsgContentType, specSaveFlag}, Action: urfavecli.Wrapper(mqtrigger.Create)}, + {Name: "update", Usage: "Update message queue trigger", Flags: []cli.Flag{mqtNameFlag, triggerNamespaceFlag, mqtTopicFlag, mqtRespTopicFlag, mqtErrorTopicFlag, mqtMaxRetries, mqtFnNameFlag, mqtMsgContentType}, Action: urfavecli.Wrapper(mqtrigger.Update)}, + {Name: "delete", Usage: "Delete message queue trigger", Flags: []cli.Flag{mqtNameFlag, triggerNamespaceFlag}, Action: urfavecli.Wrapper(mqtrigger.Delete)}, + {Name: "list", Usage: "List message queue triggers", Flags: []cli.Flag{mqtMQTypeFlag, triggerNamespaceFlag}, Action: urfavecli.Wrapper(mqtrigger.List)}, } // Recorders diff --git a/pkg/fission-cli/cmd/mqtrigger/create.go b/pkg/fission-cli/cmd/mqtrigger/create.go new file mode 100644 index 00000000..3593a311 --- /dev/null +++ b/pkg/fission-cli/cmd/mqtrigger/create.go @@ -0,0 +1,162 @@ +/* +Copyright 2019 The Fission Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package mqtrigger + +import ( + "fmt" + + "github.com/pkg/errors" + uuid "github.com/satori/go.uuid" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + fv1 "github.com/fission/fission/pkg/apis/fission.io/v1" + "github.com/fission/fission/pkg/controller/client" + "github.com/fission/fission/pkg/fission-cli/cliwrapper/cli" + "github.com/fission/fission/pkg/fission-cli/cmd" + "github.com/fission/fission/pkg/fission-cli/cmd/spec" + "github.com/fission/fission/pkg/fission-cli/log" + "github.com/fission/fission/pkg/types" +) + +type CreateSubCommand struct { + client *client.Client + trigger *fv1.MessageQueueTrigger +} + +func Create(flags cli.Input) error { + opts := CreateSubCommand{ + client: cmd.GetServer(flags), + } + return opts.do(flags) +} + +func (opts *CreateSubCommand) do(flags cli.Input) error { + err := opts.complete(flags) + if err != nil { + return err + } + return opts.run(flags) +} + +func (opts *CreateSubCommand) complete(flags cli.Input) error { + mqtName := flags.String("name") + if len(mqtName) == 0 { + mqtName = uuid.NewV4().String() + } + fnName := flags.String("function") + if len(fnName) == 0 { + log.Fatal("Need a function name to create a trigger, use --function") + } + fnNamespace := flags.String("fnNamespace") + + var mqType fv1.MessageQueueType + switch flags.String("mqtype") { + case "": + mqType = types.MessageQueueTypeNats + case types.MessageQueueTypeNats: + mqType = types.MessageQueueTypeNats + case types.MessageQueueTypeASQ: + mqType = types.MessageQueueTypeASQ + case types.MessageQueueTypeKafka: + mqType = types.MessageQueueTypeKafka + + default: + log.Fatal("Unknown message queue type, currently only \"nats-streaming, azure-storage-queue, kafka \" is supported") + + } + + // TODO: check topic availability + topic := flags.String("topic") + if len(topic) == 0 { + log.Fatal("Topic cannot be empty") + } + respTopic := flags.String("resptopic") + + if topic == respTopic { + // TODO maybe this should just be a warning, perhaps + // allow it behind a --force flag + log.Fatal("Listen topic should not equal to response topic") + } + + errorTopic := flags.String("errortopic") + + maxRetries := flags.Int("maxretries") + + if maxRetries < 0 { + log.Fatal("Maximum number of retries must be a natural number, default is 0") + } + + contentType := flags.String("contenttype") + if len(contentType) == 0 { + contentType = "application/json" + } + + err := checkMQTopicAvailability(mqType, topic, respTopic) + if err != nil { + return err + } + + opts.trigger = &fv1.MessageQueueTrigger{ + Metadata: metav1.ObjectMeta{ + Name: mqtName, + Namespace: fnNamespace, + }, + Spec: fv1.MessageQueueTriggerSpec{ + FunctionReference: fv1.FunctionReference{ + Type: types.FunctionReferenceTypeFunctionName, + Name: fnName, + }, + MessageQueueType: mqType, + Topic: topic, + ResponseTopic: respTopic, + ErrorTopic: errorTopic, + MaxRetries: maxRetries, + ContentType: contentType, + }, + } + + return nil +} + +func (opts *CreateSubCommand) run(flags cli.Input) error { + // if we're writing a spec, don't call the API + if flags.Bool("spec") { + specFile := fmt.Sprintf("mqtrigger-%v.yaml", opts.trigger.Metadata.Name) + err := spec.SpecSave(*opts.trigger, specFile) + if err != nil { + return errors.Wrap(err, "error creating message queue trigger spec") + } + return nil + } + + _, err := opts.client.MessageQueueTriggerCreate(opts.trigger) + if err != nil { + return errors.Wrap(err, "create message queue trigger") + } + + fmt.Printf("message queue trigger '%s' created\n", opts.trigger.Metadata.Name) + return nil +} + +func checkMQTopicAvailability(mqType fv1.MessageQueueType, topics ...string) error { + for _, t := range topics { + if len(t) > 0 && !fv1.IsTopicValid(mqType, t) { + return errors.Errorf("Invalid topic for %s: %s", mqType, t) + } + } + return nil +} diff --git a/pkg/fission-cli/cmd/mqtrigger/delete.go b/pkg/fission-cli/cmd/mqtrigger/delete.go new file mode 100644 index 00000000..9b194de6 --- /dev/null +++ b/pkg/fission-cli/cmd/mqtrigger/delete.go @@ -0,0 +1,67 @@ +/* +Copyright 2019 The Fission Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package mqtrigger + +import ( + "fmt" + + "github.com/pkg/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + + "github.com/fission/fission/pkg/controller/client" + "github.com/fission/fission/pkg/fission-cli/cliwrapper/cli" + "github.com/fission/fission/pkg/fission-cli/cmd" +) + +type DeleteSubCommand struct { + client *client.Client + metadata *metav1.ObjectMeta +} + +func Delete(flags cli.Input) error { + opts := DeleteSubCommand{ + client: cmd.GetServer(flags), + } + return opts.do(flags) +} + +func (opts *DeleteSubCommand) do(flags cli.Input) error { + err := opts.complete(flags) + if err != nil { + return err + } + return opts.run(flags) +} + +func (opts *DeleteSubCommand) complete(flags cli.Input) error { + m, err := cmd.GetMetadata("name", "triggerns", flags) + if err != nil { + return err + } + opts.metadata = m + return nil +} + +func (opts *DeleteSubCommand) run(flags cli.Input) error { + err := opts.client.WatchDelete(opts.metadata) + if err != nil { + return errors.Wrap(err, "error deleting message queue trigger") + } + + fmt.Printf("message queue trigger '%v' deleted\n", opts.metadata.Name) + return nil +} diff --git a/pkg/fission-cli/cmd/mqtrigger/list.go b/pkg/fission-cli/cmd/mqtrigger/list.go new file mode 100644 index 00000000..866a26e6 --- /dev/null +++ b/pkg/fission-cli/cmd/mqtrigger/list.go @@ -0,0 +1,73 @@ +/* +Copyright 2019 The Fission Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package mqtrigger + +import ( + "fmt" + "os" + "text/tabwriter" + + "github.com/pkg/errors" + + "github.com/fission/fission/pkg/controller/client" + "github.com/fission/fission/pkg/fission-cli/cliwrapper/cli" + "github.com/fission/fission/pkg/fission-cli/cmd" +) + +type ListSubCommand struct { + client *client.Client + namespace string +} + +func List(flags cli.Input) error { + opts := ListSubCommand{ + client: cmd.GetServer(flags), + } + return opts.do(flags) +} + +func (opts *ListSubCommand) do(flags cli.Input) error { + err := opts.complete(flags) + if err != nil { + return err + } + return opts.run(flags) +} + +func (opts *ListSubCommand) complete(flags cli.Input) error { + opts.namespace = flags.String("triggerns") + return nil +} + +func (opts *ListSubCommand) run(flags cli.Input) error { + mqts, err := opts.client.MessageQueueTriggerList(flags.String("mqtype"), opts.namespace) + if err != nil { + return errors.Wrap(err, "error listing message queue triggers") + } + + w := tabwriter.NewWriter(os.Stdout, 0, 0, 1, ' ', 0) + + fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\t%v\t%v\t%v\n", + "NAME", "FUNCTION_NAME", "MESSAGE_QUEUE_TYPE", "TOPIC", "RESPONSE_TOPIC", "ERROR_TOPIC", "MAX_RETRIES", "PUB_MSG_CONTENT_TYPE") + for _, mqt := range mqts { + fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\t%v\t%v\t%v\n", + mqt.Metadata.Name, mqt.Spec.FunctionReference.Name, mqt.Spec.MessageQueueType, mqt.Spec.Topic, mqt.Spec.ResponseTopic, mqt.Spec.ErrorTopic, mqt.Spec.MaxRetries, mqt.Spec.ContentType) + } + w.Flush() + + return nil +} diff --git a/pkg/fission-cli/cmd/mqtrigger/update.go b/pkg/fission-cli/cmd/mqtrigger/update.go new file mode 100644 index 00000000..6091212d --- /dev/null +++ b/pkg/fission-cli/cmd/mqtrigger/update.go @@ -0,0 +1,117 @@ +/* +Copyright 2019 The Fission Authors. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package mqtrigger + +import ( + "fmt" + + "github.com/pkg/errors" + + fv1 "github.com/fission/fission/pkg/apis/fission.io/v1" + "github.com/fission/fission/pkg/controller/client" + "github.com/fission/fission/pkg/fission-cli/cliwrapper/cli" + "github.com/fission/fission/pkg/fission-cli/cmd" +) + +type UpdateSubCommand struct { + client *client.Client + trigger *fv1.MessageQueueTrigger +} + +func Update(flags cli.Input) error { + opts := UpdateSubCommand{ + client: cmd.GetServer(flags), + } + return opts.do(flags) +} + +func (opts *UpdateSubCommand) do(flags cli.Input) error { + err := opts.complete(flags) + if err != nil { + return err + } + return opts.run(flags) +} + +func (opts *UpdateSubCommand) complete(flags cli.Input) error { + m, err := cmd.GetMetadata("name", "triggerns", flags) + if err != nil { + return err + } + + mqt, err := opts.client.MessageQueueTriggerGet(m) + if err != nil { + return errors.Wrap(err, "error getting message queue trigger") + } + + topic := flags.String("topic") + respTopic := flags.String("resptopic") + errorTopic := flags.String("errortopic") + maxRetries := flags.Int("maxretries") + fnName := flags.String("function") + contentType := flags.String("contenttype") + + // TODO : Find out if we can make a call to checkIfFunctionExists, in the same ns more importantly. + + err = checkMQTopicAvailability(mqt.Spec.MessageQueueType, topic, respTopic) + if err != nil { + return err + } + + updated := false + if len(topic) > 0 { + mqt.Spec.Topic = topic + updated = true + } + if len(respTopic) > 0 { + mqt.Spec.ResponseTopic = respTopic + updated = true + } + if len(errorTopic) > 0 { + mqt.Spec.ErrorTopic = errorTopic + updated = true + } + if maxRetries > -1 { + mqt.Spec.MaxRetries = maxRetries + updated = true + } + if len(fnName) > 0 { + mqt.Spec.FunctionReference.Name = fnName + updated = true + } + if len(contentType) > 0 { + mqt.Spec.ContentType = contentType + updated = true + } + + if !updated { + return errors.New("Nothing to update. Use --topic, --resptopic, --errortopic, --maxretries or --function.") + } + opts.trigger = mqt + + return nil +} + +func (opts *UpdateSubCommand) run(flags cli.Input) error { + _, err := opts.client.MessageQueueTriggerUpdate(opts.trigger) + if err != nil { + return errors.Wrap(err, "error updating message queue trigger") + } + + fmt.Printf("message queue trigger '%v' updated\n", opts.trigger.Metadata.Name) + return nil +} diff --git a/pkg/fission-cli/cmd/util.go b/pkg/fission-cli/cmd/util.go index 1f0aac1d..b9630553 100644 --- a/pkg/fission-cli/cmd/util.go +++ b/pkg/fission-cli/cmd/util.go @@ -129,7 +129,7 @@ func GetSpecDir(flags cli.Input) string { func GetMetadata(nameFlagText string, namespaceFlagText string, flags cli.Input) (*metav1.ObjectMeta, error) { name := flags.String(nameFlagText) if len(name) == 0 { - return nil, errors.New("need a resource name, use --name") + return nil, errors.Errorf("need a resource name, use --%v", nameFlagText) } ns := flags.String(namespaceFlagText) diff --git a/pkg/fission-cli/mqtrigger.go b/pkg/fission-cli/mqtrigger.go deleted file mode 100644 index 4a88c198..00000000 --- a/pkg/fission-cli/mqtrigger.go +++ /dev/null @@ -1,236 +0,0 @@ -/* -Copyrigtt 2017 The Fission Authors. - -Licensed under the Apache License, Version 2.0 (the "License"); -you may not use this file except in compliance with the License. -You may obtain a copy of the License at - - http://www.apache.org/licenses/LICENSE-2.0 - -Unless required by applicable law or agreed to in writing, software -distributed under the License is distributed on an "AS IS" BASIS, -WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -See the License for the specific language governing permissions and -limitations under the License. -*/ - -package fission_cli - -import ( - "fmt" - "os" - "text/tabwriter" - - "github.com/satori/go.uuid" - "github.com/urfave/cli" - metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" - - fv1 "github.com/fission/fission/pkg/apis/fission.io/v1" - "github.com/fission/fission/pkg/fission-cli/cmd/spec" - "github.com/fission/fission/pkg/fission-cli/log" - "github.com/fission/fission/pkg/fission-cli/util" - "github.com/fission/fission/pkg/types" -) - -func mqtCreate(c *cli.Context) error { - client := util.GetApiClient(c.GlobalString("server")) - - mqtName := c.String("name") - if len(mqtName) == 0 { - mqtName = uuid.NewV4().String() - } - fnName := c.String("function") - if len(fnName) == 0 { - log.Fatal("Need a function name to create a trigger, use --function") - } - fnNamespace := c.String("fnNamespace") - - var mqType fv1.MessageQueueType - switch c.String("mqtype") { - case "": - mqType = types.MessageQueueTypeNats - case types.MessageQueueTypeNats: - mqType = types.MessageQueueTypeNats - case types.MessageQueueTypeASQ: - mqType = types.MessageQueueTypeASQ - case types.MessageQueueTypeKafka: - mqType = types.MessageQueueTypeKafka - - default: - log.Fatal("Unknown message queue type, currently only \"nats-streaming, azure-storage-queue, kafka \" is supported") - - } - - // TODO: check topic availability - topic := c.String("topic") - if len(topic) == 0 { - log.Fatal("Topic cannot be empty") - } - respTopic := c.String("resptopic") - - if topic == respTopic { - // TODO maybe this should just be a warning, perhaps - // allow it behind a --force flag - log.Fatal("Listen topic should not equal to response topic") - } - - errorTopic := c.String("errortopic") - - maxRetries := c.Int("maxretries") - - if maxRetries < 0 { - log.Fatal("Maximum number of retries must be a natural number, default is 0") - } - - contentType := c.String("contenttype") - if len(contentType) == 0 { - contentType = "application/json" - } - - checkMQTopicAvailability(mqType, topic, respTopic) - - mqt := &fv1.MessageQueueTrigger{ - Metadata: metav1.ObjectMeta{ - Name: mqtName, - Namespace: fnNamespace, - }, - Spec: fv1.MessageQueueTriggerSpec{ - FunctionReference: fv1.FunctionReference{ - Type: types.FunctionReferenceTypeFunctionName, - Name: fnName, - }, - MessageQueueType: mqType, - Topic: topic, - ResponseTopic: respTopic, - ErrorTopic: errorTopic, - MaxRetries: maxRetries, - ContentType: contentType, - }, - } - - // if we're writing a spec, don't call the API - if c.Bool("spec") { - specFile := fmt.Sprintf("mqtrigger-%v.yaml", mqtName) - err := spec.SpecSave(*mqt, specFile) - util.CheckErr(err, "create message queue trigger spec") - return nil - } - - _, err := client.MessageQueueTriggerCreate(mqt) - util.CheckErr(err, "create message queue trigger") - - fmt.Printf("trigger '%s' created\n", mqtName) - return err -} - -func mqtGet(c *cli.Context) error { - return nil -} - -func mqtUpdate(c *cli.Context) error { - client := util.GetApiClient(c.GlobalString("server")) - mqtName := c.String("name") - if len(mqtName) == 0 { - log.Fatal("Need name of trigger, use --name") - } - mqtNs := c.String("triggerns") - - topic := c.String("topic") - respTopic := c.String("resptopic") - errorTopic := c.String("errortopic") - maxRetries := c.Int("maxretries") - fnName := c.String("function") - contentType := c.String("contenttype") - - mqt, err := client.MessageQueueTriggerGet(&metav1.ObjectMeta{ - Name: mqtName, - Namespace: mqtNs, - }) - util.CheckErr(err, "get Time trigger") - - // TODO : Find out if we can make a call to checkIfFunctionExists, in the same ns more importantly. - - checkMQTopicAvailability(mqt.Spec.MessageQueueType, topic, respTopic) - - updated := false - if len(topic) > 0 { - mqt.Spec.Topic = topic - updated = true - } - if len(respTopic) > 0 { - mqt.Spec.ResponseTopic = respTopic - updated = true - } - if len(errorTopic) > 0 { - mqt.Spec.ErrorTopic = errorTopic - updated = true - } - if maxRetries > -1 { - mqt.Spec.MaxRetries = maxRetries - updated = true - } - if len(fnName) > 0 { - mqt.Spec.FunctionReference.Name = fnName - updated = true - } - if len(contentType) > 0 { - mqt.Spec.ContentType = contentType - updated = true - } - - if !updated { - log.Fatal("Nothing to update. Use --topic, --resptopic, --errortopic, --maxretries or --function.") - } - - _, err = client.MessageQueueTriggerUpdate(mqt) - util.CheckErr(err, "update Time trigger") - - fmt.Printf("trigger '%v' updated\n", mqtName) - return nil -} - -func mqtDelete(c *cli.Context) error { - client := util.GetApiClient(c.GlobalString("server")) - mqtName := c.String("name") - if len(mqtName) == 0 { - log.Fatal("Need name of trigger to delete, use --name") - } - mqtNs := c.String("triggerns") - - err := client.MessageQueueTriggerDelete(&metav1.ObjectMeta{ - Name: mqtName, - Namespace: mqtNs, - }) - util.CheckErr(err, "delete trigger") - - fmt.Printf("trigger '%v' deleted\n", mqtName) - return nil -} - -func mqtList(c *cli.Context) error { - client := util.GetApiClient(c.GlobalString("server")) - mqtNs := c.String("triggerns") - - mqts, err := client.MessageQueueTriggerList(c.String("mqtype"), mqtNs) - util.CheckErr(err, "list message queue triggers") - - w := tabwriter.NewWriter(os.Stdout, 0, 0, 1, ' ', 0) - - fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\t%v\t%v\t%v\n", - "NAME", "FUNCTION_NAME", "MESSAGE_QUEUE_TYPE", "TOPIC", "RESPONSE_TOPIC", "ERROR_TOPIC", "MAX_RETRIES", "PUB_MSG_CONTENT_TYPE") - for _, mqt := range mqts { - fmt.Fprintf(w, "%v\t%v\t%v\t%v\t%v\t%v\t%v\t%v\n", - mqt.Metadata.Name, mqt.Spec.FunctionReference.Name, mqt.Spec.MessageQueueType, mqt.Spec.Topic, mqt.Spec.ResponseTopic, mqt.Spec.ErrorTopic, mqt.Spec.MaxRetries, mqt.Spec.ContentType) - } - w.Flush() - - return nil -} - -func checkMQTopicAvailability(mqType fv1.MessageQueueType, topics ...string) { - for _, t := range topics { - if len(t) > 0 && !fv1.IsTopicValid(mqType, t) { - log.Fatal(fmt.Sprintf("Invalid topic for %s: %s", mqType, t)) - } - } -}