Skip to content

Commit

Permalink
Merge pull request #2 from rodneyosodo/rod-move-supermq
Browse files Browse the repository at this point in the history 
move to supermq
  • Loading branch information
@nyagamunene
nyagamunene authored Jan 24, 2025
2 parents a65a71f + f5d57cb commit 585258b
Show file tree
Hide file tree
Showing 19 changed files with 151 additions and 161 deletions.
91 changes: 42 additions & 49 deletions cli/provision.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,8 +6,8 @@ import (
"strings"

"github.com/0x6flab/namegenerator"
smqSDK "github.com/absmach/magistrala/pkg/sdk/go"
"github.com/absmach/supermq/pkg/errors"
smqSDK "github.com/absmach/supermq/pkg/sdk"
"github.com/charmbracelet/huh"
"github.com/spf13/cobra"
)
Expand Down Expand Up @@ -36,44 +36,44 @@ var provisionCmd = &cobra.Command{
Long: `Provision necessary resources for Propeller operation.`,
Run: func(cmd *cobra.Command, args []string) {
var (
identity string
secret string
username string
password string
err error
token smqSDK.Token
domainName string
domainAlias string
domainPermission string
domain smqSDK.Domain
managerThingName string
managerThing smqSDK.Thing
propletThingName string
propletThing smqSDK.Thing
managerClientName string
managerClient smqSDK.Client
propletClientName string
propletClient smqSDK.Client
managerChannelName string
managerChannel smqSDK.Channel
)
form := huh.NewForm(
huh.NewGroup(
huh.NewInput().
Title("Enter your identity (e-mail)?").
Value(&identity).
Title("Enter your username?").
Value(&username).
Validate(func(str string) error {
if str == "" {
return errors.New("identity is required")
return errors.New("username is required")
}

return nil
}),
huh.NewInput().
Title("Enter your secret").
Title("Enter your password").
EchoMode(huh.EchoModePassword).
Value(&secret).
Value(&password).
Validate(func(str string) error {
if str == "" {
return errors.New("secret is required")
return errors.New("password is required")
}
u := smqSDK.Login{
Identity: identity,
Secret: secret,
Username: username,
Password: password,
}

token, err = smqsdk.CreateToken(u)
Expand Down Expand Up @@ -121,18 +121,18 @@ var provisionCmd = &cobra.Command{
),
huh.NewGroup(
huh.NewInput().
Title("Enter your manager thing name(leave empty to auto generate)").
Value(&managerThingName).
Title("Enter your manager client name(leave empty to auto generate)").
Value(&managerClientName).
Validate(func(str string) error {
if str == "" {
managerThingName = namegen.Generate()
managerClientName = namegen.Generate()
}
managerThing = smqSDK.Thing{
Name: managerThingName,
managerClient = smqSDK.Client{
Name: managerClientName,
Tags: []string{"manager", "propeller"},
Status: "enabled",
}
managerThing, err = smqsdk.CreateThing(managerThing, domain.ID, token.AccessToken)
managerClient, err = smqsdk.CreateClient(managerClient, domain.ID, token.AccessToken)
if err != nil {
return errors.Wrap(errFailedClientCreation, err)
}
Expand All @@ -142,18 +142,18 @@ var provisionCmd = &cobra.Command{
),
huh.NewGroup(
huh.NewInput().
Title("Enter your proplet thing name(leave empty to auto generate)").
Value(&propletThingName).
Title("Enter your proplet client name(leave empty to auto generate)").
Value(&propletClientName).
Validate(func(str string) error {
if str == "" {
propletThingName = namegen.Generate()
propletClientName = namegen.Generate()
}
propletThing = smqSDK.Thing{
Name: propletThingName,
propletClient = smqSDK.Client{
Name: propletClientName,
Tags: []string{"proplet", "propeller"},
Status: "enabled",
}
propletThing, err = smqsdk.CreateThing(propletThing, domain.ID, token.AccessToken)
propletClient, err = smqsdk.CreateClient(propletClient, domain.ID, token.AccessToken)
if err != nil {
return errors.Wrap(errFailedClientCreation, err)
}
Expand All @@ -178,21 +178,14 @@ var provisionCmd = &cobra.Command{
}

managerConns := smqSDK.Connection{
ThingID: managerThing.ID,
ChannelID: managerChannel.ID,
ClientIDs: []string{managerClient.ID, propletClient.ID},
ChannelIDs: []string{managerChannel.ID},
Types: []string{"publish", "subscribe"},
}
if err = smqsdk.Connect(managerConns, domain.ID, token.AccessToken); err != nil {
return errors.Wrap(errFailedConnectionCreation, err)
}

propletConns := smqSDK.Connection{
ThingID: propletThing.ID,
ChannelID: managerChannel.ID,
}
if err = smqsdk.Connect(propletConns, domain.ID, token.AccessToken); err != nil {
return errors.Wrap(errFailedConnectionCreation, err)
}

return nil
}),
),
Expand All @@ -207,27 +200,27 @@ var provisionCmd = &cobra.Command{
configContent := fmt.Sprintf(`# SuperMQ Configuration
[manager]
thing_id = "%s"
thing_key = "%s"
client_id = "%s"
client_key = "%s"
channel_id = "%s"
[proplet]
thing_id = "%s"
thing_key = "%s"
client_id = "%s"
client_key = "%s"
channel_id = "%s"
[proxy]
thing_id = "%s"
thing_key = "%s"
client_id = "%s"
client_key = "%s"
channel_id = "%s"`,
managerThing.ID,
managerThing.Credentials.Secret,
managerClient.ID,
managerClient.Credentials.Secret,
managerChannel.ID,
propletThing.ID,
propletThing.Credentials.Secret,
propletClient.ID,
propletClient.Credentials.Secret,
managerChannel.ID,
propletThing.ID,
propletThing.Credentials.Secret,
propletClient.ID,
propletClient.Credentials.Secret,
managerChannel.ID,
)

Expand Down
36 changes: 23 additions & 13 deletions cmd/cli/main.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,18 +3,19 @@ package main
import (
"log"

smqsdk "github.com/absmach/magistrala/pkg/sdk/go"
"github.com/absmach/propeller/cli"
"github.com/absmach/propeller/pkg/sdk"
smqsdk "github.com/absmach/supermq/pkg/sdk"
"github.com/spf13/cobra"
)

var (
tlsVerification = false
managerURL = "http://localhost:7070"
usersURL = "http://localhost:9002"
thingsURL = "http://localhost:9000"
domainsURL = "http://localhost:8189"
domainsURL = "http://localhost:9003"
clientsURL = "http://localhost:9006"
channelsURL = "http://localhost:9005"
msgContentType = string(smqsdk.CTJSONSenML)
)

Expand All @@ -33,8 +34,9 @@ func main() {

smqSDKConf := smqsdk.Config{
UsersURL: usersURL,
ThingsURL: thingsURL,
DomainsURL: domainsURL,
ClientsURL: clientsURL,
ChannelsURL: channelsURL,
MsgContentType: smqsdk.ContentType(msgContentType),
}

Expand Down Expand Up @@ -75,14 +77,6 @@ func main() {
"Users service URL",
)

rootCmd.PersistentFlags().StringVarP(
&thingsURL,
"things-url",
"t",
thingsURL,
"Things service URL",
)

rootCmd.PersistentFlags().StringVarP(
&domainsURL,
"domains-url",
Expand All @@ -91,10 +85,26 @@ func main() {
"Domains service URL",
)

rootCmd.PersistentFlags().StringVarP(
&clientsURL,
"clients-url",
"c",
clientsURL,
"Clients service URL",
)

rootCmd.PersistentFlags().StringVarP(
&channelsURL,
"channels-url",
"z",
channelsURL,
"Channels service URL",
)

rootCmd.PersistentFlags().StringVarP(
&msgContentType,
"content-type",
"c",
"t",
msgContentType,
"Message content type",
)
Expand Down
20 changes: 10 additions & 10 deletions cmd/manager/main.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -9,17 +9,17 @@ import (
"os"
"time"

"github.com/absmach/magistrala/pkg/jaeger"
"github.com/absmach/magistrala/pkg/prometheus"
"github.com/absmach/magistrala/pkg/server"
httpserver "github.com/absmach/magistrala/pkg/server/http"
"github.com/absmach/propeller"
"github.com/absmach/propeller/manager"
"github.com/absmach/propeller/manager/api"
"github.com/absmach/propeller/manager/middleware"
"github.com/absmach/propeller/pkg/mqtt"
"github.com/absmach/propeller/pkg/scheduler"
"github.com/absmach/propeller/pkg/storage"
"github.com/absmach/supermq/pkg/jaeger"
"github.com/absmach/supermq/pkg/prometheus"
"github.com/absmach/supermq/pkg/server"
httpserver "github.com/absmach/supermq/pkg/server/http"
"github.com/caarlos0/env/v11"
"github.com/google/uuid"
"go.opentelemetry.io/otel/trace"
Expand All @@ -41,8 +41,8 @@ type config struct {
MQTTQoS uint8 `env:"MANAGER_MQTT_QOS" envDefault:"2"`
MQTTTimeout time.Duration `env:"MANAGER_MQTT_TIMEOUT" envDefault:"30s"`
ChannelID string `env:"MANAGER_CHANNEL_ID"`
ThingID string `env:"MANAGER_THING_ID"`
ThingKey string `env:"MANAGER_THING_KEY"`
ClientID string `env:"MANAGER_CLIENT_ID"`
ClientKey string `env:"MANAGER_CLIENT_KEY"`
Server server.Config
OTELURL url.URL `env:"MANAGER_OTEL_URL"`
TraceRatio float64 `env:"MANAGER_TRACE_RATIO" envDefault:"0"`
Expand All @@ -61,16 +61,16 @@ func main() {
cfg.InstanceID = uuid.NewString()
}

if cfg.ThingID == "" || cfg.ThingKey == "" || cfg.ChannelID == "" {
if cfg.ClientID == "" || cfg.ClientKey == "" || cfg.ChannelID == "" {
_, err := os.Stat(configPath)
switch err {
case nil:
conf, err := propeller.LoadConfig(configPath)
if err != nil {
log.Fatalf("failed to load TOML configuration: %s", err.Error())
}
cfg.ThingID = conf.Manager.ThingID
cfg.ThingKey = conf.Manager.ThingKey
cfg.ClientID = conf.Manager.ClientID
cfg.ClientKey = conf.Manager.ClientKey
cfg.ChannelID = conf.Manager.ChannelID
default:
log.Fatalf("failed to load TOML configuration: %s", err.Error())
Expand Down Expand Up @@ -107,7 +107,7 @@ func main() {
}
tracer := tp.Tracer(svcName)

mqttPubSub, err := mqtt.NewPubSub(cfg.MQTTAddress, cfg.MQTTQoS, svcName, cfg.ThingID, cfg.ThingKey, cfg.ChannelID, cfg.MQTTTimeout, logger)
mqttPubSub, err := mqtt.NewPubSub(cfg.MQTTAddress, cfg.MQTTQoS, svcName, cfg.ClientID, cfg.ClientKey, cfg.ChannelID, cfg.MQTTTimeout, logger)
if err != nil {
logger.Error("failed to initialize mqtt pubsub", slog.String("error", err.Error()))

Expand Down
16 changes: 8 additions & 8 deletions cmd/proplet/main.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -8,11 +8,11 @@ import (
"os"
"time"

"github.com/absmach/magistrala/pkg/server"
"github.com/absmach/propeller"
"github.com/absmach/propeller/pkg/mqtt"
"github.com/absmach/propeller/proplet"
"github.com/absmach/propeller/proplet/runtimes"
"github.com/absmach/supermq/pkg/server"
"github.com/caarlos0/env/v11"
"github.com/google/uuid"
"golang.org/x/sync/errgroup"
Expand All @@ -31,8 +31,8 @@ type config struct {
MQTTQoS byte `env:"PROPLET_MQTT_QOS" envDefault:"2"`
LivelinessInterval time.Duration `env:"PROPLET_LIVELINESS_INTERVAL" envDefault:"10s"`
ChannelID string `env:"PROPLET_CHANNEL_ID"`
ThingID string `env:"PROPLET_THING_ID"`
ThingKey string `env:"PROPLET_THING_KEY"`
ClientID string `env:"PROPLET_CLIIENT_ID"`
ClientKey string `env:"PROPLET_CLIIENT_KEY"`
ExternalWasmRuntime string `env:"PROPLET_EXTERNAL_WASM_RUNTIME" envDefault:""`
}

Expand All @@ -49,16 +49,16 @@ func main() {
cfg.InstanceID = uuid.NewString()
}

if cfg.ThingID == "" || cfg.ThingKey == "" || cfg.ChannelID == "" {
if cfg.ClientID == "" || cfg.ClientKey == "" || cfg.ChannelID == "" {
_, err := os.Stat(configPath)
switch err {
case nil:
conf, err := propeller.LoadConfig(configPath)
if err != nil {
log.Fatalf("failed to load TOML configuration: %s", err.Error())
}
cfg.ThingID = conf.Proplet.ThingID
cfg.ThingKey = conf.Proplet.ThingKey
cfg.ClientID = conf.Proplet.ClientID
cfg.ClientKey = conf.Proplet.ClientKey
cfg.ChannelID = conf.Proplet.ChannelID
default:
log.Fatalf("failed to load TOML configuration: %s", err.Error())
Expand All @@ -75,7 +75,7 @@ func main() {
logger := slog.New(logHandler)
slog.SetDefault(logger)

mqttPubSub, err := mqtt.NewPubSub(cfg.MQTTAddress, cfg.MQTTQoS, cfg.InstanceID, cfg.ThingID, cfg.ThingKey, cfg.ChannelID, cfg.MQTTTimeout, logger)
mqttPubSub, err := mqtt.NewPubSub(cfg.MQTTAddress, cfg.MQTTQoS, cfg.InstanceID, cfg.ClientID, cfg.ClientKey, cfg.ChannelID, cfg.MQTTTimeout, logger)
if err != nil {
logger.Error("failed to initialize mqtt client", slog.Any("error", err))

Expand All @@ -90,7 +90,7 @@ func main() {
runtime = runtimes.NewWazeroRuntime(logger, mqttPubSub, cfg.ChannelID)
}

service, err := proplet.NewService(ctx, cfg.ChannelID, cfg.ThingID, cfg.ThingKey, cfg.LivelinessInterval, mqttPubSub, logger, runtime)
service, err := proplet.NewService(ctx, cfg.ChannelID, cfg.ClientID, cfg.ClientKey, cfg.LivelinessInterval, mqttPubSub, logger, runtime)
if err != nil {
logger.Error("failed to initialize service", slog.Any("error", err))

Expand Down
Loading

0 comments on commit 585258b

Please sign in to comment.