func (broker *ServiceBroker) MCall(callMaps map[string]map[string]interface{}) chan map[string]moleculer.Payload {
	result := make(chan map[string]moleculer.Payload, 1)
	go broker.invokeMCalls(callMaps, result)
	return result
}
MCall perform multiple calls and return all results together in a nice map indexed by name.
package broker
import (
	"errors"
	"strings"
	"time"
	"github.com/hashicorp/go-uuid"
	bus "github.com/moleculer-go/goemitter"
	"github.com/moleculer-go/moleculer"
	"github.com/moleculer-go/moleculer/cache"
	"github.com/moleculer-go/moleculer/context"
	"github.com/moleculer-go/moleculer/metrics"
	"github.com/moleculer-go/moleculer/middleware"
	"github.com/moleculer-go/moleculer/payload"
	"github.com/moleculer-go/moleculer/registry"
	"github.com/moleculer-go/moleculer/serializer"
	"github.com/moleculer-go/moleculer/service"
	log "github.com/sirupsen/logrus"
)
func mergeMaps(base, new map[string]interface{}) map[string]interface{} {
	if base == nil {
		base = map[string]interface{}{}
	}
	for key, value := range new {
		base[key] = value
	}
	return base
}
func mergeConfigs(baseConfig moleculer.Config, userConfig []*moleculer.Config) moleculer.Config {
	if len(userConfig) > 0 {
		for _, config := range userConfig {
			if config.Services != nil {
				baseConfig.Services = mergeMaps(baseConfig.Services, config.Services)
			}
			if config.LogLevel != "" {
				baseConfig.LogLevel = config.LogLevel
			}
			if config.LogFormat != "" {
				baseConfig.LogFormat = config.LogFormat
			}
			if config.DiscoverNodeID != nil {
				baseConfig.DiscoverNodeID = config.DiscoverNodeID
			}
			if config.Transporter != "" {
				baseConfig.Transporter = config.Transporter
			}
			if config.TransporterFactory != nil {
				baseConfig.TransporterFactory = config.TransporterFactory
			}
			if config.StrategyFactory != nil {
				baseConfig.StrategyFactory = config.StrategyFactory
			}
			if config.DisableInternalMiddlewares {
				baseConfig.DisableInternalMiddlewares = config.DisableInternalMiddlewares
			}
			if config.DisableInternalServices {
				baseConfig.DisableInternalServices = config.DisableInternalServices
			}
			if config.Metrics {
				baseConfig.Metrics = config.Metrics
			}
			if config.MetricsRate > 0 {
				baseConfig.MetricsRate = config.MetricsRate
			}
			if config.DontWaitForNeighbours {
				baseConfig.DontWaitForNeighbours = config.DontWaitForNeighbours
			}
			if config.Middlewares != nil {
				baseConfig.Middlewares = config.Middlewares
			}
			if config.RequestTimeout != 0 {
				baseConfig.RequestTimeout = config.RequestTimeout
			}
			if config.Namespace != "" {
				baseConfig.Namespace = config.Namespace
			}
		}
	}
	return baseConfig
}
type ServiceBroker struct {
	namespace string
	logger *log.Entry
	localBus *bus.Emitter
	registry *registry.ServiceRegistry
	middlewares *middleware.Dispatch
	cache cache.Cache
	serializer *serializer.Serializer
	services []*service.Service
	started  bool
	starting bool
	rootContext moleculer.BrokerContext
	config moleculer.Config
	delegates *moleculer.BrokerDelegates
	id string
	instanceID string
	localNode moleculer.Node
}
func (broker *ServiceBroker) LocalBus() *bus.Emitter {
	return broker.localBus
}
func (broker *ServiceBroker) stopService(svc *service.Service) {
	broker.middlewares.CallHandlers("serviceStopping", svc)
	svc.Stop(broker.rootContext.ChildActionContext("service.stop", payload.Empty()))
	broker.middlewares.CallHandlers("serviceStopped", svc)
}
func (broker *ServiceBroker) applyServiceConfig(svc *service.Service) {
	if bkrConfig, exists := broker.config.Services[svc.Name()]; exists {
		svcConfig, ok := bkrConfig.(map[string]interface{})
		if ok {
			_, ok := svcConfig["settings"]
			if ok {
				settings, ok := svcConfig["settings"].(map[string]interface{})
				if ok {
					svc.AddSettings(settings)
				} else {
					broker.logger.Error("Could not add service settings - Error converting the input settings to map[string]interface{} - Invalid format! Service Config : ", svcConfig)
				}
			}
			_, ok = svcConfig["metadata"]
			if ok {
				metadata, ok := svcConfig["metadata"].(map[string]interface{})
				if ok {
					svc.AddSettings(metadata)
				} else {
					broker.logger.Error("Could not add service metadata - Error converting the input metadata to map[string]interface{} - Invalid format! Service Config : ", svcConfig)
				}
			}
		} else {
			broker.logger.Error("Could not apply service configuration - Error converting the service config to map[string]interface{} - Invalid format! Broker Config : ", bkrConfig)
		}
	}
}
func (broker *ServiceBroker) startService(svc *service.Service) {
	broker.logger.Debug("Broker start service: ", svc.FullName())
	broker.applyServiceConfig(svc)
	broker.middlewares.CallHandlers("serviceStarting", svc)
	broker.waitForDependencies(svc)
	broker.registry.AddLocalService(svc)
	broker.middlewares.CallHandlers("serviceStarted", svc)
	svc.Start(broker.rootContext.ChildActionContext("service.start", payload.Empty()))
}
func (broker *ServiceBroker) waitForDependencies(service *service.Service) {
	if len(service.Dependencies()) == 0 {
		return
	}
	start := time.Now()
	for {
		if !broker.started {
			break
		}
		found := true
		for _, dependency := range service.Dependencies() {
			known := broker.registry.KnowService(dependency)
			if !known {
				found = false
				break
			}
		}
		if found {
			broker.logger.Debug("waitForDependencies() - All dependencies were found :) -> service: ", service.Name(), " wait For Dependencies: ", service.Dependencies())
			break
		}
		if time.Since(start) > broker.config.WaitForDependenciesTimeout {
			broker.logger.Warn("waitForDependencies() - Time out ! service: ", service.Name(), " wait For Dependencies: ", service.Dependencies())
			break
		}
		time.Sleep(time.Microsecond)
	}
}
func (broker *ServiceBroker) broadcastLocal(eventName string, params ...interface{}) {
	broker.LocalBus().EmitAsync(eventName, params)
}
func (broker *ServiceBroker) createBrokerLogger() *log.Entry {
	if strings.ToUpper(broker.config.LogFormat) == "JSON" {
		log.SetFormatter(&log.JSONFormatter{})
	} else {
		log.SetFormatter(&log.TextFormatter{})
	}
	if strings.ToUpper(broker.config.LogLevel) == "WARN" {
		log.SetLevel(log.WarnLevel)
	} else if strings.ToUpper(broker.config.LogLevel) == "DEBUG" {
		log.SetLevel(log.DebugLevel)
	} else if strings.ToUpper(broker.config.LogLevel) == "TRACE" {
		log.SetLevel(log.TraceLevel)
	} else if strings.ToUpper(broker.config.LogLevel) == "ERROR" {
		log.SetLevel(log.ErrorLevel)
	} else if strings.ToUpper(broker.config.LogLevel) == "FATAL" {
		log.SetLevel(log.FatalLevel)
	} else {
		log.SetLevel(log.InfoLevel)
	}
	brokerLogger := log.WithFields(log.Fields{
		"broker": broker.id,
	})
	return brokerLogger
}
func (broker *ServiceBroker) addService(svc *service.Service) {
	svc.SetNodeID(broker.localNode.GetID())
	broker.services = append(broker.services, svc)
	if broker.started || broker.starting {
		broker.startService(svc)
	}
	broker.logger.Debug("Broker - addService() - fullname: ", svc.FullName(), " # actions: ", len(svc.Actions()), " # events: ", len(svc.Events()))
}
func (broker *ServiceBroker) resolveSchema(svc interface{}) (moleculer.ServiceSchema, bool) {
	schema, isSchema := svc.(moleculer.ServiceSchema)
	if !isSchema {
		s, ok := svc.(*moleculer.ServiceSchema)
		if ok {
			schema = *s
			isSchema = ok
		}
	}
	return schema, isSchema
}
func (broker *ServiceBroker) createService(svc interface{}) (*service.Service, error) {
	schema, isSchema := broker.resolveSchema(svc)
	if !isSchema {
		svc, err := service.FromObject(svc, broker.delegates)
		if err != nil {
			return nil, err
		}
		return svc, nil
	}
	return service.FromSchema(schema, broker.delegates), nil
}
func (broker *ServiceBroker) WaitFor(services ...string) error {
	for _, svc := range services {
		if err := broker.waitForService(svc); err != nil {
			return err
		}
	}
	return nil
}
func (broker *ServiceBroker) WaitForNodes(nodes ...string) error {
	for _, nodeID := range nodes {
		if err := broker.waitForNode(nodeID); err != nil {
			return err
		}
	}
	return nil
}
func (broker *ServiceBroker) KnowAction(action string) bool {
	return broker.registry.KnowAction(action)
}
func (broker *ServiceBroker) WaitForActions(actions ...string) error {
	for _, action := range actions {
		if err := broker.waitAction(action); err != nil {
			return err
		}
	}
	return nil
}
func (broker *ServiceBroker) waitForService(service string) error {
	start := time.Now()
	for {
		if broker.registry.KnowService(service) {
			break
		}
		if time.Since(start) > broker.config.WaitForDependenciesTimeout {
			err := errors.New("waitForService() - Timeout ! service: " + service)
			broker.logger.Error(err)
			return err
		}
		time.Sleep(time.Microsecond)
	}
	return nil
}
func (broker *ServiceBroker) waitAction(action string) error {
	start := time.Now()
	for {
		if broker.registry.KnowAction(action) {
			break
		}
		if time.Since(start) > broker.config.WaitForDependenciesTimeout {
			err := errors.New("waitAction() - Timeout ! action: " + action)
			broker.logger.Error(err)
			return err
		}
		time.Sleep(time.Microsecond)
	}
	return nil
}
func (broker *ServiceBroker) waitForNode(nodeID string) error {
	start := time.Now()
	for {
		if broker.registry.KnowNode(nodeID) {
			break
		}
		if time.Since(start) > broker.config.WaitForDependenciesTimeout {
			err := errors.New("waitForNode() - Timeout ! nodeID: " + nodeID)
			broker.logger.Error(err)
			return err
		}
		time.Sleep(time.Microsecond)
	}
	return nil
}
func (broker *ServiceBroker) Publish(services ...interface{}) {
	for _, item := range services {
		svc, err := broker.createService(item)
		if err != nil {
			panic(errors.New("Could not publish service - error: " + err.Error()))
		}
		broker.addService(svc)
	}
}
func (broker *ServiceBroker) Start() {
	if broker.IsStarted() {
		broker.logger.Warn("broker.Start() called on a broker that already started!")
		return
	}
	broker.starting = true
	broker.logger.Info("Moleculer is starting...")
	broker.logger.Info("Node ID: ", broker.localNode.GetID())
	broker.middlewares.CallHandlers("brokerStarting", broker.delegates)
	broker.registry.Start()
	internalServices := broker.registry.LocalServices()
	for _, service := range internalServices {
		service.SetNodeID(broker.localNode.GetID())
		broker.startService(service)
	}
	for _, service := range broker.services {
		broker.startService(service)
	}
	for _, service := range internalServices {
		broker.addService(service)
	}
	broker.logger.Debug("Broker -> registry started!")
	defer broker.broadcastLocal("$broker.started")
	defer broker.middlewares.CallHandlers("brokerStarted", broker.delegates)
	broker.started = true
	broker.starting = false
	broker.logger.Info("Service Broker with ", len(broker.services), " service(s) started successfully.")
}
func (broker *ServiceBroker) Stop() {
	if !broker.started {
		broker.logger.Info("Broker is not started!")
		return
	}
	broker.logger.Info("Service Broker is stopping...")
	broker.middlewares.CallHandlers("brokerStopping", broker.delegates)
	for _, service := range broker.services {
		broker.stopService(service)
	}
	broker.registry.Stop()
	broker.started = false
	broker.broadcastLocal("$broker.stopped")
	broker.middlewares.CallHandlers("brokerStopped", broker.delegates)
}
type callPair struct {
	label  string
	result moleculer.Payload
}
func (broker *ServiceBroker) invokeMCalls(callMaps map[string]map[string]interface{}, result chan map[string]moleculer.Payload) {
	if len(callMaps) == 0 {
		result <- make(map[string]moleculer.Payload)
		return
	}
	resultChan := make(chan callPair)
	for label, content := range callMaps {
		go func(label, actionName string, params interface{}, results chan callPair) {
			result := <-broker.Call(actionName, params)
			results <- callPair{label, result}
		}(label, content["action"].(string), content["params"], resultChan)
	}
	timeoutChan := make(chan bool, 1)
	go func(timeout time.Duration) {
		time.Sleep(timeout)
		timeoutChan <- true
	}(broker.config.MCallTimeout)
	results := make(map[string]moleculer.Payload)
	for {
		select {
		case pair := <-resultChan:
			results[pair.label] = pair.result
			if len(results) == len(callMaps) {
				result <- results
				return
			}
		case <-timeoutChan:
			timeoutError := errors.New("MCall timeout error.")
			broker.logger.Error(timeoutError)
			for label, _ := range callMaps {
				if _, exists := results[label]; !exists {
					results[label] = payload.New(timeoutError)
				}
			}
			result <- results
			return
		}
	}
}