// Copyright (c) Abstract Machines // SPDX-License-Identifier: Apache-2.0 package cli import ( "encoding/csv" "encoding/json" "errors" "fmt" "io" "math/rand" "os" "path/filepath" "time" "github.com/0x6flab/namegenerator" smqsdk "github.com/absmach/magistrala/pkg/sdk" "github.com/spf13/cobra" ) const ( jsonExt = ".json" csvExt = ".csv" PublishType = "publish" SubscribeType = "subscribe" ) var ( msgFormat = `[{"bn":"provision:", "bu":"V", "t": %d, "bver":5, "n":"voltage", "u":"V", "v":%d}]` namesgenerator = namegenerator.NewGenerator() ) var cmdProvision = []cobra.Command{ { Use: "clients ", Short: "Provision clients", Long: `Bulk create clients`, Run: func(cmd *cobra.Command, args []string) { if len(args) != 3 { logUsageCmd(*cmd, cmd.Use) return } if _, err := os.Stat(args[0]); os.IsNotExist(err) { logErrorCmd(*cmd, err) return } clients, err := clientsFromFile(args[0]) if err != nil { logErrorCmd(*cmd, err) return } clients, err = sdk.CreateClients(cmd.Context(), clients, args[1], args[2]) if err != nil { logErrorCmd(*cmd, err) return } logJSONCmd(*cmd, clients) }, }, { Use: "channels ", Short: "Provision channels", Long: `Bulk create channels`, Run: func(cmd *cobra.Command, args []string) { if len(args) != 3 { logUsageCmd(*cmd, cmd.Use) return } channels, err := channelsFromFile(args[0]) if err != nil { logErrorCmd(*cmd, err) return } var chs []smqsdk.Channel for _, c := range channels { c, err = sdk.CreateChannel(cmd.Context(), c, args[1], args[2]) if err != nil { logErrorCmd(*cmd, err) return } chs = append(chs, c) } channels = chs logJSONCmd(*cmd, channels) }, }, { Use: "connect ", Short: "Provision connections", Long: `Bulk connect clients to channels`, Run: func(cmd *cobra.Command, args []string) { if len(args) != 3 { logUsageCmd(*cmd, cmd.Use) return } connIDs, err := connectionsFromFile(args[0]) if err != nil { logErrorCmd(*cmd, err) return } for _, conn := range connIDs { if err := sdk.Connect(cmd.Context(), conn, args[1], args[2]); err != nil { logErrorCmd(*cmd, err) return } } logOKCmd(*cmd) }, }, { Use: "test", Short: "test", Long: `Provisions test setup: one test user, two clients and two channels. \ Connect both clients to one of the channels, \ and only on client to other channel.`, Run: func(cmd *cobra.Command, args []string) { numClients := 2 numChan := 2 clients := []smqsdk.Client{} channels := []smqsdk.Channel{} if len(args) != 0 { logUsageCmd(*cmd, cmd.Use) return } // Create test user name := namesgenerator.Generate() user := smqsdk.User{ FirstName: name, Email: fmt.Sprintf("%s@email.com", name), Credentials: smqsdk.Credentials{ Username: name, Secret: "12345678", }, Status: smqsdk.EnabledStatus, } user, err := sdk.CreateUser(cmd.Context(), user, "") if err != nil { logErrorCmd(*cmd, err) return } ut, err := sdk.CreateToken(cmd.Context(), smqsdk.Login{Username: user.Credentials.Username, Password: user.Credentials.Secret}) if err != nil { logErrorCmd(*cmd, err) return } // create domain domain := smqsdk.Domain{ Name: fmt.Sprintf("%s-domain", name), Status: smqsdk.EnabledStatus, } domain, err = sdk.CreateDomain(cmd.Context(), domain, ut.AccessToken) if err != nil { logErrorCmd(*cmd, err) return } ut, err = sdk.CreateToken(cmd.Context(), smqsdk.Login{Username: user.Email, Password: user.Credentials.Secret}) if err != nil { logErrorCmd(*cmd, err) return } // Create clients for i := 0; i < numClients; i++ { t := smqsdk.Client{ Name: fmt.Sprintf("%s-client-%d", name, i), Status: smqsdk.EnabledStatus, } clients = append(clients, t) } clients, err = sdk.CreateClients(cmd.Context(), clients, domain.ID, ut.AccessToken) if err != nil { logErrorCmd(*cmd, err) return } // Create channels for i := 0; i < numChan; i++ { c := smqsdk.Channel{ Name: fmt.Sprintf("%s-channel-%d", name, i), Status: smqsdk.EnabledStatus, } c, err = sdk.CreateChannel(cmd.Context(), c, domain.ID, ut.AccessToken) if err != nil { logErrorCmd(*cmd, err) return } channels = append(channels, c) } // Connect clients to channels - first client to both channels, second only to first conIDs := smqsdk.Connection{ ChannelIDs: []string{channels[0].ID}, ClientIDs: []string{clients[0].ID}, Types: []string{PublishType, SubscribeType}, } if err := sdk.Connect(cmd.Context(), conIDs, domain.ID, ut.AccessToken); err != nil { logErrorCmd(*cmd, err) return } conIDs = smqsdk.Connection{ ChannelIDs: []string{channels[1].ID}, ClientIDs: []string{clients[0].ID}, Types: []string{PublishType, SubscribeType}, } if err := sdk.Connect(cmd.Context(), conIDs, domain.ID, ut.AccessToken); err != nil { logErrorCmd(*cmd, err) return } conIDs = smqsdk.Connection{ ChannelIDs: []string{channels[0].ID}, ClientIDs: []string{clients[1].ID}, Types: []string{PublishType, SubscribeType}, } if err := sdk.Connect(cmd.Context(), conIDs, domain.ID, ut.AccessToken); err != nil { logErrorCmd(*cmd, err) return } // send message to test connectivity if err := sdk.SendMessage(cmd.Context(), domain.ID, channels[0].ID, clients[0].Credentials.Secret, fmt.Sprintf(msgFormat, time.Now().Unix(), rand.Int())); err != nil { logErrorCmd(*cmd, err) return } if err := sdk.SendMessage(cmd.Context(), domain.ID, channels[0].ID, clients[1].Credentials.Secret, fmt.Sprintf(msgFormat, time.Now().Unix(), rand.Int())); err != nil { logErrorCmd(*cmd, err) return } if err := sdk.SendMessage(cmd.Context(), domain.ID, channels[1].ID, clients[0].Credentials.Secret, fmt.Sprintf(msgFormat, time.Now().Unix(), rand.Int())); err != nil { logErrorCmd(*cmd, err) return } logJSONCmd(*cmd, user, ut, clients, channels) }, }, } // NewProvisionCmd returns provision command. func NewProvisionCmd() *cobra.Command { cmd := cobra.Command{ Use: "provision [clients | channels | connect | test]", Short: "Provision clients and channels from a config file", Long: `Provision clients and channels: use json or csv file to bulk provision clients and channels`, } for i := range cmdProvision { cmd.AddCommand(&cmdProvision[i]) } return &cmd } func clientsFromFile(path string) ([]smqsdk.Client, error) { if _, err := os.Stat(path); os.IsNotExist(err) { return []smqsdk.Client{}, err } file, err := os.OpenFile(path, os.O_RDONLY, os.ModePerm) if err != nil { return []smqsdk.Client{}, err } defer file.Close() clients := []smqsdk.Client{} switch filepath.Ext(path) { case csvExt: reader := csv.NewReader(file) for { l, err := reader.Read() if err == io.EOF { break } if err != nil { return []smqsdk.Client{}, err } if len(l) < 1 { return []smqsdk.Client{}, errors.New("empty line found in file") } client := smqsdk.Client{ Name: l[0], } clients = append(clients, client) } case jsonExt: err := json.NewDecoder(file).Decode(&clients) if err != nil { return []smqsdk.Client{}, err } default: return []smqsdk.Client{}, err } return clients, nil } func channelsFromFile(path string) ([]smqsdk.Channel, error) { if _, err := os.Stat(path); os.IsNotExist(err) { return []smqsdk.Channel{}, err } file, err := os.OpenFile(path, os.O_RDONLY, os.ModePerm) if err != nil { return []smqsdk.Channel{}, err } defer file.Close() channels := []smqsdk.Channel{} switch filepath.Ext(path) { case csvExt: reader := csv.NewReader(file) for { l, err := reader.Read() if err == io.EOF { break } if err != nil { return []smqsdk.Channel{}, err } if len(l) < 1 { return []smqsdk.Channel{}, errors.New("empty line found in file") } channel := smqsdk.Channel{ Name: l[0], } channels = append(channels, channel) } case jsonExt: err := json.NewDecoder(file).Decode(&channels) if err != nil { return []smqsdk.Channel{}, err } default: return []smqsdk.Channel{}, err } return channels, nil } func connectionsFromFile(path string) ([]smqsdk.Connection, error) { if _, err := os.Stat(path); os.IsNotExist(err) { return []smqsdk.Connection{}, err } file, err := os.OpenFile(path, os.O_RDONLY, os.ModePerm) if err != nil { return []smqsdk.Connection{}, err } defer file.Close() connections := []smqsdk.Connection{} switch filepath.Ext(path) { case csvExt: reader := csv.NewReader(file) for { l, err := reader.Read() if err == io.EOF { break } if err != nil { return []smqsdk.Connection{}, err } if len(l) < 1 { return []smqsdk.Connection{}, errors.New("empty line found in file") } connections = append(connections, smqsdk.Connection{ ClientIDs: []string{l[0]}, ChannelIDs: []string{l[1]}, Types: []string{PublishType, SubscribeType}, }) } case jsonExt: err := json.NewDecoder(file).Decode(&connections) if err != nil { return []smqsdk.Connection{}, err } default: return []smqsdk.Connection{}, err } return connections, nil }