-
Notifications
You must be signed in to change notification settings - Fork 16
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
19 changed files
with
197 additions
and
456 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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,68 @@ | ||
| package sdk | ||
|
|
||
| import ( | ||
| "errors" | ||
| "fmt" | ||
| "log" | ||
|
|
||
| "github.com/nats-io/nats.go" | ||
| ) | ||
|
|
||
| type NATSClient struct { | ||
| conn *nats.Conn | ||
| js nats.JetStreamContext | ||
| config natsConfig | ||
| } | ||
|
|
||
| func NewNATSClient() (*NATSClient, error) { | ||
| config, err := loadNatsConfig() | ||
| if err != nil { | ||
| return nil, errors.New("Unable to load the nats configurations , error :" + err.Error()) | ||
| } | ||
| options := []nats.Option{} | ||
| if config.EnableToken { | ||
| options = append(options, nats.Token(config.NatsToken)) | ||
| } | ||
| if config.MtlsConfig.IsEnabled { | ||
| tlsConfig, err := createTLSConfig(config.MtlsConfig) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
| options = append(options, nats.Secure(tlsConfig)) | ||
| } | ||
| conn, err := nats.Connect(config.NatsAddress, options...) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| js, err := conn.JetStream() | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| return &NATSClient{conn: conn, js: js, config: *config}, nil | ||
| } | ||
|
|
||
| func (natsCli *NATSClient) CreateStream(streamName string) error { | ||
| stream, err := natsCli.js.StreamInfo(streamName) | ||
| log.Printf("Retrieved stream %s", fmt.Sprintf("%v", stream)) | ||
| if err != nil { | ||
| log.Printf("Error getting stream %s", err) | ||
| } | ||
| if stream == nil { | ||
| log.Printf("creating stream %q and subjects %q", streamName, streamName+".*") | ||
| _, err = natsCli.js.AddStream(&nats.StreamConfig{ | ||
| Name: streamName, | ||
| Subjects: []string{streamName + ".*"}, | ||
| }) | ||
| if err != nil { | ||
| return err | ||
| } | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (natsCli *NATSClient) Publish(subject string, data []byte) error { | ||
| _, err := natsCli.js.Publish(subject, data) | ||
| 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,28 @@ | ||
| package sdk | ||
|
|
||
| import ( | ||
| "github.com/kelseyhightower/envconfig" | ||
| "github.com/pkg/errors" | ||
| ) | ||
|
|
||
| type natsConfig struct { | ||
| NatsAddress string `envconfig:"NATS_ADDRESS"` | ||
| NatsToken string `envconfig:"NATS_TOKEN"` | ||
| MtlsConfig mtlsConfig | ||
| EnableToken bool `envconfig:"ENABLE_TOKEN"` | ||
| } | ||
|
|
||
| type mtlsConfig struct { | ||
| CertificateFilePath string `envconfig:"CERT_FILE" default:""` | ||
| KeyFilePath string `envconfig:"KEY_FILE" default:""` | ||
| CAFilePath string `envconfig:"CA_FILE" default:""` | ||
| IsEnabled bool `envconfig:"ENABLE_MTLS_NATS" default:"false"` | ||
| } | ||
|
|
||
| func loadNatsConfig() (*natsConfig, error) { | ||
| natsConf := &natsConfig{} | ||
| if err := envconfig.Process("", natsConf); err != nil { | ||
| return nil, errors.WithStack(err) | ||
| } | ||
| return natsConf, nil | ||
| } |
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,77 @@ | ||
| package sdk | ||
|
|
||
| import ( | ||
| "crypto/tls" | ||
| "crypto/x509" | ||
| "errors" | ||
| "fmt" | ||
| "io" | ||
| "os" | ||
| ) | ||
|
|
||
| func createTLSConfig(config mtlsConfig) (*tls.Config, error) { | ||
| certPEM, keyPEM, CACertPEM, err := readMtlsCerts(config.CertificateFilePath, config.KeyFilePath, config.CAFilePath) | ||
| if err != nil { | ||
| return nil, errors.New("unable to read the mtls certificates error:" + err.Error()) | ||
| } | ||
| cert, err := tls.X509KeyPair(certPEM, keyPEM) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("error loading X509 key pair from PEM: %w", err) | ||
| } | ||
|
|
||
| caCertPool := x509.NewCertPool() | ||
| caCertPool.AppendCertsFromPEM(CACertPEM) | ||
| tlsConfig := &tls.Config{ | ||
| Certificates: []tls.Certificate{cert}, | ||
| RootCAs: caCertPool, | ||
| InsecureSkipVerify: false, | ||
| } | ||
| return tlsConfig, nil | ||
| } | ||
|
|
||
| func readMtlsCerts(certificateFilePath, keyFilePath, CAFilePath string) (certPEM, keyPEM, CACertPEM []byte, err error) { | ||
| certPEM, err = readMtlsFileContents(certificateFilePath) | ||
| if err != nil { | ||
| err = fmt.Errorf("error while reading cert file: %w", err) | ||
| return | ||
| } | ||
|
|
||
| keyPEM, err = readMtlsFileContents(keyFilePath) | ||
| if err != nil { | ||
| err = fmt.Errorf("error while reading key file: %w", err) | ||
| return | ||
| } | ||
|
|
||
| CACertPEM, err = readMtlsFileContents(CAFilePath) | ||
| if err != nil { | ||
| err = fmt.Errorf("error while reading CAcert file: %w", err) | ||
| return | ||
| } | ||
|
|
||
| return | ||
|
|
||
| } | ||
|
|
||
| func openMtlsCertFile(filepath string) (f *os.File, err error) { | ||
| f, err = os.Open(filepath) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("failed to open mtls certificate file: %w", err) | ||
| } | ||
| return f, nil | ||
| } | ||
|
|
||
| func readMtlsFileContents(filePath string) ([]byte, error) { | ||
| file, err := openMtlsCertFile(filePath) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| defer file.Close() | ||
|
|
||
| contents, err := io.ReadAll(file) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("error while reading file %s:%w", filePath, err) | ||
| } | ||
|
|
||
| return contents, nil | ||
| } |
Oops, something went wrong.