func (pubsub *PubSub) neighbours() int64 {
	return int64(len(pubsub.knownNeighbours))
}
neighbours return the total number of known neighbours.
package pubsub
import (
	"errors"
	"fmt"
	"math"
	"os"
	"strings"
	"sync"
	"time"
	"github.com/moleculer-go/moleculer/version"
	"github.com/moleculer-go/moleculer/payload"
	"github.com/moleculer-go/moleculer/context"
	"github.com/moleculer-go/moleculer/transit"
	"github.com/moleculer-go/moleculer/transit/memory"
	"github.com/moleculer-go/moleculer/transit/nats"
	"github.com/moleculer-go/moleculer/util"
	"github.com/moleculer-go/moleculer"
	"github.com/moleculer-go/moleculer/serializer"
	log "github.com/sirupsen/logrus"
)
type PubSub struct {
	logger               *log.Entry
	transport            transit.Transport
	broker               *moleculer.BrokerDelegates
	isConnected          bool
	pendingRequests      map[string]pendingRequest
	pendingRequestsMutex *sync.Mutex
	serializer           serializer.Serializer
	knownNeighbours   map[string]int64
	neighboursTimeout time.Duration
	neighboursMutex   *sync.Mutex
	brokerStarted     bool
}
const DATATYPE_UNDEFINED = 0
const DATATYPE_NULL = 1
const DATATYPE_JSON = 2
const DATATYPE_BUFFER = 3
func (pubsub *PubSub) onServiceAdded(values ...interface{}) {
	if pubsub.isConnected && pubsub.brokerStarted {
		localNodeID := pubsub.broker.LocalNode().GetID()
		isLocalServiceAdded := false
		for _, value := range values {
			if value.(map[string]string)["nodeID"] == localNodeID {
				isLocalServiceAdded = true
				break
			}
		}
		if isLocalServiceAdded {
			pubsub.broker.LocalNode().IncreaseSequence()
			pubsub.broadcastNodeInfo("")
		}
	}
}
func (pubsub *PubSub) onBrokerStarted(values ...interface{}) {
	if pubsub.isConnected {
		pubsub.broadcastNodeInfo("")
		pubsub.brokerStarted = true
	}
}
func Create(broker *moleculer.BrokerDelegates) transit.Transit {
	pendingRequests := make(map[string]pendingRequest)
	knownNeighbours := make(map[string]int64)
	transitImpl := PubSub{
		broker:               broker,
		isConnected:          false,
		pendingRequests:      pendingRequests,
		logger:               broker.Logger("Transit", ""),
		serializer:           serializer.New(broker),
		neighboursTimeout:    broker.Config.NeighboursCheckTimeout,
		knownNeighbours:      knownNeighbours,
		neighboursMutex:      &sync.Mutex{},
		pendingRequestsMutex: &sync.Mutex{},
	}
	broker.Bus().On("$node.disconnected", transitImpl.onNodeDisconnected)
	broker.Bus().On("$node.connected", transitImpl.onNodeConnected)
	broker.Bus().On("$broker.started", transitImpl.onBrokerStarted)
	broker.Bus().On("$registry.service.added", transitImpl.onServiceAdded)
	return &transitImpl
}
func (pubsub *PubSub) pendingRequestsByNode(nodeId string) []pendingRequest {
	list := []pendingRequest{}
	for _, p := range pubsub.pendingRequests {
		if p.context.TargetNodeID() == nodeId {
			list = append(list, p)
		}
	}
	return list
}
func (pubsub *PubSub) requestTimedOut(resultChan *chan moleculer.Payload, context moleculer.BrokerContext) func() {
	pError := payload.New(errors.New("request timeout"))
	return func() {
		pubsub.logger.Debug("requestTimedOut() nodeID: ", context.TargetNodeID())
		pubsub.pendingRequestsMutex.Lock()
		defer pubsub.pendingRequestsMutex.Unlock()
		p, exists := pubsub.pendingRequests[context.ID()]
		if exists {
			(*p.resultChan) <- pError
			p.timer.Stop()
			delete(pubsub.pendingRequests, p.context.ID())
		}
	}
}
func (pubsub *PubSub) onNodeDisconnected(values ...interface{}) {
	pubsub.pendingRequestsMutex.Lock()
	var nodeID string = values[0].(string)
	pending := pubsub.pendingRequestsByNode(nodeID)
	pubsub.logger.Debug("onNodeDisconnected() nodeID: ", nodeID, " pending: ", len(pending))
	if len(pending) > 0 {
		pError := payload.New(fmt.Errorf("Node %s disconnected. The request was canceled.", nodeID))
		for _, p := range pending {
			(*p.resultChan) <- pError
			p.timer.Stop()
			delete(pubsub.pendingRequests, p.context.ID())
		}
	}
	pubsub.pendingRequestsMutex.Unlock()
	pubsub.neighboursMutex.Lock()
	delete(pubsub.knownNeighbours, nodeID)
	pubsub.neighboursMutex.Unlock()
}
func (pubsub *PubSub) onNodeConnected(values ...interface{}) {
	nodeID := values[0].(string)
	neighbours := values[1].(int64)
	pubsub.logger.Debug("onNodeConnected() nodeID: ", nodeID, " neighbours: ", neighbours)
	pubsub.neighboursMutex.Lock()
	pubsub.knownNeighbours[nodeID] = neighbours
	pubsub.neighboursMutex.Unlock()
}
func isNats(v string) bool {
	return strings.Index(v, "nats://") > -1
}
func (pubsub *PubSub) createTransport() transit.Transport {
	var transport transit.Transport
	if pubsub.broker.Config.TransporterFactory != nil {
		pubsub.logger.Info("Transporter: Custom factory")
		transport = pubsub.broker.Config.TransporterFactory().(transit.Transport)
	} else if pubsub.broker.Config.Transporter == "STAN" {
		pubsub.logger.Info("Transporter: NatsStreamingTransporter")
		transport = pubsub.createStanTransporter()
	} else if isNats(pubsub.broker.Config.Transporter) {
		pubsub.logger.Info("Transporter: NatsTransporter")
		transport = pubsub.createNatsTransporter()
	} else {
		pubsub.logger.Info("Transporter: Memory")
		transport = pubsub.createMemoryTransporter()
	}
	transport.SetPrefix(resolveNamespace(pubsub.broker.Config.Namespace))
	transport.SetNodeID(pubsub.broker.LocalNode().GetID())
	transport.SetSerializer(pubsub.serializer)
	return transport
}
func resolveNamespace(namespace string) string {
	if namespace != "" {
		return "MOL-" + namespace
	}
	return "MOL"
}
func (pubsub *PubSub) createMemoryTransporter() transit.Transport {
	pubsub.logger.Debug("createMemoryTransporter() ... ")
	logger := pubsub.logger.WithField("transport", "memory")
	mem := memory.Create(logger, &memory.SharedMemory{})
	return &mem
}
func (pubsub *PubSub) createNatsTransporter() transit.Transport {
	pubsub.logger.Debug("createNatsTransporter()")
	return nats.CreateNatsTransporter(nats.NATSOptions{
		URL:            pubsub.broker.Config.Transporter,
		Name:           pubsub.broker.LocalNode().GetID(),
		Logger:         pubsub.logger.WithField("transport", "nats"),
		Serializer:     pubsub.serializer,
		AllowReconnect: true,
		ReconnectWait:  time.Second * 2,
		MaxReconnect:   -1,
	})
}
func (pubsub *PubSub) createStanTransporter() transit.Transport {
	broker := pubsub.broker
	logger := broker.Logger("transport", "stan")
	url := "stan://" + os.Getenv("STAN_HOST") + ":4222"
	clusterID := "test-cluster"
	localNodeID := broker.LocalNode().GetID()
	clientID := strings.ReplaceAll(localNodeID, ".", "_")
	options := nats.StanOptions{
		url,
		clusterID,
		clientID,
		logger,
		pubsub.serializer,
		func(message moleculer.Payload) bool {
			sender := message.Get("sender").String()
			return sender != localNodeID
		},
	}
	stanTransporter := nats.CreateStanTransporter(options)
	var transport transit.Transport = &stanTransporter
	return transport
}
type pendingRequest struct {
	context    moleculer.BrokerContext
	resultChan *chan moleculer.Payload
	timer      *time.Timer
}
func (pubsub *PubSub) checkMaxQueueSize() {
}
func (pubsub *PubSub) waitForNeighbours() bool {
	if pubsub.broker.Config.DontWaitForNeighbours {
		return true
	}
	start := time.Now()
	for {
		expected := pubsub.expectedNeighbours()
		neighbours := pubsub.neighbours()
		if expected <= neighbours && (expected > 0 || neighbours > 0) {
			pubsub.logger.Debug("waitForNeighbours() - received info from all expected neighbours :) -> expected: ", expected)
			return true
		}
		if time.Since(start) > pubsub.neighboursTimeout {
			pubsub.logger.Warn("waitForNeighbours() - Time out ! did not receive info from all expected neighbours: ", expected, "  INFOs received: ", neighbours)
			return false
		}
		if !pubsub.isConnected {
			return false
		}
		time.Sleep(pubsub.broker.Config.WaitForNeighboursInterval)
	}
}
func (pubsub *PubSub) DiscoverNodes() chan bool {
	result := make(chan bool)
	go func() {
		pubsub.DiscoverNode("")
		result <- pubsub.waitForNeighbours()
	}()
	return result
}
func (pubsub *PubSub) SendHeartbeat() {
	node := pubsub.broker.LocalNode().ExportAsMap()
	payload := map[string]interface{}{
		"sender": node["id"],
		"cpu":    node["cpu"],
		"cpuSeq": node["cpuSeq"],
		"ver":    version.MoleculerProtocol(),
	}
	message, err := pubsub.serializer.MapToPayload(&payload)
	if err == nil {
		pubsub.transport.Publish("HEARTBEAT", "", message)
	}
}
func (pubsub *PubSub) DiscoverNode(nodeID string) {
	payload := map[string]interface{}{
		"sender": pubsub.broker.LocalNode().GetID(),
		"ver":    version.MoleculerProtocol(),
	}
	message, err := pubsub.serializer.MapToPayload(&payload)
	if err == nil {
		pubsub.transport.Publish("DISCOVER", nodeID, message)
	}
}
func (pubsub *PubSub) Emit(context moleculer.BrokerContext) {
	targetNodeID := context.TargetNodeID()
	payload := context.AsMap()
	payload["sender"] = pubsub.broker.LocalNode().GetID()
	payload["ver"] = version.MoleculerProtocol()
	if context.Payload().Exists() {
		payload["dataType"] = DATATYPE_JSON
	} else {
		payload["dataType"] = DATATYPE_NULL
	}
	pubsub.logger.Trace("Emit() targetNodeID: ", targetNodeID, " payload: ", payload)
	message, err := pubsub.serializer.MapToPayload(&payload)
	if err != nil {
		pubsub.logger.Error("Emit() Error serializing the payload: ", payload, " error: ", err)
		panic(fmt.Errorf("Error trying to serialize the payload. Likely issues with the action params. Error: %s", err))
	}
	pubsub.transport.Publish("EVENT", targetNodeID, message)
}
func (pubsub *PubSub) Request(context moleculer.BrokerContext) chan moleculer.Payload {
	pubsub.checkMaxQueueSize()
	resultChan := make(chan moleculer.Payload)
	targetNodeID := context.TargetNodeID()
	payload := context.AsMap()
	payload["sender"] = pubsub.broker.LocalNode().GetID()
	payload["ver"] = version.MoleculerProtocol()
	if context.Payload().Exists() {
		payload["paramsType"] = DATATYPE_JSON
	} else {
		payload["paramsType"] = DATATYPE_NULL
	}
	pubsub.logger.Trace("Request() targetNodeID: ", targetNodeID, " payload: ", payload)
	message, err := pubsub.serializer.MapToPayload(&payload)
	if err != nil {
		pubsub.logger.Error("Request() Error serializing the payload: ", payload, " error: ", err)
		panic(fmt.Errorf("Error trying to serialize the payload. Likely issues with the action params. Error: %s", err))
	}
	pubsub.pendingRequestsMutex.Lock()
	pubsub.logger.Debug("Request() pending request id: ", context.ID(), " targetNodeId: ", context.TargetNodeID())
	pubsub.pendingRequests[context.ID()] = pendingRequest{
		context,
		&resultChan,
		time.AfterFunc(
			pubsub.broker.Config.RequestTimeout,
			pubsub.requestTimedOut(&resultChan, context)),
	}
	pubsub.pendingRequestsMutex.Unlock()
	pubsub.transport.Publish("REQ", targetNodeID, message)
	return resultChan
}
func (pubsub *PubSub) validate(handler func(message moleculer.Payload)) transit.TransportHandler {
	return func(msg moleculer.Payload) {
		valid := pubsub.validateVersion(msg) && !pubsub.sameHost(msg)
		if valid {
			handler(msg)
		} else {
			pubsub.logger.Trace("Discarding invalid msg -> ", msg.Value())
		}
	}
}
func (pubsub *PubSub) sameHost(msg moleculer.Payload) bool {
	sender := msg.Get("sender").String()
	localNodeID := pubsub.broker.LocalNode().GetID()
	return sender == localNodeID
}
func (pubsub *PubSub) validateVersion(msg moleculer.Payload) bool {
	msgVersion := msg.Get("ver").String()
	if msgVersion == version.MoleculerProtocol() {
		return true
	} else {
		pubsub.logger.Error("Discarding msg - wronging version: ", msgVersion, " expected: ", version.MoleculerProtocol(), " msg: ", msg)
		return false
	}
}
func (pubsub *PubSub) reponseHandler() transit.TransportHandler {
	return func(message moleculer.Payload) {
		pubsub.pendingRequestsMutex.Lock()
		defer pubsub.pendingRequestsMutex.Unlock()
		id := message.Get("id").String()
		sender := message.Get("sender").String()
		pubsub.logger.Debug("reponseHandler() - response arrived from nodeID: ", sender, " context id: ", id)
		request, exists := pubsub.pendingRequests[id]
		if !exists {
			pubsub.logger.Debug("reponseHandler() - discarding response -> request does not exist for id: ", id, " - message: ", message.Value())
			return
		}
		if request.resultChan == nil {
			pubsub.logger.Debug("reponseHandler() - discarding response -> request.resultChan is nil! - message: ", message.Value(), " pending context: ", request.context)
			return
		}
		request.timer.Stop()
		defer delete(pubsub.pendingRequests, id)
		var result moleculer.Payload
		if message.Get("success").Bool() {
			result = message.Get("data")
		} else {
			result = pubsub.parseError(message)
		}
		pubsub.logger.Trace("reponseHandler() id: ", id, " result: ", result)
		(*request.resultChan) <- result
	}
}
func (pubsub *PubSub) parseError(message moleculer.Payload) moleculer.Payload {
	if pubsub.isMoleculerJSError(message) {
		return payload.New(pubsub.moleculerJSError(message))
	}
	return payload.New(errors.New(message.Get("error").String()))
}
func (pubsub *PubSub) isMoleculerJSError(message moleculer.Payload) bool {
	return message.Get("error").Get("message").Exists()
}
func (pubsub *PubSub) moleculerJSError(message moleculer.Payload) error {
	msg := message.Get("error").Get("message").String()
	if message.Get("error").Get("stack").Exists() {
		pubsub.logger.Error(message.Get("error").Get("stack").Value())
	}
	return errors.New(msg)
}
func (pubsub *PubSub) sendResponse(context moleculer.BrokerContext, response moleculer.Payload) {
	targetNodeID := context.TargetNodeID()
	if targetNodeID == "" {
		panic(errors.New("sendResponse() targetNodeID is required !"))
	}
	pubsub.logger.Tracef("sendResponse() reponse type: %T ", response)
	values := make(map[string]interface{})
	values["sender"] = pubsub.broker.LocalNode().GetID()
	values["ver"] = version.MoleculerProtocol()
	values["id"] = context.ID()
	values["meta"] = context.Meta()
	if response.Exists() {
		values["dataType"] = DATATYPE_JSON
	} else {
		values["dataType"] = DATATYPE_NULL
	}
	if response.IsError() {
		var errMap map[string]string
		actionError, isActionError := response.Value().(ActionError)
		if isActionError {
			errMap = map[string]string{
				"message": actionError.Error(),
				"stack":   actionError.Stack(),
				"name":    "Error",
			}
		} else {
			errMap = map[string]string{
				"message": response.String(),
				"name":    "Error",
			}
		}
		values["success"] = false
		values["error"] = errMap
	} else {
		values["success"] = true
		values["data"] = response.Value()
	}
	message, err := pubsub.serializer.MapToPayload(&values)
	if err != nil {
		pubsub.logger.Error("sendResponse() Erro serializing the values: ", values, " error: ", err)
		panic(err)
	}
	pubsub.logger.Trace("sendResponse() targetNodeID: ", targetNodeID, " values: ", values, " message: ", message)
	pubsub.transport.Publish("RES", targetNodeID, message)
}
type ActionError interface {
	Error() string
	Stack() string
}
func parseParamsType(value moleculer.Payload) string {
	if !value.Exists() {
		return "1"  
	}
	return value.String()
}
func (pubsub *PubSub) requestHandler() transit.TransportHandler {
	return func(message moleculer.Payload) {
		paramsType := parseParamsType(message.Get("paramsType"))
		if paramsType != "1" && paramsType != "2" {
			errMsg := "Expecting paramsType == 2 (JSON) or 1 (Null) - received: " + paramsType
			pubsub.logger.Error(errMsg)
			pubsub.sendResponse(context.ActionContext(pubsub.broker, nil), payload.Error(errMsg))
			return
		}
		values := pubsub.serializer.PayloadToContextMap(message)
		context := context.ActionContext(pubsub.broker, values)
		result := <-pubsub.broker.ActionDelegate(context)
		pubsub.sendResponse(context, result)
	}
}
func (pubsub *PubSub) eventHandler() transit.TransportHandler {
	return func(message moleculer.Payload) {
		values := pubsub.serializer.PayloadToContextMap(message)
		context := context.EventContext(pubsub.broker, values)
		pubsub.broker.HandleRemoteEvent(context)
	}
}
func (pubsub *PubSub) expectedNeighbours() int64 {
	neighbours := pubsub.neighbours()
	if neighbours == 0 {
		return 0
	}
	var total int64
	pubsub.neighboursMutex.Lock()
	for _, value := range pubsub.knownNeighbours {
		total = total + value
	}
	pubsub.neighboursMutex.Unlock()
	return total / neighbours
}