Bug/srivastavasujay/issue#677

broker/nats/nats.go

@@ -1,6 +1,7 @@
 package nats
 
 import (
+	"encoding/json"
 	"log"
 	"strings"
 	"sync"
@@ -25,7 +26,7 @@ type Options struct {
 
 // Nats will implement Nats subscribe and publish functionality
 type Nats struct {
-	ec *nats.EncodedConn
+	ec *nats.Conn
 	wg *sync.WaitGroup
 }
 
@@ -57,25 +58,21 @@ func New(opts Options) (broker.Handler, error) {
 		return nil, ErrConnect(err)
 	}
 
-	ec, err := nats.NewEncodedConn(nc, nats.JSON_ENCODER)
-	if err != nil {
-		return nil, ErrEncodedConn(err)
-	}
-
-	return &Nats{ec: ec}, nil
+	return &Nats{ec: nc, wg: &sync.WaitGroup{}}, nil
 }
+
 func (n *Nats) ConnectedEndpoints() (endpoints []string) {
-	for _, server := range n.ec.Conn.Servers() {
+	for _, server := range n.ec.Servers() {
 		endpoints = append(endpoints, strings.TrimPrefix(server, "nats://"))
 	}
 	return
 }
 
 func (n *Nats) Info() string {
-	if n.ec == nil || n.ec.Conn == nil {
+	if n.ec == nil {
 		return broker.NotConnected
 	}
-	return n.ec.Conn.Opts.Name
+	return n.ec.Opts.Name
 }
 
 func (n *Nats) CloseConnection() {
@@ -84,7 +81,11 @@ func (n *Nats) CloseConnection() {
 
 // Publish - to publish messages
 func (n *Nats) Publish(subject string, message *broker.Message) error {
-	err := n.ec.Publish(subject, message)
+	data, err := json.Marshal(message)
+	if err != nil {
+		return err
+	}
+	err = n.ec.Publish(subject, data)
 	if err != nil {
 		return ErrPublish(err)
 	}
@@ -93,20 +94,26 @@ func (n *Nats) Publish(subject string, message *broker.Message) error {
 
 // PublishWithChannel - to publish messages with channel
 func (n *Nats) PublishWithChannel(subject string, msgch chan *broker.Message) error {
-	err := n.ec.BindSendChan(subject, msgch)
-	if err != nil {
-		return ErrPublish(err)
+	for msg := range msgch {
+		data, err := json.Marshal(msg)
+		if err != nil {
+			return err
+		}
+		err = n.ec.Publish(subject, data)
+		if err != nil {
+			return ErrPublish(err)
+		}
 	}
 	return nil
 }
 
 // Subscribe - for subscribing messages
 // TODO Ques: Do we want to unsubscribe
-// TODO will the method-user just subsribe, how will it handle the received messages?
+// TODO will the method-user just subscribe, how will it handle the received messages?
 func (n *Nats) Subscribe(subject, queue string, message []byte) error {
 	n.wg.Add(1)
 	_, err := n.ec.QueueSubscribe(subject, queue, func(msg *nats.Msg) {
-		message = msg.Data
+		copy(message, msg.Data)
 		n.wg.Done()
 	})
 	if err != nil {
@@ -119,7 +126,15 @@ func (n *Nats) Subscribe(subject, queue string, message []byte) error {
 
 // SubscribeWithChannel will publish all the messages received to the given channel
 func (n *Nats) SubscribeWithChannel(subject, queue string, msgch chan *broker.Message) error {
-	_, err := n.ec.BindRecvQueueChan(subject, queue, msgch)
+	_, err := n.ec.QueueSubscribe(subject, queue, func(msg *nats.Msg) {
+		var message broker.Message
+		err := json.Unmarshal(msg.Data, &message)
+		if err != nil {
+			log.Printf("Error unmarshalling message: %v", err)
+			return
+		}
+		msgch <- &message
+	})
 	if err != nil {
 		return ErrQueueSubscribe(err)
 	}

models/meshmodel/core/policies/rego_policy_relationship.go

@@ -9,10 +9,10 @@ import (
 	"github.com/layer5io/meshkit/models/meshmodel/registry/v1alpha3"
 	"github.com/layer5io/meshkit/utils"
 	"github.com/meshery/schemas/models/v1beta1/pattern"
-	"github.com/open-policy-agent/opa/rego"
-	"github.com/open-policy-agent/opa/storage"
-	"github.com/open-policy-agent/opa/storage/inmem"
-	"github.com/open-policy-agent/opa/topdown/print"
+	"github.com/open-policy-agent/opa/v1/rego"
+	"github.com/open-policy-agent/opa/v1/storage"
+	"github.com/open-policy-agent/opa/v1/storage/inmem"
+	"github.com/open-policy-agent/opa/v1/topdown/print"
 	"github.com/sirupsen/logrus"
 )
 
@@ -117,5 +117,4 @@ func (r *Rego) RegoPolicyHandler(designFile pattern.PatternFile, regoQueryString
 	}
 
 	return evaluationResponse, nil
-
 }

models/oci/oci.go

@@ -147,7 +147,7 @@ func PushToOCIRegistry(dirPath, registryAdd, repositoryAdd, imageTag, username,
 	opts := oras.PackManifestOptions{
 		Layers: fileDescriptors,
 	}
-	manifestDescriptor, packageErr := oras.PackManifest(ctx, fs, oras.PackManifestVersion1_1_RC4, artifactType, opts)
+	manifestDescriptor, packageErr := oras.PackManifest(ctx, fs, oras.PackManifestVersion1_1, artifactType, opts)
 	if packageErr != nil {
 		return ErrGettingLayer(packageErr)
 	}