Repository: Tochemey/goakt
Pull request #139 (2023-10-22 19:34:21)

feat: add NATS discovery provider

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/Tochemey/goakt/pull/139?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) Report
Attention: `5 lines` in your changes are missing coverage. Please review.
> Comparison is base [(`6595ebd`)](https://app.codecov.io/gh/Tochemey/goakt/commit/6595ebd6cc8d647f908e8f81716e4e40187ce924?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) 90.48% compared to head [(`b7943db`)](https://app.codecov.io/gh/Tochemey/goakt/pull/139?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) 90.25%.

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##             main     #139      +/-   ##
==========================================
- Coverage   90.48%   90.25%   -0.23%     
==========================================
  Files          37       37              
  Lines        3322     3326       +4     
==========================================
- Hits         3006     3002       -4     
- Misses        249      257       +8     
  Partials       67       67              
```


| [Files](https://app.codecov.io/gh/Tochemey/goakt/pull/139?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene) | Coverage Δ | |
|---|---|---|
| [cluster/cluster.go](https://app.codecov.io/gh/Tochemey/goakt/pull/139?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene#diff-Y2x1c3Rlci9jbHVzdGVyLmdv) | `76.54% <100.00%> (ø)` | |
| [cluster/discovery.go](https://app.codecov.io/gh/Tochemey/goakt/pull/139?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene#diff-Y2x1c3Rlci9kaXNjb3ZlcnkuZ28=) | `94.73% <100.00%> (ø)` | |
| [discovery/node.go](https://app.codecov.io/gh/Tochemey/goakt/pull/139?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene#diff-ZGlzY292ZXJ5L25vZGUuZ28=) | `100.00% <100.00%> (ø)` | |
| [discovery/kubernetes/discovery.go](https://app.codecov.io/gh/Tochemey/goakt/pull/139?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene#diff-ZGlzY292ZXJ5L2t1YmVybmV0ZXMvZGlzY292ZXJ5Lmdv) | `66.89% <0.00%> (-0.92%)` | :arrow_down: |
| [discovery/mdns/discovery.go](https://app.codecov.io/gh/Tochemey/goakt/pull/139?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene#diff-ZGlzY292ZXJ5L21kbnMvZGlzY292ZXJ5Lmdv) | `78.57% <76.92%> (-0.88%)` | :arrow_down: |

... and [1 file with indirect coverage changes](https://app.codecov.io/gh/Tochemey/goakt/pull/139/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene)


</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/Tochemey/goakt/pull/139?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=Arsene).


Patch:
diff --git a/cluster/cluster.go b/cluster/cluster.go
index d52ff60..9714a9e 100644
--- a/cluster/cluster.go
+++ b/cluster/cluster.go
@@ -79,7 +79,7 @@ type Cluster struct {
 	kvStore olric.DMap
 
 	// specifies the Cluster host
-	host *node
+	host *discovery.Node
 
 	// specifies the hasher
 	hasher hash.Hasher
@@ -117,7 +117,7 @@ func New(name string, serviceDiscovery *discovery.ServiceDiscovery, opts ...Opti
 	}
 
 	// get the host info
-	hostNode, err := hostNode()
+	hostNode, err := discovery.HostNode()
 	// handle the error
 	if err != nil {
 		cl.logger.Error(errors.Wrap(err, "failed get the host node.💥"))
diff --git a/cluster/discovery.go b/cluster/discovery.go
index ed942d7..4f575b4 100644
--- a/cluster/discovery.go
+++ b/cluster/discovery.go
@@ -131,5 +131,5 @@ func (d *discoveryProvider) DiscoverPeers() ([]string, error) {
 
 // Close implementation
 func (d *discoveryProvider) Close() error {
-	return nil
+	return d.provider.Close()
 }
diff --git a/discovery/errors.go b/discovery/errors.go
index e2f2353..6c29e9a 100644
--- a/discovery/errors.go
+++ b/discovery/errors.go
@@ -33,4 +33,6 @@ var (
 	ErrNotInitialized = errors.New("provider not initialized")
 	// ErrAlreadyRegistered is used when attempting to re-register the provider
 	ErrAlreadyRegistered = errors.New("provider already registered")
+	// ErrNotRegistered is used when attempting to de-register the provider
+	ErrNotRegistered = errors.New("provider is not registered")
 )
diff --git a/discovery/kubernetes/discovery.go b/discovery/kubernetes/discovery.go
index 5c92e56..d12a0ec 100644
--- a/discovery/kubernetes/discovery.go
+++ b/discovery/kubernetes/discovery.go
@@ -51,8 +51,8 @@ const (
 	RemotingPortName        = "remoting-port"
 )
 
-// option represents the kubernetes provider option
-type option struct {
+// discoConfig represents the kubernetes provider discoConfig
+type discoConfig struct {
 	// Provider specifies the provider name
 	Provider string
 	// NameSpace specifies the namespace
@@ -65,13 +65,13 @@ type option struct {
 
 // Discovery represents the kubernetes discovery
 type Discovery struct {
-	option *option
+	option *discoConfig
 	client kubernetes.Interface
 	mu     sync.Mutex
 
 	stopChan chan struct{}
 	// states whether the actor system has started or not
-	isInitialized *atomic.Bool
+	initialized *atomic.Bool
 }
 
 // enforce compilation error
@@ -80,14 +80,14 @@ var _ discovery.Provider = &Discovery{}
 // NewDiscovery returns an instance of the kubernetes discovery provider
 func NewDiscovery() *Discovery {
 	// create an instance of
-	k8 := &Discovery{
-		mu:            sync.Mutex{},
-		stopChan:      make(chan struct{}, 1),
-		isInitialized: atomic.NewBool(false),
-		option:        &option{},
+	discovery := &Discovery{
+		mu:          sync.Mutex{},
+		stopChan:    make(chan struct{}, 1),
+		initialized: atomic.NewBool(false),
+		option:      &discoConfig{},
 	}
 
-	return k8
+	return discovery
 }
 
 // ID returns the discovery provider id
@@ -102,7 +102,7 @@ func (d *Discovery) Initialize() error {
 	// release the lock
 	defer d.mu.Unlock()
 	// first check whether the discovery provider is running
-	if d.isInitialized.Load() {
+	if d.initialized.Load() {
 		return discovery.ErrAlreadyInitialized
 	}
 
@@ -122,7 +122,7 @@ func (d *Discovery) SetConfig(meta discovery.Config) error {
 	defer d.mu.Unlock()
 
 	// first check whether the discovery provider is running
-	if d.isInitialized.Load() {
+	if d.initialized.Load() {
 		return discovery.ErrAlreadyInitialized
 	}
 
@@ -138,7 +138,7 @@ func (d *Discovery) Register() error {
 
 	// first check whether the discovery provider has started
 	// avoid to re-register the discovery
-	if d.isInitialized.Load() {
+	if d.initialized.Load() {
 		return discovery.ErrAlreadyRegistered
 	}
 
@@ -157,7 +157,7 @@ func (d *Discovery) Register() error {
 	// set the k8 client
 	d.client = client
 	// set initialized
-	d.isInitialized = atomic.NewBool(true)
+	d.initialized = atomic.NewBool(true)
 	return nil
 }
 
@@ -169,11 +169,11 @@ func (d *Discovery) Deregister() error {
 	defer d.mu.Unlock()
 
 	// first check whether the discovery provider has started
-	if !d.isInitialized.Load() {
+	if !d.initialized.Load() {
 		return discovery.ErrNotInitialized
 	}
 	// set the initialized to false
-	d.isInitialized = atomic.NewBool(false)
+	d.initialized = atomic.NewBool(false)
 	// stop the watchers
 	close(d.stopChan)
 	// return
@@ -183,7 +183,7 @@ func (d *Discovery) Deregister() error {
 // DiscoverPeers returns a list of known nodes.
 func (d *Discovery) DiscoverPeers() ([]string, error) {
 	// first check whether the discovery provider is running
-	if !d.isInitialized.Load() {
+	if !d.initialized.Load() {
 		return nil, discovery.ErrNotInitialized
 	}
 
@@ -249,10 +249,15 @@ MainLoop:
 	return addresses.ToSlice(), nil
 }
 
-// setConfig sets the kubernetes option
+// Close closes the provider
+func (d *Discovery) Close() error {
+	return nil
+}
+
+// setConfig sets the kubernetes discoConfig
 func (d *Discovery) setConfig(config discovery.Config) (err error) {
 	// create an instance of option
-	option := new(option)
+	option := new(discoConfig)
 	// extract the namespace
 	option.NameSpace, err = config.GetString(Namespace)
 	// handle the error in case the namespace value is not properly set
diff --git a/discovery/mdns/discovery.go b/discovery/mdns/discovery.go
index e3f2daa..0597c5c 100644
--- a/discovery/mdns/discovery.go
+++ b/discovery/mdns/discovery.go
@@ -46,8 +46,8 @@ const (
 	IPv6        = "ipv6"
 )
 
-// option represents the mDNS provider option
-type option struct {
+// discoConfig represents the mDNS provider discoConfig
+type discoConfig struct {
 	// Provider specifies the provider name
 	Provider string
 	// Service specifies the service name
@@ -64,12 +64,12 @@ type option struct {
 
 // Discovery defines the mDNS discovery provider
 type Discovery struct {
-	option *option
+	option *discoConfig
 	mu     sync.Mutex
 
 	stopChan chan struct{}
-	// states whether the actor system has started or not
-	isInitialized *atomic.Bool
+
+	initialized *atomic.Bool
 
 	// resolver is used to browse for service discovery
 	resolver *zeroconf.Resolver
@@ -84,10 +84,10 @@ var _ discovery.Provider = &Discovery{}
 func NewDiscovery() *Discovery {
 	// create an instance of
 	d := &Discovery{
-		mu:            sync.Mutex{},
-		stopChan:      make(chan struct{}, 1),
-		isInitialized: atomic.NewBool(false),
-		option:        &option{},
+		mu:          sync.Mutex{},
+		stopChan:    make(chan struct{}, 1),
+		initialized: atomic.NewBool(false),
+		option:      &discoConfig{},
 	}
 
 	return d
@@ -105,7 +105,7 @@ func (d *Discovery) Initialize() error {
 	// release the lock
 	defer d.mu.Unlock()
 	// first check whether the discovery provider is running
-	if d.isInitialized.Load() {
+	if d.initialized.Load() {
 		return discovery.ErrAlreadyInitialized
 	}
 
@@ -126,7 +126,7 @@ func (d *Discovery) Register() error {
 
 	// first check whether the discovery provider has started
 	// avoid to re-register the discovery
-	if d.isInitialized.Load() {
+	if d.initialized.Load() {
 		return discovery.ErrAlreadyRegistered
 	}
 
@@ -150,7 +150,7 @@ func (d *Discovery) Register() error {
 	d.server = srv
 
 	// set initialized
-	d.isInitialized = atomic.NewBool(true)
+	d.initialized = atomic.NewBool(true)
 	return nil
 }
 
@@ -162,12 +162,12 @@ func (d *Discovery) Deregister() error {
 	defer d.mu.Unlock()
 
 	// first check whether the discovery provider has started
-	if !d.isInitialized.Load() {
+	if !d.initialized.Load() {
 		return discovery.ErrNotInitialized
 	}
 
 	// set the initialized to false
-	d.isInitialized = atomic.NewBool(false)
+	d.initialized = atomic.NewBool(false)
 
 	// shutdown the registered service
 	if d.server != nil {
@@ -180,6 +180,11 @@ func (d *Discovery) Deregister() error {
 	return nil
 }
 
+// Close closes the provider
+func (d *Discovery) Close() error {
+	return nil
+}
+
 // SetConfig registers the underlying discovery configuration
 func (d *Discovery) SetConfig(config discovery.Config) error {
 	// acquire the lock
@@ -188,7 +193,7 @@ func (d *Discovery) SetConfig(config discovery.Config) error {
 	defer d.mu.Unlock()
 
 	// first check whether the discovery provider is running
-	if d.isInitialized.Load() {
+	if d.initialized.Load() {
 		return discovery.ErrAlreadyInitialized
 	}
 
@@ -231,7 +236,7 @@ func (d *Discovery) SetConfig(config discovery.Config) error {
 // DiscoverPeers returns a list of known nodes.
 func (d *Discovery) DiscoverPeers() ([]string, error) {
 	// first check whether the discovery provider is running
-	if !d.isInitialized.Load() {
+	if !d.initialized.Load() {
 		return nil, discovery.ErrNotInitialized
 	}
 	// init entries channel
@@ -277,10 +282,10 @@ func (d *Discovery) DiscoverPeers() ([]string, error) {
 	return addresses.ToSlice(), nil
 }
 
-// setOptions sets the kubernetes option
+// setOptions sets the kubernetes discoConfig
 func (d *Discovery) setOptions(config discovery.Config) (err error) {
 	// create an instance of Option
-	option := new(option)
+	option := new(discoConfig)
 	// extract the service name
 	option.ServiceName, err = config.GetString(ServiceName)
 	// handle the error in case the service instance value is not properly set
diff --git a/discovery/nats/discovery.go b/discovery/nats/discovery.go
new file mode 100644
index 0000000..6d2c345
--- /dev/null
+++ b/discovery/nats/discovery.go
@@ -0,0 +1,418 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2022-2023 Tochemey
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package nats
+
+import (
+	"net"
+	"strconv"
+	"sync"
+	"time"
+
+	"github.com/flowchartsman/retry"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/encoders/protobuf"
+	"github.com/tochemey/goakt/discovery"
+	internalpb "github.com/tochemey/goakt/internal/v1"
+	"github.com/tochemey/goakt/log"
+	"go.uber.org/atomic"
+)
+
+const (
+	NatsServer      string = "nats-server"       // NatsServer specifies the Nats server Address
+	NatsSubject     string = "nats-subject"      // NatsSubject specifies the NATs subject
+	ActorSystemName        = "actor_system_name" // ActorSystemName specifies the actor system name
+	ApplicationName        = "app_name"          // ApplicationName specifies the application name. This often matches the actor system name
+	Timeout                = "timeout"           // Timeout specifies the discovery timeout. The default value is 1 second
+)
+
+// discoConfig represents the nats provider discoConfig
+type discoConfig struct {
+	// NatsServer defines the nats server
+	// nats://host:port of a nats server
+	NatsServer string
+	// NatsSubject defines the custom NATS subject
+	NatsSubject string
+	// The actor system name
+	ActorSystemName string
+	// ApplicationName specifies the running application
+	ApplicationName string
+	// Timeout defines the nodes discovery timeout
+	Timeout time.Duration
+}
+
+// Discovery represents the kubernetes discovery
+type Discovery struct {
+	config *discoConfig
+	mu     sync.Mutex
+
+	initialized *atomic.Bool
+	registered  *atomic.Bool
+
+	// define the nats connection
+	natsConnection *nats.EncodedConn
+	// define a slice of subscriptions
+	subscriptions []*nats.Subscription
+
+	// defines the host node
+	hostNode *discovery.Node
+
+	// define a logger
+	logger log.Logger
+}
+
+// enforce compilation error
+var _ discovery.Provider = &Discovery{}
+
+// NewDiscovery returns an instance of the kubernetes discovery provider
+func NewDiscovery(opts ...Option) *Discovery {
+	// create an instance of
+	discovery := &Discovery{
+		mu:          sync.Mutex{},
+		initialized: atomic.NewBool(false),
+		registered:  atomic.NewBool(false),
+		config:      &discoConfig{},
+		logger:      log.DefaultLogger,
+	}
+
+	// apply the various options
+	for _, opt := range opts {
+		opt.Apply(discovery)
+	}
+
+	return discovery
+}
+
+// ID returns the discovery provider id
+func (d *Discovery) ID() string {
+	return "nats"
+}
+
+// Initialize initializes the plugin: registers some internal data structures, clients etc.
+func (d *Discovery) Initialize() error {
+	// acquire the lock
+	d.mu.Lock()
+	// release the lock
+	defer d.mu.Unlock()
+
+	// first check whether the discovery provider is running
+	if d.initialized.Load() {
+		return discovery.ErrAlreadyInitialized
+	}
+
+	// set the default discovery timeout
+	if d.config.Timeout <= 0 {
+		d.config.Timeout = time.Second
+	}
+
+	// grab the host node
+	hostNode, err := discovery.HostNode()
+	// handle the error
+	if err != nil {
+		return err
+	}
+
+	// create the nats connection option
+	opts := nats.GetDefaultOptions()
+	opts.Url = d.config.NatsServer
+	//opts.Servers = n.Config.Servers
+	opts.Name = hostNode.Name
+	opts.ReconnectWait = 2 * time.Second
+	opts.MaxReconnect = -1
+
+	var (
+		connection *nats.Conn
+	)
+
+	// let us connect using an exponential backoff mechanism
+	// we will attempt to connect 5 times to see whether there have started successfully or not.
+	const (
+		maxRetries = 5
+	)
+	// create a new instance of retrier that will try a maximum of five times, with
+	// an initial delay of 100 ms and a maximum delay of opts.ReconnectWait
+	retrier := retry.NewRetrier(maxRetries, 100*time.Millisecond, opts.ReconnectWait)
+	// handle the retry error
+	err = retrier.Run(func() error {
+		// connect to the NATs server
+		connection, err = opts.Connect()
+		if err != nil {
+			return err
+		}
+		// successful connection
+		return nil
+	})
+
+	// create the NATs connection encoder
+	encodedConn, err := nats.NewEncodedConn(connection, protobuf.PROTOBUF_ENCODER)
+	// handle the error
+	if err != nil {
+		return err
+	}
+	// set the connection
+	d.natsConnection = encodedConn
+	// set initialized
+	d.initialized = atomic.NewBool(true)
+	// set the host node
+	d.hostNode = hostNode
+	return nil
+}
+
+// Register registers this node to a service discovery directory.
+func (d *Discovery) Register() error {
+	// acquire the lock
+	d.mu.Lock()
+	// release the lock
+	defer d.mu.Unlock()
+
+	// first check whether the discovery provider has started
+	// avoid to re-register the discovery
+	if d.registered.Load() {
+		return discovery.ErrAlreadyRegistered
+	}
+
+	// create the subscription handler
+	subscriptionHandler := func(subj, reply string, msg *internalpb.NatsMessage) {
+		switch msg.GetMessageType() {
+		case internalpb.NatsMessageType_NATS_MESSAGE_TYPE_DEREGISTER:
+			// add logging information
+			d.logger.Infof("received an de-registration request from peer[name=%s, host=%s, port=%d]",
+				msg.GetName(), msg.GetHost(), msg.GetPort())
+		case internalpb.NatsMessageType_NATS_MESSAGE_TYPE_REGISTER:
+			// add logging information
+			d.logger.Infof("received an registration request from peer[name=%s, host=%s, port=%d]",
+				msg.GetName(), msg.GetHost(), msg.GetPort())
+		case internalpb.NatsMessageType_NATS_MESSAGE_TYPE_REQUEST:
+			// add logging information
+			d.logger.Infof("received an identification request from peer[name=%s, host=%s, port=%d]",
+				msg.GetName(), msg.GetHost(), msg.GetPort())
+			// send the reply
+			replyMessage := &internalpb.NatsMessage{
+				Host:        d.hostNode.Host,
+				Port:        int32(d.hostNode.GossipPort),
+				Name:        d.hostNode.Name,
+				MessageType: internalpb.NatsMessageType_NATS_MESSAGE_TYPE_RESPONSE,
+			}
+
+			// send the reply and handle the error
+			if err := d.natsConnection.Publish(reply, replyMessage); err != nil {
+				d.logger.Errorf("failed to reply for identification request from peer[name=%s, host=%s, port=%d]",
+					msg.GetName(), msg.GetHost(), msg.GetPort())
+			}
+		}
+	}
+	// start listening to incoming messages
+	subscription, err := d.natsConnection.Subscribe(d.config.NatsSubject, subscriptionHandler)
+	// return any eventual error
+	if err != nil {
+		return err
+	}
+
+	// add the subscription to the list of subscriptions
+	d.subscriptions = append(d.subscriptions, subscription)
+	// set the registration flag to true
+	d.registered = atomic.NewBool(true)
+	return nil
+}
+
+// Deregister removes this node from a service discovery directory.
+func (d *Discovery) Deregister() error {
+	// acquire the lock
+	d.mu.Lock()
+	// release the lock
+	defer d.mu.Unlock()
+
+	// unregister
+	defer func() {
+		d.registered.Store(false)
+	}()
+
+	// first check whether the discovery provider has been registered or not
+	if !d.registered.Load() {
+		return discovery.ErrNotRegistered
+	}
+
+	// shutdown all the subscriptions
+	for _, subscription := range d.subscriptions {
+		// unsubscribe and return when there is an error
+		if err := subscription.Unsubscribe(); err != nil {
+			return err
+		}
+	}
+
+	// send the de-registration message to notify peers
+	if d.natsConnection != nil {
+		// send a message to deregister stating we are out
+		return d.natsConnection.Publish(d.config.NatsSubject, &internalpb.NatsMessage{
+			Host:        d.hostNode.Host,
+			Port:        int32(d.hostNode.GossipPort),
+			Name:        d.hostNode.Name,
+			MessageType: internalpb.NatsMessageType_NATS_MESSAGE_TYPE_DEREGISTER,
+		})
+	}
+	return nil
+}
+
+// SetConfig registers the underlying discovery configuration
+func (d *Discovery) SetConfig(config discovery.Config) error {
+	// acquire the lock
+	d.mu.Lock()
+	// release the lock
+	defer d.mu.Unlock()
+
+	// first check whether the discovery provider is running
+	if d.initialized.Load() {
+		return discovery.ErrAlreadyInitialized
+	}
+
+	return d.setConfig(config)
+}
+
+// DiscoverPeers returns a list of known nodes.
+func (d *Discovery) DiscoverPeers() ([]string, error) {
+	// acquire the lock
+	d.mu.Lock()
+	// release the lock
+	defer d.mu.Unlock()
+
+	// first check whether the discovery provider is running
+	if !d.initialized.Load() {
+		return nil, discovery.ErrNotInitialized
+	}
+
+	// later check the provider is registered
+	if !d.registered.Load() {
+		return nil, discovery.ErrNotRegistered
+	}
+
+	// Set up a reply channel, then broadcast for all peers to
+	// report their presence.
+	// collect as many responses as possible in the given timeout.
+	inbox := nats.NewInbox()
+	recv := make(chan *internalpb.NatsMessage)
+
+	// bind to receive messages
+	sub, err := d.natsConnection.BindRecvChan(inbox, recv)
+	if err != nil {
+		return nil, err
+	}
+
+	// send registration request and return in case of error
+	if err = d.natsConnection.PublishRequest(d.config.NatsSubject, inbox, &internalpb.NatsMessage{
+		Host:        d.hostNode.Host,
+		Port:        int32(d.hostNode.GossipPort),
+		Name:        d.hostNode.Name,
+		MessageType: internalpb.NatsMessageType_NATS_MESSAGE_TYPE_REQUEST,
+	}); err != nil {
+		return nil, err
+	}
+
+	// define the list of peers to lookup
+	var peers []string
+	// define the timeout
+	timeout := time.After(d.config.Timeout)
+	// get the host node gossip address
+	me := d.hostNode.GossipAddress()
+
+	// loop till we have enough nodes
+	for {
+		select {
+		case m, ok := <-recv:
+			if !ok {
+				// Subscription is closed
+				return peers, nil
+			}
+
+			// get the found peer address
+			addr := net.JoinHostPort(m.GetHost(), strconv.Itoa(int(m.GetPort())))
+			// ignore this node
+			if addr == me {
+				// Ignore a reply from self
+				continue
+			}
+
+			// add the peer to the list of discovered peers
+			peers = append(peers, addr)
+
+		case <-timeout:
+			// close the receiving channel
+			_ = sub.Unsubscribe()
+			close(recv)
+		}
+	}
+}
+
+// Close closes the provider
+func (d *Discovery) Close() error {
+	// set the initialized to false
+	d.initialized.Store(false)
+	// set registered to false
+	d.registered.Store(false)
+
+	if d.natsConnection != nil {
+		// close all the subscriptions
+		for _, subscription := range d.subscriptions {
+			// unsubscribe and ignore the error
+			_ = subscription.Unsubscribe()
+		}
+
+		// flush all messages
+		return d.natsConnection.Flush()
+	}
+	return nil
+}
+
+// setConfig sets the kubernetes discoConfig
+func (d *Discovery) setConfig(config discovery.Config) (err error) {
+	// create an instance of option
+	option := new(discoConfig)
+	// extract the nats server address
+	option.NatsServer, err = config.GetString(NatsServer)
+	// handle the error in case the nats server value is not properly set
+	if err != nil {
+		return err
+	}
+	// extract the nats subject address
+	option.NatsSubject, err = config.GetString(NatsSubject)
+	// handle the error in case the nats subject value is not properly set
+	if err != nil {
+		return err
+	}
+
+	// extract the actor system name
+	option.ActorSystemName, err = config.GetString(ActorSystemName)
+	// handle the error in case the actor system name value is not properly set
+	if err != nil {
+		return err
+	}
+	// extract the application name
+	option.ApplicationName, err = config.GetString(ApplicationName)
+	// handle the error in case the application name value is not properly set
+	if err != nil {
+		return err
+	}
+	// in case none of the above extraction fails then set the option
+	d.config = option
+	return nil
+}
diff --git a/discovery/nats/option.go b/discovery/nats/option.go
new file mode 100644
index 0000000..1bf4907
--- /dev/null
+++ b/discovery/nats/option.go
@@ -0,0 +1,50 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2022-2023 Tochemey
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package nats
+
+import "github.com/tochemey/goakt/log"
+
+// Option is the interface that applies a configuration option.
+type Option interface {
+	// Apply sets the Option value of a config.
+	Apply(disco *Discovery)
+}
+
+var _ Option = OptionFunc(nil)
+
+// OptionFunc implements the Option interface.
+type OptionFunc func(disco *Discovery)
+
+// Apply applies the Cluster's option
+func (f OptionFunc) Apply(disco *Discovery) {
+	f(disco)
+}
+
+// WithLogger sets the logger
+func WithLogger(logger log.Logger) Option {
+	return OptionFunc(func(disco *Discovery) {
+		disco.logger = logger
+	})
+}
diff --git a/cluster/node.go b/discovery/node.go
similarity index 87%
rename from cluster/node.go
rename to discovery/node.go
index 583e903..7e66be8 100644
--- a/cluster/node.go
+++ b/discovery/node.go
@@ -22,7 +22,7 @@
  * SOFTWARE.
  */
 
-package cluster
+package discovery
 
 import (
 	"net"
@@ -31,7 +31,7 @@ import (
 	"github.com/caarlos0/env/v9"
 )
 
-// hostNodeConfig helps read the host node settings
+// hostNodeConfig helps read the host Node settings
 type hostNodeConfig struct {
 	GossipPort   int    `env:"GOSSIP_PORT"`
 	ClusterPort  int    `env:"CLUSTER_PORT"`
@@ -40,8 +40,8 @@ type hostNodeConfig struct {
 	Host         string `env:"NODE_IP"`
 }
 
-// node represents a discovered node
-type node struct {
+// Node represents a discovered Node
+type Node struct {
 	// Name specifies the discovered node's Name
 	Name string
 	// Host specifies the discovered node's Host
@@ -55,17 +55,17 @@ type node struct {
 }
 
 // ClusterAddress returns address the node's peers will use to connect to
-func (n node) ClusterAddress() string {
+func (n Node) ClusterAddress() string {
 	return net.JoinHostPort(n.Host, strconv.Itoa(n.ClusterPort))
 }
 
 // GossipAddress returns the node discovery address
-func (n node) GossipAddress() string {
+func (n Node) GossipAddress() string {
 	return net.JoinHostPort(n.Host, strconv.Itoa(n.GossipPort))
 }
 
-// hostNode returns the node where the discovery provider is running
-func hostNode() (*node, error) {
+// HostNode returns the Node where the discovery provider is running
+func HostNode() (*Node, error) {
 	// load the host node configuration
 	cfg := &hostNodeConfig{}
 	opts := env.Options{RequiredIfNoDef: true, UseFieldNameByDefault: false}
@@ -73,7 +73,7 @@ func hostNode() (*node, error) {
 		return nil, err
 	}
 	// create the host node
-	return &node{
+	return &Node{
 		Name:         cfg.Name,
 		Host:         cfg.Host,
 		GossipPort:   cfg.GossipPort,
diff --git a/discovery/provider.go b/discovery/provider.go
index dbead9a..af4d19d 100644
--- a/discovery/provider.go
+++ b/discovery/provider.go
@@ -38,4 +38,6 @@ type Provider interface {
 	SetConfig(config Config) error
 	// DiscoverPeers returns a list discovered nodes' addresses.
 	DiscoverPeers() ([]string, error)
+	// Close closes the provider
+	Close() error
 }
diff --git a/examples/actor-cluster/k8s/cmd/run.go b/examples/actor-cluster/k8s/cmd/run.go
index 64286d3..9f638c8 100644
--- a/examples/actor-cluster/k8s/cmd/run.go
+++ b/examples/actor-cluster/k8s/cmd/run.go
@@ -57,7 +57,7 @@ var runCmd = &cobra.Command{
 		// use the address default log. real-life implement the log interface`
 		logger := log.New(log.DebugLevel, os.Stdout)
 
-		// create the k8 configuration
+		// instantiate the k8 discovery provider
 		disco := kubernetes.NewDiscovery()
 		// define the discovery options
 		discoOptions := discovery.Config{
diff --git a/go.mod b/go.mod
index d7f0f3b..48ebe42 100644
--- a/go.mod
+++ b/go.mod
@@ -10,10 +10,13 @@ require (
 	github.com/cenkalti/backoff v2.2.1+incompatible
 	github.com/cespare/xxhash/v2 v2.2.0
 	github.com/deckarep/golang-set/v2 v2.3.1
+	github.com/flowchartsman/retry v1.2.0
 	github.com/google/go-cmp v0.6.0
 	github.com/google/uuid v1.3.1
 	github.com/grandcat/zeroconf v1.0.0
 	github.com/joho/godotenv v1.5.1
+	github.com/nats-io/nats-server/v2 v2.10.3
+	github.com/nats-io/nats.go v1.31.0
 	github.com/pkg/errors v0.9.1
 	github.com/spf13/cobra v1.7.0
 	github.com/stretchr/testify v1.8.4
@@ -65,12 +68,17 @@ require (
 	github.com/inconshreveable/mousetrap v1.1.0 // indirect
 	github.com/josharian/intern v1.0.0 // indirect
 	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/klauspost/compress v1.17.0 // indirect
 	github.com/mailru/easyjson v0.7.7 // indirect
 	github.com/miekg/dns v1.1.55 // indirect
+	github.com/minio/highwayhash v1.0.2 // indirect
 	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
 	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/mschoch/smat v0.2.0 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
+	github.com/nats-io/jwt/v2 v2.5.2 // indirect
+	github.com/nats-io/nkeys v0.4.5 // indirect
+	github.com/nats-io/nuid v1.0.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
@@ -81,6 +89,7 @@ require (
 	github.com/vmihailenco/msgpack/v5 v5.3.5 // indirect
 	github.com/vmihailenco/tagparser/v2 v2.0.0 // indirect
 	go.uber.org/multierr v1.11.0 // indirect
+	golang.org/x/crypto v0.14.0 // indirect
 	golang.org/x/mod v0.12.0 // indirect
 	golang.org/x/oauth2 v0.11.0 // indirect
 	golang.org/x/sync v0.3.0 // indirect
diff --git a/go.sum b/go.sum
index 19f0da6..dde2ba4 100644
--- a/go.sum
+++ b/go.sum
@@ -52,6 +52,8 @@ github.com/emicklei/go-restful/v3 v3.11.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRr
 github.com/evanphx/json-patch v5.6.0+incompatible h1:jBYDEEiFBPxA0v50tFdvOzQQTCvpL6mnFh5mB2/l16U=
 github.com/evanphx/json-patch v5.6.0+incompatible/go.mod h1:50XU6AFN0ol/bzJsmQLiYLvXMP4fmwYFNcr97nuDLSk=
 github.com/fatih/color v1.7.0/go.mod h1:Zm6kSWBoL9eyXnKyktHP6abPY2pDugNf5KwzbycvMj4=
+github.com/flowchartsman/retry v1.2.0 h1:qDhlw6RNufXz6RGr+IiYimFpMMkt77SUSHY5tgFaUCU=
+github.com/flowchartsman/retry v1.2.0/go.mod h1:+sfx8OgCCiAr3t5jh2Gk+T0fRTI+k52edaYxURQxY64=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
 github.com/fsnotify/fsnotify v1.4.9 h1:hsms1Qyu0jgnwNXIxa+/V/PDsU6CfLf6CNO8H7IWoS4=
 github.com/fsnotify/fsnotify v1.4.9/go.mod h1:znqG4EE+3YCdAaPaxE2ZRY/06pZUdp0tY4IgpuI1SZQ=
@@ -157,6 +159,8 @@ github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHm
 github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
+github.com/klauspost/compress v1.17.0 h1:Rnbp4K9EjcDuVuHtd0dgA4qNuv9yKDYKK1ulpJwgrqM=
+github.com/klauspost/compress v1.17.0/go.mod h1:ntbaceVETuRiXiv4DpjP66DpAtAGkEQskQzEyD//IeE=
 github.com/konsorten/go-windows-terminal-sequences v1.0.1/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
 github.com/kr/logfmt v0.0.0-20140226030751-b84e30acd515/go.mod h1:+0opPa2QZZtGFBFZlji/RkVcI2GknAs/DXo4wKdlNEc=
 github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
@@ -176,6 +180,8 @@ github.com/miekg/dns v1.1.27/go.mod h1:KNUDUusw/aVsxyTYZM1oqvCicbwhgbNgztCETuNZ7
 github.com/miekg/dns v1.1.45/go.mod h1:e3IlAVfNqAllflbibAZEWOXOQ+Ynzk/dDozDxY7XnME=
 github.com/miekg/dns v1.1.55 h1:GoQ4hpsj0nFLYe+bWiCToyrBEJXkQfOOIvFGFy0lEgo=
 github.com/miekg/dns v1.1.55/go.mod h1:uInx36IzPl7FYnDcMeVWxj9byh7DutNykX4G9Sj60FY=
+github.com/minio/highwayhash v1.0.2 h1:Aak5U0nElisjDCfPSG79Tgzkn2gl66NxOMspRrKnA/g=
+github.com/minio/highwayhash v1.0.2/go.mod h1:BQskDq+xkJ12lmlUUi7U0M5Swg3EWR+dLTk+kldvVxY=
 github.com/mitchellh/cli v1.0.0/go.mod h1:hNIlj7HEI86fIcpObd7a0FcrxTWetlwJDGcceTlRvqc=
 github.com/mitchellh/go-wordwrap v1.0.0/go.mod h1:ZXFpozHsX6DPmq2I0TCekCxypsnAUbP2oI0UX1GXzOo=
 github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
@@ -190,6 +196,16 @@ github.com/mschoch/smat v0.2.0/go.mod h1:kc9mz7DoBKqDyiRL7VZN8KvXQMWeTaVnttLRXOl
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
 github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
 github.com/mwitkow/go-conntrack v0.0.0-20161129095857-cc309e4a2223/go.mod h1:qRWi+5nqEBWmkhHvq77mSJWrCKwh8bxhgT7d/eI7P4U=
+github.com/nats-io/jwt/v2 v2.5.2 h1:DhGH+nKt+wIkDxM6qnVSKjokq5t59AZV5HRcFW0zJwU=
+github.com/nats-io/jwt/v2 v2.5.2/go.mod h1:24BeQtRwxRV8ruvC4CojXlx/WQ/VjuwlYiH+vu/+ibI=
+github.com/nats-io/nats-server/v2 v2.10.3 h1:nk2QVLpJUh3/AhZCJlQdTfj2oeLDvWnn1Z6XzGlNFm0=
+github.com/nats-io/nats-server/v2 v2.10.3/go.mod h1:lzrskZ/4gyMAh+/66cCd+q74c6v7muBypzfWhP/MAaM=
+github.com/nats-io/nats.go v1.31.0 h1:/WFBHEc/dOKBF6qf1TZhrdEfTmOZ5JzdJ+Y3m6Y/p7E=
+github.com/nats-io/nats.go v1.31.0/go.mod h1:di3Bm5MLsoB4Bx61CBTsxuarI36WbhAwOm8QrW39+i8=
+github.com/nats-io/nkeys v0.4.5 h1:Zdz2BUlFm4fJlierwvGK+yl20IAKUm7eV6AAZXEhkPk=
+github.com/nats-io/nkeys v0.4.5/go.mod h1:XUkxdLPTufzlihbamfzQ7mw/VGx6ObUs+0bN5sNvt64=
+github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw=
+github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c=
 github.com/nxadm/tail v1.4.4/go.mod h1:kenIhsEOeOJmVchQTgglprH7qJGnHDVpk1VPCcaMI8A=
 github.com/nxadm/tail v1.4.8 h1:nPr65rt6Y5JFSKQO7qToXr7pePgD6Gwiw05lkbyAQTE=
 github.com/nxadm/tail v1.4.8/go.mod h1:+ncqLTQzXmGhMZNUePPaPqPvBxHAIsmXswZKocGu+AU=
@@ -294,6 +310,9 @@ golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACk
 golang.org/x/crypto v0.0.0-20190923035154-9ee001bba392/go.mod h1:/lpIB1dKB+9EgE3H3cr1v9wB50oz8l4C4h62xy7jSTY=
 golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
+golang.org/x/crypto v0.14.0 h1:wBqGXzWJW6m1XrIKlAH0Hs1JJ7+9KBwnIO8v66Q9cHc=
+golang.org/x/crypto v0.14.0/go.mod h1:MVFd36DqK4CsrnJYDkBA3VC4m2GkXAM0PvzMCn4JQf4=
+golang.org/x/exp v0.0.0-20230905200255-921286631fa9 h1:GoHiUyI/Tp2nVkLI2mCxVkOjsbSXD66ic0XW0js0R9g=
 golang.org/x/mod v0.1.1-0.20191105210325-c90efee705ee/go.mod h1:QqPTAvyqsEbceGzBzNggFXnrqF1CaUcvgkdR5Ot7KZg=
 golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
@@ -331,6 +350,7 @@ golang.org/x/sys v0.0.0-20180823144017-11551d06cbcc/go.mod h1:STP8DvDyc/dI5b8T5h
 golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20180909124046-d0be0721c37e/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20181116152217-5ac8a444bdc5/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.0.0-20190130150945-aca44879d564/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
 golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20190422165155-953cdadca894/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
diff --git a/internal/v1/nats.pb.go b/internal/v1/nats.pb.go
new file mode 100644
index 0000000..7e54185
--- /dev/null
+++ b/internal/v1/nats.pb.go
@@ -0,0 +1,263 @@
+// Code generated by protoc-gen-go. DO NOT EDIT.
+// versions:
+// 	protoc-gen-go v1.31.0
+// 	protoc        (unknown)
+// source: internal/v1/nats.proto
+
+package internalpb
+
+import (
+	_ "github.com/tochemey/goakt/pb/address/v1"
+	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
+	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
+	_ "google.golang.org/protobuf/types/known/anypb"
+	_ "google.golang.org/protobuf/types/known/timestamppb"
+	reflect "reflect"
+	sync "sync"
+)
+
+const (
+	// Verify that this generated code is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
+	// Verify that runtime/protoimpl is sufficiently up-to-date.
+	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
+)
+
+// NatsMessageType defines the NATs message type
+type NatsMessageType int32
+
+const (
+	NatsMessageType_NATS_MESSAGE_TYPE_REGISTER   NatsMessageType = 0
+	NatsMessageType_NATS_MESSAGE_TYPE_DEREGISTER NatsMessageType = 1
+	NatsMessageType_NATS_MESSAGE_TYPE_REQUEST    NatsMessageType = 2
+	NatsMessageType_NATS_MESSAGE_TYPE_RESPONSE   NatsMessageType = 3
+)
+
+// Enum value maps for NatsMessageType.
+var (
+	NatsMessageType_name = map[int32]string{
+		0: "NATS_MESSAGE_TYPE_REGISTER",
+		1: "NATS_MESSAGE_TYPE_DEREGISTER",
+		2: "NATS_MESSAGE_TYPE_REQUEST",
+		3: "NATS_MESSAGE_TYPE_RESPONSE",
+	}
+	NatsMessageType_value = map[string]int32{
+		"NATS_MESSAGE_TYPE_REGISTER":   0,
+		"NATS_MESSAGE_TYPE_DEREGISTER": 1,
+		"NATS_MESSAGE_TYPE_REQUEST":    2,
+		"NATS_MESSAGE_TYPE_RESPONSE":   3,
+	}
+)
+
+func (x NatsMessageType) Enum() *NatsMessageType {
+	p := new(NatsMessageType)
+	*p = x
+	return p
+}
+
+func (x NatsMessageType) String() string {
+	return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x))
+}
+
+func (NatsMessageType) Descriptor() protoreflect.EnumDescriptor {
+	return file_internal_v1_nats_proto_enumTypes[0].Descriptor()
+}
+
+func (NatsMessageType) Type() protoreflect.EnumType {
+	return &file_internal_v1_nats_proto_enumTypes[0]
+}
+
+func (x NatsMessageType) Number() protoreflect.EnumNumber {
+	return protoreflect.EnumNumber(x)
+}
+
+// Deprecated: Use NatsMessageType.Descriptor instead.
+func (NatsMessageType) EnumDescriptor() ([]byte, []int) {
+	return file_internal_v1_nats_proto_rawDescGZIP(), []int{0}
+}
+
+// NatsMessage defines the NATs message
+// used by the discovery provider
+type NatsMessage struct {
+	state         protoimpl.MessageState
+	sizeCache     protoimpl.SizeCache
+	unknownFields protoimpl.UnknownFields
+
+	// Specifies the host name of the client node
+	Host string `protobuf:"bytes,1,opt,name=host,proto3" json:"host,omitempty"`
+	// Specifies the port of the client node
+	Port int32 `protobuf:"varint,2,opt,name=port,proto3" json:"port,omitempty"`
+	// Specifies the client name
+	Name string `protobuf:"bytes,3,opt,name=name,proto3" json:"name,omitempty"`
+	// Specifies the message type
+	MessageType NatsMessageType `protobuf:"varint,4,opt,name=message_type,json=messageType,proto3,enum=internal.v1.NatsMessageType" json:"message_type,omitempty"`
+}
+
+func (x *NatsMessage) Reset() {
+	*x = NatsMessage{}
+	if protoimpl.UnsafeEnabled {
+		mi := &file_internal_v1_nats_proto_msgTypes[0]
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		ms.StoreMessageInfo(mi)
+	}
+}
+
+func (x *NatsMessage) String() string {
+	return protoimpl.X.MessageStringOf(x)
+}
+
+func (*NatsMessage) ProtoMessage() {}
+
+func (x *NatsMessage) ProtoReflect() protoreflect.Message {
+	mi := &file_internal_v1_nats_proto_msgTypes[0]
+	if protoimpl.UnsafeEnabled && x != nil {
+		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
+		if ms.LoadMessageInfo() == nil {
+			ms.StoreMessageInfo(mi)
+		}
+		return ms
+	}
+	return mi.MessageOf(x)
+}
+
+// Deprecated: Use NatsMessage.ProtoReflect.Descriptor instead.
+func (*NatsMessage) Descriptor() ([]byte, []int) {
+	return file_internal_v1_nats_proto_rawDescGZIP(), []int{0}
+}
+
+func (x *NatsMessage) GetHost() string {
+	if x != nil {
+		return x.Host
+	}
+	return ""
+}
+
+func (x *NatsMessage) GetPort() int32 {
+	if x != nil {
+		return x.Port
+	}
+	return 0
+}
+
+func (x *NatsMessage) GetName() string {
+	if x != nil {
+		return x.Name
+	}
+	return ""
+}
+
+func (x *NatsMessage) GetMessageType() NatsMessageType {
+	if x != nil {
+		return x.MessageType
+	}
+	return NatsMessageType_NATS_MESSAGE_TYPE_REGISTER
+}
+
+var File_internal_v1_nats_proto protoreflect.FileDescriptor
+
+var file_internal_v1_nats_proto_rawDesc = []byte{
+	0x0a, 0x16, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x76, 0x31, 0x2f, 0x6e, 0x61,
+	0x74, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x12, 0x0b, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e,
+	0x61, 0x6c, 0x2e, 0x76, 0x31, 0x1a, 0x18, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x2f, 0x76,
+	0x31, 0x2f, 0x61, 0x64, 0x64, 0x72, 0x65, 0x73, 0x73, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x1a,
+	0x19, 0x67, 0x6f, 0x6f, 0x67, 0x6c, 0x65, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66,
+	0x2f, 0x61, 0x6e, 0x79, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x1a, 0x1f, 0x67, 0x6f, 0x6f, 0x67,
+	0x6c, 0x65, 0x2f, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x62, 0x75, 0x66, 0x2f, 0x74, 0x69, 0x6d, 0x65,
+	0x73, 0x74, 0x61, 0x6d, 0x70, 0x2e, 0x70, 0x72, 0x6f, 0x74, 0x6f, 0x22, 0x8a, 0x01, 0x0a, 0x0b,
+	0x4e, 0x61, 0x74, 0x73, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x12, 0x12, 0x0a, 0x04, 0x68,
+	0x6f, 0x73, 0x74, 0x18, 0x01, 0x20, 0x01, 0x28, 0x09, 0x52, 0x04, 0x68, 0x6f, 0x73, 0x74, 0x12,
+	0x12, 0x0a, 0x04, 0x70, 0x6f, 0x72, 0x74, 0x18, 0x02, 0x20, 0x01, 0x28, 0x05, 0x52, 0x04, 0x70,
+	0x6f, 0x72, 0x74, 0x12, 0x12, 0x0a, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x18, 0x03, 0x20, 0x01, 0x28,
+	0x09, 0x52, 0x04, 0x6e, 0x61, 0x6d, 0x65, 0x12, 0x3f, 0x0a, 0x0c, 0x6d, 0x65, 0x73, 0x73, 0x61,
+	0x67, 0x65, 0x5f, 0x74, 0x79, 0x70, 0x65, 0x18, 0x04, 0x20, 0x01, 0x28, 0x0e, 0x32, 0x1c, 0x2e,
+	0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2e, 0x76, 0x31, 0x2e, 0x4e, 0x61, 0x74, 0x73,
+	0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x54, 0x79, 0x70, 0x65, 0x52, 0x0b, 0x6d, 0x65, 0x73,
+	0x73, 0x61, 0x67, 0x65, 0x54, 0x79, 0x70, 0x65, 0x2a, 0x92, 0x01, 0x0a, 0x0f, 0x4e, 0x61, 0x74,
+	0x73, 0x4d, 0x65, 0x73, 0x73, 0x61, 0x67, 0x65, 0x54, 0x79, 0x70, 0x65, 0x12, 0x1e, 0x0a, 0x1a,
+	0x4e, 0x41, 0x54, 0x53, 0x5f, 0x4d, 0x45, 0x53, 0x53, 0x41, 0x47, 0x45, 0x5f, 0x54, 0x59, 0x50,
+	0x45, 0x5f, 0x52, 0x45, 0x47, 0x49, 0x53, 0x54, 0x45, 0x52, 0x10, 0x00, 0x12, 0x20, 0x0a, 0x1c,
+	0x4e, 0x41, 0x54, 0x53, 0x5f, 0x4d, 0x45, 0x53, 0x53, 0x41, 0x47, 0x45, 0x5f, 0x54, 0x59, 0x50,
+	0x45, 0x5f, 0x44, 0x45, 0x52, 0x45, 0x47, 0x49, 0x53, 0x54, 0x45, 0x52, 0x10, 0x01, 0x12, 0x1d,
+	0x0a, 0x19, 0x4e, 0x41, 0x54, 0x53, 0x5f, 0x4d, 0x45, 0x53, 0x53, 0x41, 0x47, 0x45, 0x5f, 0x54,
+	0x59, 0x50, 0x45, 0x5f, 0x52, 0x45, 0x51, 0x55, 0x45, 0x53, 0x54, 0x10, 0x02, 0x12, 0x1e, 0x0a,
+	0x1a, 0x4e, 0x41, 0x54, 0x53, 0x5f, 0x4d, 0x45, 0x53, 0x53, 0x41, 0x47, 0x45, 0x5f, 0x54, 0x59,
+	0x50, 0x45, 0x5f, 0x52, 0x45, 0x53, 0x50, 0x4f, 0x4e, 0x53, 0x45, 0x10, 0x03, 0x42, 0x9d, 0x01,
+	0x0a, 0x0f, 0x63, 0x6f, 0x6d, 0x2e, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x2e, 0x76,
+	0x31, 0x42, 0x09, 0x4e, 0x61, 0x74, 0x73, 0x50, 0x72, 0x6f, 0x74, 0x6f, 0x48, 0x02, 0x50, 0x01,
+	0x5a, 0x30, 0x67, 0x69, 0x74, 0x68, 0x75, 0x62, 0x2e, 0x63, 0x6f, 0x6d, 0x2f, 0x74, 0x6f, 0x63,
+	0x68, 0x65, 0x6d, 0x65, 0x79, 0x2f, 0x67, 0x6f, 0x61, 0x6b, 0x74, 0x2f, 0x69, 0x6e, 0x74, 0x65,
+	0x72, 0x6e, 0x61, 0x6c, 0x2f, 0x76, 0x31, 0x3b, 0x69, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c,
+	0x70, 0x62, 0xa2, 0x02, 0x03, 0x49, 0x58, 0x58, 0xaa, 0x02, 0x0b, 0x49, 0x6e, 0x74, 0x65, 0x72,
+	0x6e, 0x61, 0x6c, 0x2e, 0x56, 0x31, 0xca, 0x02, 0x0b, 0x49, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61,
+	0x6c, 0x5c, 0x56, 0x31, 0xe2, 0x02, 0x17, 0x49, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x5c,
+	0x56, 0x31, 0x5c, 0x47, 0x50, 0x42, 0x4d, 0x65, 0x74, 0x61, 0x64, 0x61, 0x74, 0x61, 0xea, 0x02,
+	0x0c, 0x49, 0x6e, 0x74, 0x65, 0x72, 0x6e, 0x61, 0x6c, 0x3a, 0x3a, 0x56, 0x31, 0x62, 0x06, 0x70,
+	0x72, 0x6f, 0x74, 0x6f, 0x33,
+}
+
+var (
+	file_internal_v1_nats_proto_rawDescOnce sync.Once
+	file_internal_v1_nats_proto_rawDescData = file_internal_v1_nats_proto_rawDesc
+)
+
+func file_internal_v1_nats_proto_rawDescGZIP() []byte {
+	file_internal_v1_nats_proto_rawDescOnce.Do(func() {
+		file_internal_v1_nats_proto_rawDescData = protoimpl.X.CompressGZIP(file_internal_v1_nats_proto_rawDescData)
+	})
+	return file_internal_v1_nats_proto_rawDescData
+}
+
+var file_internal_v1_nats_proto_enumTypes = make([]protoimpl.EnumInfo, 1)
+var file_internal_v1_nats_proto_msgTypes = make([]protoimpl.MessageInfo, 1)
+var file_internal_v1_nats_proto_goTypes = []interface{}{
+	(NatsMessageType)(0), // 0: internal.v1.NatsMessageType
+	(*NatsMessage)(nil),  // 1: internal.v1.NatsMessage
+}
+var file_internal_v1_nats_proto_depIdxs = []int32{
+	0, // 0: internal.v1.NatsMessage.message_type:type_name -> internal.v1.NatsMessageType
+	1, // [1:1] is the sub-list for method output_type
+	1, // [1:1] is the sub-list for method input_type
+	1, // [1:1] is the sub-list for extension type_name
+	1, // [1:1] is the sub-list for extension extendee
+	0, // [0:1] is the sub-list for field type_name
+}
+
+func init() { file_internal_v1_nats_proto_init() }
+func file_internal_v1_nats_proto_init() {
+	if File_internal_v1_nats_proto != nil {
+		return
+	}
+	if !protoimpl.UnsafeEnabled {
+		file_internal_v1_nats_proto_msgTypes[0].Exporter = func(v interface{}, i int) interface{} {
+			switch v := v.(*NatsMessage); i {
+			case 0:
+				return &v.state
+			case 1:
+				return &v.sizeCache
+			case 2:
+				return &v.unknownFields
+			default:
+				return nil
+			}
+		}
+	}
+	type x struct{}
+	out := protoimpl.TypeBuilder{
+		File: protoimpl.DescBuilder{
+			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
+			RawDescriptor: file_internal_v1_nats_proto_rawDesc,
+			NumEnums:      1,
+			NumMessages:   1,
+			NumExtensions: 0,
+			NumServices:   0,
+		},
+		GoTypes:           file_internal_v1_nats_proto_goTypes,
+		DependencyIndexes: file_internal_v1_nats_proto_depIdxs,
+		EnumInfos:         file_internal_v1_nats_proto_enumTypes,
+		MessageInfos:      file_internal_v1_nats_proto_msgTypes,
+	}.Build()
+	File_internal_v1_nats_proto = out.File
+	file_internal_v1_nats_proto_rawDesc = nil
+	file_internal_v1_nats_proto_goTypes = nil
+	file_internal_v1_nats_proto_depIdxs = nil
+}
diff --git a/protos/goakt/internal/v1/nats.proto b/protos/goakt/internal/v1/nats.proto
new file mode 100644
index 0000000..ed91d05
--- /dev/null
+++ b/protos/goakt/internal/v1/nats.proto
@@ -0,0 +1,30 @@
+syntax = "proto3";
+
+package internal.v1;
+
+import "address/v1/address.proto";
+import "google/protobuf/any.proto";
+import "google/protobuf/timestamp.proto";
+
+option go_package = "github.com/tochemey/goakt/internal/v1;internalpb";
+
+// NatsMessageType defines the NATs message type
+enum NatsMessageType {
+  NATS_MESSAGE_TYPE_REGISTER = 0;
+  NATS_MESSAGE_TYPE_DEREGISTER = 1;
+  NATS_MESSAGE_TYPE_REQUEST = 2;
+  NATS_MESSAGE_TYPE_RESPONSE = 3;
+}
+
+// NatsMessage defines the NATs message
+// used by the discovery provider
+message NatsMessage {
+  // Specifies the host name of the client node
+  string host = 1;
+  // Specifies the port of the client node
+  int32 port = 2;
+  // Specifies the client name
+  string name = 3;
+  // Specifies the message type
+  NatsMessageType message_type = 4;
+}
diff --git a/readme.md b/readme.md
index 96ac8c9..f8610df 100644
--- a/readme.md
+++ b/readme.md
@@ -38,6 +38,7 @@ Also, check reference section at the end of the post for more material regarding
     - [Discovery Providers](#built-in-discovery-providers)
         - [Kubernetes](#kubernetes-discovery-provider-setup)
         - [mDNS](#mdns-discovery-provider-setup)
+        - [NATS](#nats-discovery-provider-setup)
 - [Examples](#examples)
 - [Contribution](#contribution)
     - [Local Test and Linter](#test--linter)
@@ -226,7 +227,7 @@ The following outlines the cluster mode operations which can help have a healthy
 
 #### Built-in Discovery Providers
 
-##### Kubernetes Discovery Provider setup
+##### Kubernetes Discovery Provider Setup
 
 To get the kubernetes discovery working as expected, the following pod labels need to be set:
 
@@ -241,6 +242,27 @@ engine to work as expected:
 - `cluster-port`: help the cluster engine to communicate with other GoAkt nodes in the cluster
 - `remoting-port`: help for remoting messaging between actors
 
+###### Get Started
+
+```go
+const (
+  namespace          = "default"
+  applicationName    = "accounts"
+  actorSystemName    = "AccountsSystem"
+)
+// instantiate the k8 discovery provider
+disco := kubernetes.NewDiscovery()
+// define the discovery options
+discoOptions := discovery.Config{
+    kubernetes.ApplicationName: applicationName,
+    kubernetes.ActorSystemName: actorSystemName,
+    kubernetes.Namespace:       namespace,
+}
+// define the service discovery
+serviceDiscovery := discovery.NewServiceDiscovery(disco, discoOptions)
+// pass the service discovery when enabling cluster mode in the actor system
+```
+
 ###### Role Based Access
 
 You’ll also have to grant the Service Account that your pods run under access to list pods. The following configuration
@@ -278,13 +300,43 @@ roleRef:
 A working example can be found [here](./examples/actor-cluster/k8s) with a
 small [doc](./examples/actor-cluster/k8s/doc.md) showing how to run it.
 
-#### mDNS Discovery Provider setup
+##### mDNS Discovery Provider Setup
 
 - `Service Name`: the service name
 - `Domain`: The mDNS discovery domain
 - `Port`: The mDNS discovery port
 - `IPv6`: States whether to lookup for IPv6 addresses.
 
+##### NATS Discovery Provider Setup
+
+To use the NATS discovery provider one needs to provide the following:
+
+- `NATS Server Address`: the NATS Server address
+- `NATS Subject`: the NATS subject to use
+- `Actor System Name`: the actor system name
+- `Application Name`: the application name
+
+```go
+const (
+  natsServerAddr = "nats://localhost:4248"
+  natsSubject = "goakt-gossip"
+  applicationName        = "accounts"
+  actorSystemName = "AccountsSystem"
+)
+// instantiate the NATS discovery provider
+disco := nats.NewDiscovery()
+// define the discovery options
+discoOptions := discovery.Config{
+  ApplicationName: applicationName,
+  ActorSystemName: actorSystemName,
+  NatsServer:      natsServer,
+  NatsSubject:     natsSubject,
+}
+// define the service discovery
+serviceDiscovery := discovery.NewServiceDiscovery(disco, discoOptions)
+// pass the service discovery when enabling cluster mode in the actor system
+```
+
 ### Examples
 
 Kindly check out the [examples'](./examples) folder.


Test patch:
diff --git a/actors/actor_system_test.go b/actors/actor_system_test.go
index b3a6775..98b86b7 100644
--- a/actors/actor_system_test.go
+++ b/actors/actor_system_test.go
@@ -171,6 +171,7 @@ func TestActorSystem(t *testing.T) {
 		provider.EXPECT().Deregister().Return(nil)
 		provider.EXPECT().SetConfig(config).Return(nil)
 		provider.EXPECT().DiscoverPeers().Return(addrs, nil)
+		provider.EXPECT().Close().Return(nil)
 
 		// start the actor system
 		err = newActorSystem.Start(ctx)
diff --git a/cluster/cluster_test.go b/cluster/cluster_test.go
index d25a8c5..1222328 100644
--- a/cluster/cluster_test.go
+++ b/cluster/cluster_test.go
@@ -67,6 +67,7 @@ func TestCluster(t *testing.T) {
 		provider.EXPECT().Deregister().Return(nil)
 		provider.EXPECT().SetConfig(config).Return(nil)
 		provider.EXPECT().DiscoverPeers().Return(addrs, nil)
+		provider.EXPECT().Close().Return(nil)
 
 		// create the service discovery
 		serviceDiscovery := discovery.NewServiceDiscovery(provider, config)
@@ -125,6 +126,7 @@ func TestCluster(t *testing.T) {
 		provider.EXPECT().Deregister().Return(nil)
 		provider.EXPECT().SetConfig(config).Return(nil)
 		provider.EXPECT().DiscoverPeers().Return(addrs, nil)
+		provider.EXPECT().Close().Return(nil)
 
 		// create the service discovery
 		serviceDiscovery := discovery.NewServiceDiscovery(provider, config)
diff --git a/cluster/discovery_test.go b/cluster/discovery_test.go
index 62e0d32..ec9c254 100644
--- a/cluster/discovery_test.go
+++ b/cluster/discovery_test.go
@@ -347,6 +347,7 @@ func TestDiscoveryProvider(t *testing.T) {
 	t.Run("With Close", func(t *testing.T) {
 		// mock the underlying discovery provider
 		provider := new(testkit.Provider)
+		provider.EXPECT().Close().Return(nil)
 		wrapper := &discoveryProvider{
 			provider: provider,
 			log:      log.DefaultLogger.StdLogger(),
diff --git a/discovery/kubernetes/discovery_test.go b/discovery/kubernetes/discovery_test.go
index 577d0c1..48ffa9c 100644
--- a/discovery/kubernetes/discovery_test.go
+++ b/discovery/kubernetes/discovery_test.go
@@ -161,9 +161,9 @@ func TestDiscovery(t *testing.T) {
 		client := testclient.NewSimpleClientset(pods...)
 		// create the kubernetes discovery provider
 		provider := Discovery{
-			client:        client,
-			isInitialized: atomic.NewBool(true),
-			option: &option{
+			client:      client,
+			initialized: atomic.NewBool(true),
+			option: &discoConfig{
 				NameSpace:       ns,
 				ActorSystemName: actorSystemName,
 				ApplicationName: appName,
@@ -182,6 +182,14 @@ func TestDiscovery(t *testing.T) {
 		}
 
 		assert.ElementsMatch(t, expected, actual)
+		assert.NoError(t, provider.Close())
+	})
+	t.Run("With DiscoverPeers: not initialized", func(t *testing.T) {
+		provider := NewDiscovery()
+		peers, err := provider.DiscoverPeers()
+		assert.Error(t, err)
+		assert.Empty(t, peers)
+		assert.EqualError(t, err, discovery.ErrNotInitialized.Error())
 	})
 	t.Run("With SetConfig", func(t *testing.T) {
 		// create the various config option
@@ -207,7 +215,7 @@ func TestDiscovery(t *testing.T) {
 		actorSystemName := "AccountsSystem"
 		// create the instance of provider
 		provider := NewDiscovery()
-		provider.isInitialized = atomic.NewBool(true)
+		provider.initialized = atomic.NewBool(true)
 		// create the config
 		config := discovery.Config{
 			ApplicationName: applicationName,
@@ -220,6 +228,50 @@ func TestDiscovery(t *testing.T) {
 		assert.Error(t, err)
 		assert.EqualError(t, err, discovery.ErrAlreadyInitialized.Error())
 	})
+	t.Run("With SetConfig: actor system not set", func(t *testing.T) {
+		// create the various config option
+		namespace := "default"
+		applicationName := "accounts"
+		// create the instance of provider
+		provider := NewDiscovery()
+		// create the config
+		config := discovery.Config{
+			ApplicationName: applicationName,
+			Namespace:       namespace,
+		}
+
+		// set config
+		assert.Error(t, provider.SetConfig(config))
+	})
+	t.Run("With SetConfig: application name not set", func(t *testing.T) {
+		// create the various config option
+		namespace := "default"
+		actorSystemName := "AccountsSystem"
+		// create the instance of provider
+		provider := NewDiscovery()
+		// create the config
+		config := discovery.Config{
+			ActorSystemName: actorSystemName,
+			Namespace:       namespace,
+		}
+
+		// set config
+		assert.Error(t, provider.SetConfig(config))
+	})
+	t.Run("With SetConfig: namespace not set", func(t *testing.T) {
+		applicationName := "accounts"
+		actorSystemName := "AccountsSystem"
+		// create the instance of provider
+		provider := NewDiscovery()
+		// create the config
+		config := discovery.Config{
+			ApplicationName: applicationName,
+			ActorSystemName: actorSystemName,
+		}
+
+		// set config
+		assert.Error(t, provider.SetConfig(config))
+	})
 	t.Run("With Initialize", func(t *testing.T) {
 		// create the various config option
 		namespace := "default"
@@ -241,21 +293,21 @@ func TestDiscovery(t *testing.T) {
 	t.Run("With Initialize: already initialized", func(t *testing.T) {
 		// create the instance of provider
 		provider := NewDiscovery()
-		provider.isInitialized = atomic.NewBool(true)
+		provider.initialized = atomic.NewBool(true)
 		assert.Error(t, provider.Initialize())
 	})
 	t.Run("With Deregister", func(t *testing.T) {
 		// create the instance of provider
 		provider := NewDiscovery()
 		// for the sake of the test
-		provider.isInitialized = atomic.NewBool(true)
+		provider.initialized = atomic.NewBool(true)
 		assert.NoError(t, provider.Deregister())
 	})
 	t.Run("With Deregister when not initialized", func(t *testing.T) {
 		// create the instance of provider
 		provider := NewDiscovery()
 		// for the sake of the test
-		provider.isInitialized = atomic.NewBool(false)
+		provider.initialized = atomic.NewBool(false)
 		err := provider.Deregister()
 		assert.Error(t, err)
 		assert.EqualError(t, err, discovery.ErrNotInitialized.Error())
diff --git a/discovery/mdns/discovery_test.go b/discovery/mdns/discovery_test.go
index 5a50104..200471d 100644
--- a/discovery/mdns/discovery_test.go
+++ b/discovery/mdns/discovery_test.go
@@ -181,7 +181,7 @@ func TestDiscovery(t *testing.T) {
 		domain := "local."
 		// create the instance of provider
 		provider := NewDiscovery()
-		provider.isInitialized = atomic.NewBool(true)
+		provider.initialized = atomic.NewBool(true)
 		// create the config
 		config := discovery.Config{
 			Service:     serviceType,
@@ -220,7 +220,7 @@ func TestDiscovery(t *testing.T) {
 	t.Run("With Initialize: already initialized", func(t *testing.T) {
 		// create the instance of provider
 		provider := NewDiscovery()
-		provider.isInitialized = atomic.NewBool(true)
+		provider.initialized = atomic.NewBool(true)
 		assert.Error(t, provider.Initialize())
 	})
 	t.Run("With Register", func(t *testing.T) {
@@ -245,10 +245,10 @@ func TestDiscovery(t *testing.T) {
 		require.NoError(t, provider.Register())
 
 		time.Sleep(time.Second)
-		require.True(t, provider.isInitialized.Load())
+		require.True(t, provider.initialized.Load())
 		require.NoError(t, provider.Deregister())
 		time.Sleep(time.Second)
-		assert.False(t, provider.isInitialized.Load())
+		assert.False(t, provider.initialized.Load())
 	})
 	t.Run("With Register when already registered", func(t *testing.T) {
 		// create the various config option
@@ -272,26 +272,26 @@ func TestDiscovery(t *testing.T) {
 		require.NoError(t, provider.Register())
 
 		time.Sleep(time.Second)
-		require.True(t, provider.isInitialized.Load())
+		require.True(t, provider.initialized.Load())
 		err := provider.Register()
 		require.Error(t, err)
 		require.EqualError(t, err, discovery.ErrAlreadyRegistered.Error())
 		require.NoError(t, provider.Deregister())
 		time.Sleep(time.Second)
-		assert.False(t, provider.isInitialized.Load())
+		assert.False(t, provider.initialized.Load())
 	})
 	t.Run("With Deregister", func(t *testing.T) {
 		// create the instance of provider
 		provider := NewDiscovery()
 		// for the sake of the test
-		provider.isInitialized = atomic.NewBool(true)
+		provider.initialized = atomic.NewBool(true)
 		assert.NoError(t, provider.Deregister())
 	})
 	t.Run("With Deregister when not initialized", func(t *testing.T) {
 		// create the instance of provider
 		provider := NewDiscovery()
 		// for the sake of the test
-		provider.isInitialized = atomic.NewBool(false)
+		provider.initialized = atomic.NewBool(false)
 		err := provider.Deregister()
 		assert.Error(t, err)
 		assert.EqualError(t, err, discovery.ErrNotInitialized.Error())
@@ -319,7 +319,7 @@ func TestDiscovery(t *testing.T) {
 
 		// wait for registration to be completed
 		time.Sleep(time.Second)
-		require.True(t, provider.isInitialized.Load())
+		require.True(t, provider.initialized.Load())
 
 		// discover peers
 		peers, err := provider.DiscoverPeers()
@@ -328,4 +328,11 @@ func TestDiscovery(t *testing.T) {
 
 		assert.NoError(t, provider.Deregister())
 	})
+	t.Run("With DiscoverPeers: not initialized", func(t *testing.T) {
+		provider := NewDiscovery()
+		peers, err := provider.DiscoverPeers()
+		assert.Error(t, err)
+		assert.Empty(t, peers)
+		assert.EqualError(t, err, discovery.ErrNotInitialized.Error())
+	})
 }
diff --git a/discovery/nats/discovery_test.go b/discovery/nats/discovery_test.go
new file mode 100644
index 0000000..c84d542
--- /dev/null
+++ b/discovery/nats/discovery_test.go
@@ -0,0 +1,384 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2022-2023 Tochemey
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package nats
+
+import (
+	"os"
+	"strconv"
+	"testing"
+	"time"
+
+	"github.com/tochemey/goakt/log"
+
+	natsserver "github.com/nats-io/nats-server/v2/server"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"github.com/tochemey/goakt/discovery"
+	"github.com/travisjeffery/go-dynaport"
+	"go.uber.org/atomic"
+)
+
+func startNatsServer(t *testing.T) *natsserver.Server {
+	t.Helper()
+	serv, err := natsserver.NewServer(&natsserver.Options{
+		Host: "127.0.0.1",
+		Port: -1,
+	})
+
+	require.NoError(t, err)
+
+	ready := make(chan bool)
+	go func() {
+		ready <- true
+		serv.Start()
+	}()
+	<-ready
+
+	if !serv.ReadyForConnections(2 * time.Second) {
+		t.Fatalf("nats-io server failed to start")
+	}
+
+	return serv
+}
+
+func newPeer(t *testing.T, serverAddr string) *Discovery {
+	// generate the ports for the single node
+	nodePorts := dynaport.Get(3)
+	gossipPort := nodePorts[0]
+	clusterPort := nodePorts[1]
+	remotingPort := nodePorts[2]
+
+	// create a Cluster node
+	host := "localhost"
+	// set the environments
+	require.NoError(t, os.Setenv("GOSSIP_PORT", strconv.Itoa(gossipPort)))
+	require.NoError(t, os.Setenv("CLUSTER_PORT", strconv.Itoa(clusterPort)))
+	require.NoError(t, os.Setenv("REMOTING_PORT", strconv.Itoa(remotingPort)))
+	require.NoError(t, os.Setenv("NODE_NAME", "testNode"))
+	require.NoError(t, os.Setenv("NODE_IP", host))
+
+	// create the various config option
+	applicationName := "accounts"
+	actorSystemName := "AccountsSystem"
+	natsSubject := "some-subject"
+	// create the instance of provider
+	provider := NewDiscovery()
+
+	// create the config
+	config := discovery.Config{
+		ApplicationName: applicationName,
+		ActorSystemName: actorSystemName,
+		NatsServer:      serverAddr,
+		NatsSubject:     natsSubject,
+	}
+
+	// set config
+	err := provider.SetConfig(config)
+	require.NoError(t, err)
+
+	// initialize
+	err = provider.Initialize()
+	require.NoError(t, err)
+	// clear the env var
+	require.NoError(t, os.Unsetenv("GOSSIP_PORT"))
+	require.NoError(t, os.Unsetenv("CLUSTER_PORT"))
+	require.NoError(t, os.Unsetenv("REMOTING_PORT"))
+	require.NoError(t, os.Unsetenv("NODE_NAME"))
+	require.NoError(t, os.Unsetenv("NODE_IP"))
+	// return the provider
+	return provider
+}
+
+func TestDiscovery(t *testing.T) {
+	t.Run("With a new instance", func(t *testing.T) {
+		// create the instance of provider
+		provider := NewDiscovery()
+		require.NotNil(t, provider)
+		// assert that provider implements the Discovery interface
+		// this is a cheap test
+		// assert the type of svc
+		assert.IsType(t, &Discovery{}, provider)
+		var p interface{} = provider
+		_, ok := p.(discovery.Provider)
+		assert.True(t, ok)
+	})
+	t.Run("With ID assertion", func(t *testing.T) {
+		// cheap test
+		// create the instance of provider
+		provider := NewDiscovery()
+		require.NotNil(t, provider)
+		assert.Equal(t, "nats", provider.ID())
+	})
+	t.Run("With SetConfig", func(t *testing.T) {
+		// create the various config option
+		natsServer := "nats://127.0.0.1:2322"
+		applicationName := "accounts"
+		actorSystemName := "AccountsSystem"
+		natsSubject := "some-subject"
+		// create the instance of provider
+		provider := NewDiscovery()
+		// create the config
+		config := discovery.Config{
+			ApplicationName: applicationName,
+			ActorSystemName: actorSystemName,
+			NatsServer:      natsServer,
+			NatsSubject:     natsSubject,
+		}
+
+		// set config
+		assert.NoError(t, provider.SetConfig(config))
+	})
+	t.Run("With SetConfig: already initialized", func(t *testing.T) {
+		// start the NATS server
+		srv := startNatsServer(t)
+		// create the various config option
+		natsServer := srv.Addr().String()
+		applicationName := "accounts"
+		actorSystemName := "AccountsSystem"
+		natsSubject := "some-subject"
+		// create the instance of provider
+		provider := NewDiscovery()
+		provider.initialized = atomic.NewBool(true)
+		// create the config
+		config := discovery.Config{
+			ApplicationName: applicationName,
+			ActorSystemName: actorSystemName,
+			NatsServer:      natsServer,
+			NatsSubject:     natsSubject,
+		}
+
+		// set config
+		err := provider.SetConfig(config)
+		assert.Error(t, err)
+		assert.EqualError(t, err, discovery.ErrAlreadyInitialized.Error())
+		// stop the NATS server
+		t.Cleanup(srv.Shutdown)
+	})
+	t.Run("With SetConfig: nats server not set", func(t *testing.T) {
+		applicationName := "accounts"
+		actorSystemName := "AccountsSystem"
+		natsSubject := "some-subject"
+		// create the instance of provider
+		provider := NewDiscovery()
+		// create the config
+		config := discovery.Config{
+			ApplicationName: applicationName,
+			ActorSystemName: actorSystemName,
+			NatsSubject:     natsSubject,
+		}
+
+		// set config
+		assert.Error(t, provider.SetConfig(config))
+	})
+	t.Run("With SetConfig: nats subject not set", func(t *testing.T) {
+		// create the various config option
+		natsServer := "nats://127.0.0.1:2322"
+		applicationName := "accounts"
+		actorSystemName := "AccountsSystem"
+		// create the instance of provider
+		provider := NewDiscovery()
+		// create the config
+		config := discovery.Config{
+			ApplicationName: applicationName,
+			ActorSystemName: actorSystemName,
+			NatsServer:      natsServer,
+		}
+
+		// set config
+		assert.Error(t, provider.SetConfig(config))
+	})
+	t.Run("With SetConfig: actor system not set", func(t *testing.T) {
+		// create the various config option
+		natsServer := "nats://127.0.0.1:2322"
+		applicationName := "accounts"
+		natsSubject := "some-subject"
+		// create the instance of provider
+		provider := NewDiscovery()
+		// create the config
+		config := discovery.Config{
+			ApplicationName: applicationName,
+			NatsServer:      natsServer,
+			NatsSubject:     natsSubject,
+		}
+
+		// set config
+		assert.Error(t, provider.SetConfig(config))
+	})
+	t.Run("With SetConfig: application name not set", func(t *testing.T) {
+		// create the various config option
+		natsServer := "nats://127.0.0.1:2322"
+		actorSystemName := "AccountsSystem"
+		natsSubject := "some-subject"
+		// create the instance of provider
+		provider := NewDiscovery()
+		// create the config
+		config := discovery.Config{
+			ActorSystemName: actorSystemName,
+			NatsServer:      natsServer,
+			NatsSubject:     natsSubject,
+		}
+
+		// set config
+		assert.Error(t, provider.SetConfig(config))
+	})
+	t.Run("With Initialize", func(t *testing.T) {
+		// start the NATS server
+		srv := startNatsServer(t)
+
+		// generate the ports for the single node
+		nodePorts := dynaport.Get(3)
+		gossipPort := nodePorts[0]
+		clusterPort := nodePorts[1]
+		remotingPort := nodePorts[2]
+
+		// create a Cluster node
+		host := "localhost"
+		// set the environments
+		t.Setenv("GOSSIP_PORT", strconv.Itoa(gossipPort))
+		t.Setenv("CLUSTER_PORT", strconv.Itoa(clusterPort))
+		t.Setenv("REMOTING_PORT", strconv.Itoa(remotingPort))
+		t.Setenv("NODE_NAME", "testNode")
+		t.Setenv("NODE_IP", host)
+
+		// create the various config option
+		natsServer := srv.Addr().String()
+		applicationName := "accounts"
+		actorSystemName := "AccountsSystem"
+		natsSubject := "some-subject"
+		// create the instance of provider
+		provider := NewDiscovery(WithLogger(log.DiscardLogger))
+
+		// create the config
+		config := discovery.Config{
+			ApplicationName: applicationName,
+			ActorSystemName: actorSystemName,
+			NatsServer:      natsServer,
+			NatsSubject:     natsSubject,
+		}
+
+		// set config
+		err := provider.SetConfig(config)
+		require.NoError(t, err)
+
+		// initialize
+		err = provider.Initialize()
+		assert.NoError(t, err)
+
+		// stop the NATS server
+		t.Cleanup(srv.Shutdown)
+	})
+	t.Run("With Initialize: already initialized", func(t *testing.T) {
+		// create the instance of provider
+		provider := NewDiscovery()
+		provider.initialized = atomic.NewBool(true)
+		assert.Error(t, provider.Initialize())
+	})
+	t.Run("With Initialize: with host config not set", func(t *testing.T) {
+		// create the instance of provider
+		provider := NewDiscovery()
+		assert.Error(t, provider.Initialize())
+	})
+	t.Run("With Register: already registered", func(t *testing.T) {
+		// create the instance of provider
+		provider := NewDiscovery()
+		provider.registered = atomic.NewBool(true)
+		err := provider.Register()
+		assert.Error(t, err)
+		assert.EqualError(t, err, discovery.ErrAlreadyRegistered.Error())
+	})
+	t.Run("With Deregister: already not registered", func(t *testing.T) {
+		// create the instance of provider
+		provider := NewDiscovery()
+		err := provider.Deregister()
+		assert.Error(t, err)
+		assert.EqualError(t, err, discovery.ErrNotRegistered.Error())
+	})
+	t.Run("With DiscoverPeers", func(t *testing.T) {
+		// start the NATS server
+		srv := startNatsServer(t)
+		// create two peers
+		client1 := newPeer(t, srv.Addr().String())
+		client2 := newPeer(t, srv.Addr().String())
+
+		// no discovery is allowed unless registered
+		peers, err := client1.DiscoverPeers()
+		require.Error(t, err)
+		assert.EqualError(t, err, discovery.ErrNotRegistered.Error())
+		require.Empty(t, peers)
+
+		peers, err = client2.DiscoverPeers()
+		require.Error(t, err)
+		assert.EqualError(t, err, discovery.ErrNotRegistered.Error())
+		require.Empty(t, peers)
+
+		// register client 2
+		require.NoError(t, client2.Register())
+		peers, err = client2.DiscoverPeers()
+		require.NoError(t, err)
+		require.Empty(t, peers)
+
+		// register client 1
+		require.NoError(t, client1.Register())
+		peers, err = client1.DiscoverPeers()
+		require.NoError(t, err)
+		require.NotEmpty(t, peers)
+		require.Len(t, peers, 1)
+		discoveredNodeAddr := client2.hostNode.GossipAddress()
+		require.Equal(t, peers[0], discoveredNodeAddr)
+
+		// discover more peers from client 2
+		peers, err = client2.DiscoverPeers()
+		require.NoError(t, err)
+		require.NotEmpty(t, peers)
+		require.Len(t, peers, 1)
+		discoveredNodeAddr = client1.hostNode.GossipAddress()
+		require.Equal(t, peers[0], discoveredNodeAddr)
+
+		// de-register client 2
+		require.NoError(t, client2.Deregister())
+		peers, err = client2.DiscoverPeers()
+		require.Error(t, err)
+		assert.EqualError(t, err, discovery.ErrNotRegistered.Error())
+		require.Empty(t, peers)
+
+		// client-1 cannot see the deregistered client
+		peers, err = client1.DiscoverPeers()
+		require.NoError(t, err)
+		require.Empty(t, peers)
+
+		require.NoError(t, client1.Close())
+		require.NoError(t, client2.Close())
+
+		// stop the NATS server
+		t.Cleanup(srv.Shutdown)
+	})
+	t.Run("With DiscoverPeers: not initialized", func(t *testing.T) {
+		provider := NewDiscovery()
+		peers, err := provider.DiscoverPeers()
+		assert.Error(t, err)
+		assert.Empty(t, peers)
+		assert.EqualError(t, err, discovery.ErrNotInitialized.Error())
+	})
+}
diff --git a/discovery/nats/option_test.go b/discovery/nats/option_test.go
new file mode 100644
index 0000000..25252ad
--- /dev/null
+++ b/discovery/nats/option_test.go
@@ -0,0 +1,54 @@
+/*
+ * MIT License
+ *
+ * Copyright (c) 2022-2023 Tochemey
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package nats
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/tochemey/goakt/log"
+)
+
+func TestOptions(t *testing.T) {
+	testCases := []struct {
+		name     string
+		option   Option
+		expected Discovery
+	}{
+		{
+			name:     "WithLogger",
+			option:   WithLogger(log.DefaultLogger),
+			expected: Discovery{logger: log.DefaultLogger},
+		},
+	}
+
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			var discovery Discovery
+			tc.option.Apply(&discovery)
+			assert.Equal(t, tc.expected, discovery)
+		})
+	}
+}
diff --git a/cluster/node_test.go b/discovery/node_test.go
similarity index 96%
rename from cluster/node_test.go
rename to discovery/node_test.go
index b2fb33a..68622f5 100644
--- a/cluster/node_test.go
+++ b/discovery/node_test.go
@@ -22,7 +22,7 @@
  * SOFTWARE.
  */
 
-package cluster
+package discovery
 
 import (
 	"fmt"
@@ -49,7 +49,7 @@ func TestGetHostNode(t *testing.T) {
 		t.Setenv("NODE_NAME", "testNode")
 		t.Setenv("NODE_IP", host)
 
-		node, err := hostNode()
+		node, err := HostNode()
 		require.NoError(t, err)
 		require.NotNil(t, node)
 		clusterAddr := node.ClusterAddress()
@@ -58,7 +58,7 @@ func TestGetHostNode(t *testing.T) {
 		assert.Equal(t, fmt.Sprintf("%s:%d", host, gossipPort), gossipAddr)
 	})
 	t.Run("With host node env vars not set", func(t *testing.T) {
-		node, err := hostNode()
+		node, err := HostNode()
 		require.Error(t, err)
 		require.Nil(t, node)
 	})
diff --git a/testkit/discovery/provider.go b/testkit/discovery/provider.go
index a6c0b5a..2a216ca 100644
--- a/testkit/discovery/provider.go
+++ b/testkit/discovery/provider.go
@@ -20,6 +20,47 @@ func (_m *Provider) EXPECT() *Provider_Expecter {
 	return &Provider_Expecter{mock: &_m.Mock}
 }
 
+// Close provides a mock function with given fields:
+func (_m *Provider) Close() error {
+	ret := _m.Called()
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func() error); ok {
+		r0 = rf()
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// Provider_Close_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Close'
+type Provider_Close_Call struct {
+	*mock.Call
+}
+
+// Close is a helper method to define mock.On call
+func (_e *Provider_Expecter) Close() *Provider_Close_Call {
+	return &Provider_Close_Call{Call: _e.mock.On("Close")}
+}
+
+func (_c *Provider_Close_Call) Run(run func()) *Provider_Close_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run()
+	})
+	return _c
+}
+
+func (_c *Provider_Close_Call) Return(_a0 error) *Provider_Close_Call {
+	_c.Call.Return(_a0)
+	return _c
+}
+
+func (_c *Provider_Close_Call) RunAndReturn(run func() error) *Provider_Close_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
 // Deregister provides a mock function with given fields:
 func (_m *Provider) Deregister() error {
 	ret := _m.Called()