- add support for TLS certificate

- fix return error on transport create

grpc.go

@@ -2,6 +2,7 @@ package kubemq
 
 import (
 	"context"
+	"crypto/x509"
 	"fmt"
 	"io"
 	"time"
@@ -23,11 +24,25 @@ type gRPCTransport struct {
 func newGRPCTransport(ctx context.Context, opts *Options) (Transport, *ServerInfo, error) {
 	var connOptions []grpc.DialOption
 	if opts.isSecured {
-		creds, err := credentials.NewClientTLSFromFile(opts.certFile, opts.serverOverrideDomain)
-		if err != nil {
-			return nil, nil, fmt.Errorf("could not load tls cert: %s", err)
+		if opts.certFile != "" {
+			creds, err := credentials.NewClientTLSFromFile(opts.certFile, opts.serverOverrideDomain)
+			if err != nil {
+				return nil, nil, fmt.Errorf("could not load tls cert: %s", err)
+			}
+			connOptions = append(connOptions, grpc.WithTransportCredentials(creds))
+		} else if opts.certData != "" {
+			certPool := x509.NewCertPool()
+			if !certPool.AppendCertsFromPEM([]byte(opts.certData)) {
+				return nil, nil, fmt.Errorf("credentials: failed to append certificates to pool")
+			}
+			creds := credentials.NewClientTLSFromCert(certPool, opts.serverOverrideDomain)
+
+			connOptions = append(connOptions, grpc.WithTransportCredentials(creds))
+
+		} else {
+			return nil, nil, fmt.Errorf("no valid tls security provided")
 		}
-		connOptions = append(connOptions, grpc.WithTransportCredentials(creds))
+
 	} else {
 		connOptions = append(connOptions, grpc.WithInsecure())
 	}
@@ -56,7 +71,7 @@ func newGRPCTransport(ctx context.Context, opts *Options) (Transport, *ServerInf
 
 	si, err := g.Ping(ctx)
 	if err != nil {
-		return nil, &ServerInfo{}, nil
+		return nil, &ServerInfo{}, err
 	}
 
 	return g, si, nil

options.go

@@ -23,6 +23,7 @@ type Options struct {
 	port                 int
 	isSecured            bool
 	certFile             string
+	certData             string
 	serverOverrideDomain string
 	token                string
 	clientId             string
@@ -76,6 +77,17 @@ func WithCredentials(certFile, serverOverrideDomain string) Option {
 	})
 }
 
+// WithCertificate - set secured TLS credentials from the input certificate data for client.
+// serverNameOverride is for testing only. If set to a non empty string,
+// it will override the virtual host name of authority (e.g. :authority header field) in requests.
+func WithCertificate(certData, serverOverrideDomain string) Option {
+	return newFuncOption(func(o *Options) {
+		o.isSecured = true
+		o.certData = certData
+		o.serverOverrideDomain = serverOverrideDomain
+	})
+}
+
 // WithToken - set KubeMQ token to be used for KubeMQ connection - not mandatory, only if enforced by the KubeMQ server
 func WithToken(token string) Option {
 	return newFuncOption(func(o *Options) {
@@ -98,7 +110,7 @@ func WithReceiveBufferSize(size int) Option {
 }
 
 // WithDefaultChannel - set default channel for any outbound requests
-func WithDefualtChannel(channel string) Option {
+func WithDefaultChannel(channel string) Option {
 	return newFuncOption(func(o *Options) {
 		o.defaultChannel = channel
 	})
@@ -139,18 +151,18 @@ func GetDefaultOptions() *Options {
 func (o *Options) Validate() error {
 	switch o.transportType {
 	case TransportTypeGRPC:
-		if o.host== "" {
+		if o.host == "" {
 			return errors.New("invalid host")
 		}
-		if o.port<=0 {
+		if o.port <= 0 {
 			return errors.New("invalid port")
 		}
 	case TransportTypeRest:
-		if o.restUri== "" {
+		if o.restUri == "" {
 			return errors.New("invalid address uri")
 		}
 	default:
 		return errors.New("no transport type was set")
 	}
 	return nil
-}
+}

rest.go

@@ -132,7 +132,7 @@ func newRestTransport(ctx context.Context, opts *Options) (Transport, *ServerInf
 	}
 	si, err := rt.Ping(ctx)
 	if err != nil {
-		return nil, &ServerInfo{}, nil
+		return nil, &ServerInfo{}, err
 	}
 	return rt, si, nil
 }