-
Notifications
You must be signed in to change notification settings - Fork 26
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #199 from OS-Mind/main
Add WAMP Producer
- Loading branch information
Showing
9 changed files
with
192 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,10 @@ | ||
| { | ||
| "wamp_uri": "ws://localhost:9009/ws", | ||
| "username": "admin", | ||
| "password": "password", | ||
| "realm": "realm1", | ||
| "topic": "example.hello", | ||
| "serType": "json", | ||
| "compress": true, | ||
| "authid": "clientJR" | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,100 @@ | ||
| package wamp | ||
|
|
||
| import ( | ||
| "context" | ||
| "encoding/json" | ||
| "os" | ||
|
|
||
| "github.com/gammazero/nexus/v3/client" | ||
| "github.com/gammazero/nexus/v3/wamp" | ||
| "github.com/rs/zerolog/log" | ||
| ) | ||
|
|
||
| type Config struct { | ||
| WampURI string `json:"wamp_uri"` | ||
| Username string `json:"username"` | ||
| Password string `json:"password"` | ||
| Realm string `json:"realm"` | ||
| Topic string `json:"topic"` | ||
| SerType string `json:"serType"` | ||
| Compress bool `json:"compress"` | ||
| Authid string `json:"authid"` | ||
| } | ||
|
|
||
| type Producer struct { | ||
| client client.Client | ||
| realm string | ||
| topic string | ||
| authid string | ||
| } | ||
|
|
||
| func (p *Producer) Initialize(ctx context.Context, configFile string) { | ||
| var config Config | ||
| file, err := os.ReadFile(configFile) | ||
| if err != nil { | ||
| log.Fatal().Err(err).Msg("Failed to read configuration file") | ||
| } | ||
| err = json.Unmarshal(file, &config) | ||
| if err != nil { | ||
| log.Fatal().Err(err).Msg("Failed to parse configuration parameters") | ||
| } | ||
| var wampclient *client.Client | ||
|
|
||
| // Get requested serialization. | ||
| serialization := client.JSON | ||
| switch config.SerType { | ||
| case "json": | ||
| case "msgpack": | ||
| serialization = client.MSGPACK | ||
| case "cbor": | ||
| serialization = client.CBOR | ||
| default: | ||
| log.Fatal().Err(err).Msg("Invalid serialization, muse be one of: json, msgpack, cbor") | ||
| } | ||
|
|
||
| cfg := client.Config{ | ||
| Realm: config.Realm, | ||
| Serialization: serialization, | ||
| HelloDetails: wamp.Dict{ | ||
| "authid": config.Authid, | ||
| }, | ||
| } | ||
|
|
||
| if config.Compress { | ||
| cfg.WsCfg.EnableCompression = true | ||
| } | ||
|
|
||
| addr := config.WampURI | ||
|
|
||
| wampclient, err = client.ConnectNet(context.Background(), addr, cfg) | ||
| if err != nil { | ||
| log.Fatal().Err(err).Msg("Can't connect to WAMP Router") | ||
| } | ||
| // defer wampclient.Close() | ||
|
|
||
| p.realm = config.Realm | ||
| p.topic = config.Topic | ||
| p.authid = config.Authid | ||
|
|
||
| p.client = *wampclient | ||
| } | ||
|
|
||
| func (p *Producer) Produce(ctx context.Context, k []byte, v []byte, _ any) { | ||
| data := string(v) | ||
| args := wamp.List{data} | ||
| opts := wamp.Dict{ | ||
| "authid": p.authid, | ||
| } | ||
| err := p.client.Publish(p.topic, opts, args, nil) | ||
| if err != nil { | ||
| log.Fatal().Err(err).Msgf("publish error: %s", err) | ||
| } | ||
| } | ||
|
|
||
| func (p *Producer) Close(ctx context.Context) error { | ||
| err := p.client.Close() | ||
| if err != nil { | ||
| log.Warn().Err(err).Msg("Failed to close WAMP connection") | ||
| } | ||
| return err | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,58 @@ | ||
| //go:build exclude | ||
|
|
||
| package wamp | ||
|
|
||
| import ( | ||
| "context" | ||
| "testing" | ||
|
|
||
| "github.com/jrnd-io/jr/pkg/producers/wamp" | ||
| ) | ||
|
|
||
| func TestProducer_Initialize(t *testing.T) { | ||
| configFile := "config.json.example" | ||
|
|
||
| producer, err := wamp.ProducerFactory("wamp") | ||
| if err != nil { | ||
| t.Fatalf("Error reading configuration file: %v", err) | ||
| } | ||
| err = producer.Initialize(configFile) | ||
| if err != nil { | ||
| t.Fatalf("Error reading configuration file: %v", err) | ||
| } | ||
| } | ||
|
|
||
| func TestProducer_Close(t *testing.T) { | ||
| configFile := "config.json.example" | ||
|
|
||
| producer, err := wamp.ProducerFactory("wamp") | ||
| if err != nil { | ||
| t.Fatalf("Error reading configuration file: %v", err) | ||
| } | ||
| err = producer.Initialize(configFile) | ||
| if err != nil { | ||
| t.Fatalf("Error reading configuration file: %v", err) | ||
| } | ||
|
|
||
| producer.Close() | ||
| } | ||
|
|
||
| func TestProducer_Produce(t *testing.T) { | ||
| configFile := "config.json.example" | ||
|
|
||
| producer, err := wamp.ProducerFactory("wamp") | ||
| if err != nil { | ||
| t.Fatalf("Error reading configuration file: %v", err) | ||
| } | ||
| err = producer.Initialize(configFile) | ||
| if err != nil { | ||
| t.Fatalf("Error initializing producer: %v", err) | ||
| } | ||
|
|
||
| ctx := context.Background() | ||
| key := "loo" | ||
| val := "foo" | ||
| exp := 0 | ||
| producer.Produce(ctx, key, val, exp) | ||
|
|
||
| } |