feat (stop cycle) broker, registry and trasit stop lifecycle

broker/broker.go

@@ -22,13 +22,14 @@ import (
 )
 
 var defaultConfig = moleculer.BrokerConfig{
-	LogLevel:              "INFO",
-	LogFormat:             "TEXT",
-	DiscoverNodeID:        DiscoverNodeID,
-	Transporter:           "MEMORY",
-	HeartbeatFrequency:    5 * time.Second,
-	HeartbeatTimeout:      30 * time.Second,
-	OfflineCheckFrequency: 20 * time.Second,
+	LogLevel:               "INFO",
+	LogFormat:              "TEXT",
+	DiscoverNodeID:         DiscoverNodeID,
+	Transporter:            "MEMORY",
+	HeartbeatFrequency:     5 * time.Second,
+	HeartbeatTimeout:       30 * time.Second,
+	OfflineCheckFrequency:  20 * time.Second,
+	NeighboursCheckTimeout: 2 * time.Second,
 }
 
 // DiscoverNodeID - should return the node id for this machine

moleculer.go

@@ -63,13 +63,14 @@ type Mixin struct {
 }
 
 type BrokerConfig struct {
-	LogLevel              string
-	LogFormat             string
-	DiscoverNodeID        func() string
-	Transporter           string
-	HeartbeatFrequency    time.Duration
-	HeartbeatTimeout      time.Duration
-	OfflineCheckFrequency time.Duration
+	LogLevel               string
+	LogFormat              string
+	DiscoverNodeID         func() string
+	Transporter            string
+	HeartbeatFrequency     time.Duration
+	HeartbeatTimeout       time.Duration
+	OfflineCheckFrequency  time.Duration
+	NeighboursCheckTimeout time.Duration
 }
 
 type ActionHandler func(context Context, params Params) interface{}

registry/registry.go

@@ -100,6 +100,8 @@ func (registry *ServiceRegistry) Stop() {
 	registry.logger.Debug("Registry Stop() ")
 	registry.stoping = true
 	<-registry.transit.Disconnect()
+	registry.logger.Debug("Transit Disconnected -> Registry Full Stop!")
+
 }
 
 // Start : start the registry background processes.

transit/nats/stan.go

@@ -67,10 +67,21 @@ func (transporter *StanTransporter) Connect() chan bool {
 func (transporter *StanTransporter) Disconnect() chan bool {
 	endChan := make(chan bool)
 	go func() {
+		transporter.logger.Debug("Disconnect() # of subscriptions: ", len(transporter.subscriptions))
 		for _, sub := range transporter.subscriptions {
-			sub.Unsubscribe()
+			error := sub.Unsubscribe()
+			if error != nil {
+				transporter.logger.Error("Disconnect() error when unsubscribing stan subscription: ", error)
+			}
 		}
-		transporter.connection.Close()
+		transporter.logger.Debug("Disconnect() subscriptions unsubscribed.")
+		error := transporter.connection.Close()
+		if error == nil {
+			transporter.logger.Debug("Disconnect() stan connection closed :)")
+		} else {
+			transporter.logger.Error("Disconnect() error when closing stan connection :( ", error)
+		}
+
 		transporter.connection = nil
 		endChan <- true
 	}()

transit/nats/stan_test.go

@@ -1,6 +1,8 @@
 package nats_test
 
 import (
+	"fmt"
+
 	"github.com/moleculer-go/moleculer/util"
 	. "github.com/onsi/ginkgo"
 	. "github.com/onsi/gomega"
@@ -102,25 +104,25 @@ var _ = Describe("Transit", func() {
 	}
 
 	It("Should create multiple brokers, connect and disconnect and make sure stan resources are closed/released properly.", func() {
-
+		logLevel := "DEBUG"
 		brokersLoop := 100
 		for i := 0; i < brokersLoop; i++ {
 			brokr1 := broker.FromConfig(&moleculer.BrokerConfig{
-				LogLevel:    "INFO",
+				LogLevel:    logLevel,
 				Transporter: "STAN",
 			})
 			addUserService(brokr1)
 			brokr1.Start()
 
 			brokr2 := broker.FromConfig(&moleculer.BrokerConfig{
-				LogLevel:    "INFO",
+				LogLevel:    logLevel,
 				Transporter: "STAN",
 			})
 			addContactService(brokr2)
 			brokr2.Start()
 
 			brokr3 := broker.FromConfig(&moleculer.BrokerConfig{
-				LogLevel:    "INFO",
+				LogLevel:    logLevel,
 				Transporter: "STAN",
 			})
 			addProfileService(brokr3)
@@ -131,6 +133,8 @@ var _ = Describe("Transit", func() {
 			Expect(len(newList)).Should(Equal(arraySize + 6))
 
 			stopBrokers(brokr1, brokr2, brokr3)
+
+			fmt.Println("**** One More Loop -> Total: ", i)
 		}
 
 	})

transit/pubsub/pubsub.go

@@ -46,7 +46,7 @@ func Create(broker moleculer.BrokerDelegates) transit.Transit {
 		pendingRequests:      pendingRequests,
 		logger:               broker.Logger("Transit", ""),
 		serializer:           serializer.FromConfig(broker),
-		neighboursTimeout:    1 * time.Second,
+		neighboursTimeout:    broker.Config.NeighboursCheckTimeout,
 		knownNeighbours:      knownNeighbours,
 		neighboursMutex:      &sync.Mutex{},
 		pendingRequestsMutex: &sync.Mutex{},
@@ -150,7 +150,7 @@ func (pubsub *PubSub) waitForNeighbours() bool {
 		expected := pubsub.expectedNeighbours()
 		neighbours := pubsub.neighbours()
 		if expected <= neighbours && expected > 0 && neighbours > 0 {
-			pubsub.logger.Info("waitForNeighbours() - received INDO from all expected neighbours: ", expected)
+			pubsub.logger.Info("waitForNeighbours() - received info from all expected neighbours :) -> expected: ", expected)
 			return true
 		}
 		if time.Since(start) > pubsub.neighboursTimeout {
@@ -395,9 +395,8 @@ func (pubsub *PubSub) Disconnect() chan bool {
 	}
 	pubsub.logger.Info("PubSub - Disconnecting transport...")
 	pubsub.sendDisconnect()
-	endChan = pubsub.transport.Disconnect()
 	pubsub.isConnected = false
-	return endChan
+	return pubsub.transport.Disconnect()
 }
 
 // Connect : connect the transit with the transporter, subscribe to all events and start publishing its node info
@@ -411,7 +410,7 @@ func (pubsub *PubSub) Connect() chan bool {
 	pubsub.transport = pubsub.createTransport(pubsub.broker)
 	go func() {
 		pubsub.isConnected = <-pubsub.transport.Connect()
-		pubsub.logger.Debug("Transport Connected!")
+		pubsub.logger.Debug("PubSub - Transport Connected!")
 		if pubsub.isConnected {
 			pubsub.subscribe()
 		}