moleculer

package module
v0.3.11 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Sep 5, 2025 License: MIT Imports: 7 Imported by: 30

README

Moleculer Go

🚀 Progressive microservices framework for Go Moleculer Gopher Gopher

Inspired and compatible with Moleculer JS

Simple, fast, light and fun to develop with. Also easy, very easy to test ;)

Gitter Drone.io Build Status Go Report Card

Get Started

Example

package main

import (
	"fmt"

	"github.com/moleculer-go/moleculer"
	"github.com/moleculer-go/moleculer/broker"
)

type MathService struct {
}

func (s MathService) Name() string {
	return "math"
}

func (s *MathService) Add(params moleculer.Payload) int {
	return params.Get("a").Int() + params.Get("b").Int()
}

func (s *MathService) Sub(a int, b int) int {
	return a - b
}

func main() {
	var bkr = broker.New(&moleculer.Config{LogLevel: "error"})
	bkr.Publish(&MathService{})
	bkr.Start()
	result := <-bkr.Call("math.add", map[string]int{
		"a": 10,
		"b": 130,
	})
	fmt.Println("result: ", result.Int())
	//$ result: 140
	bkr.Stop()
}

Features

  • Service Broker
  • Transit and Transport
  • Actions (request-reply)
  • Events
  • Mixins
  • Load balancing for actions and events (random round-robin)
  • Service registry & dynamic service discovery
  • Versioned services
  • Middlewares
  • NATS Streaming Transporter
  • TCP Transporter
  • Redis Transporter
  • JSON Serializer

Installation

$ go get github.com/moleculer-go/moleculer

Running examples


# simple moleculer db example with memory adaptor
$ go run github.com/moleculer-go/store/examples/users

# simple moleculer db example with Mongo adaptor
$ go run github.com/moleculer-go/store/examples/usersMongo

# simple moleculer db example with SQLite adaptor
$ go run github.com/moleculer-go/store/examples/usersSQLite

# complex moleculer db example with population of fields by other services
$ go run github.com/moleculer-go/store/examples/populates

Running tests

# integration tests require mongo, nats streaming and rabbitmq

# run mongo
docker run -d -p 27017:27017 mongo

# run nats-streaming
docker run -d -p 4222:4222 nats-streaming -mc 0


# run rabbitmq
docker run -d -p 5672:5672 rabbitmq

# running all tests
go test ./...
# or
ginkgo -r

Documentation

Index

Constants

This section is empty.

Variables

View Source 
var DefaultConfig = Config{
	LogLevel:                   "INFO",
	LogFormat:                  "TEXT",
	DiscoverNodeID:             discoverNodeID,
	Transporter:                "MEMORY",
	UpdateNodeMetricsFrequency: 5 * time.Second,
	HeartbeatFrequency:         5 * time.Second,
	HeartbeatTimeout:           15 * time.Second,
	OfflineCheckFrequency:      20 * time.Second,
	OfflineTimeout:             10 * time.Minute,
	DontWaitForNeighbours:      true,
	NeighboursCheckTimeout:     2 * time.Second,
	WaitForDependenciesTimeout: 2 * time.Second,
	Metrics:                    false,
	MetricsRate:                1,
	DisableInternalServices:    false,
	DisableInternalMiddlewares: false,
	Created:                    func() {},
	Started:                    func() {},
	Stopped:                    func() {},
	MaxCallLevel:               100,
	RetryPolicy: &RetryPolicy{
		Enabled: false,
	},
	RequestTimeout:            3 * time.Second,
	MCallTimeout:              5 * time.Second,
	WaitForNeighboursInterval: 200 * time.Millisecond,

	TCPOptions: map[string]interface{}{
		"UdpDiscovery":          true,
		"UdpReuseAddr":          true,
		"UdpPort":               4445,
		"UdpBindAddress":        "",
		"UdpPeriod":             30 * time.Second,
		"UdpMaxDiscovery":       0,
		"WorkerPoolSize":        20,
		"ConnectionTimeout":     30 * time.Second,
		"IdleConnectionTimeout": 60 * time.Second,
		"UdpMulticast":          "239.0.0.0",
		"UdpMulticastTTL":       1,
		"UdpBroadcast":          []string{},
		"Port":                  0,
		"Urls":                  []string{},
		"UseHostname":           true,
		"GossipPeriod":          2,
		"MaxConnections":        32,
		"MaxPacketSize":         1024 * 1024,
		"Prefix":                "",
		"NodeId":                "",
		"Namespace":             "",
		"Logger":                nil,
		"Serializer":            nil,
		"ValidateMsg":           nil,
	},
}

Functions

This section is empty.

Types

type Action 

type Action struct {
	Name        string
	Handler     ActionHandler
	Schema      ActionSchema
	Settings    map[string]interface{}
	Description string
}

type ActionDelegateFunc 

type ActionDelegateFunc func(context BrokerContext, opts ...Options) chan Payload

type ActionHandler 

type ActionHandler func(context Context, params Payload) interface{}

type ActionSchema 

type ActionSchema interface {
}

ActionSchema is used by the validation engine to check if parameters sent to the action are valid.

type BrokerContext 

type BrokerContext interface {
	Call(actionName string, params interface{}, opts ...Options) chan Payload
	Emit(eventName string, params interface{}, groups ...string)

	ChildActionContext(actionName string, params Payload, opts ...Options) BrokerContext
	ChildEventContext(eventName string, params Payload, groups []string, broadcast bool) BrokerContext

	ActionName() string
	EventName() string
	Payload() Payload
	Groups() []string
	IsBroadcast() bool
	Caller() string

	//export context info in a map[string]
	AsMap() map[string]interface{}

	SetTargetNodeID(targetNodeID string)
	TargetNodeID() string

	ID() string
	RequestID() string
	Meta() Payload
	UpdateMeta(Payload)
	Logger() *log.Entry

	Publish(...interface{})
	WaitFor(services ...string) error
}

type BrokerContextFunc 

type BrokerContextFunc func() BrokerContext

type BrokerDelegates 

type BrokerDelegates struct {
	InstanceID         InstanceIDFunc
	LocalNode          LocalNodeFunc
	Logger             LoggerFunc
	Bus                BusFunc
	IsStarted          isStartedFunc
	Config             Config
	MultActionDelegate MultActionDelegateFunc
	ActionDelegate     ActionDelegateFunc
	EmitEvent          EmitEventFunc
	BroadcastEvent     EmitEventFunc
	HandleRemoteEvent  EmitEventFunc
	ServiceForAction   ServiceForActionFunc
	BrokerContext      BrokerContextFunc
	MiddlewareHandler  MiddlewareHandlerFunc
	Publish            PublishFunc
	WaitFor            WaitForFunc
}

Needs Refactoring..2 broker interfaces.. one for regiwstry.. and for for all others.

type BusFunc 

type BusFunc func() *bus.Emitter

type Config 

type Config struct {
	LogLevel                   string
	LogFormat                  string
	DiscoverNodeID             func() string
	Transporter                string
	TransporterFactory         TransporterFactoryFunc
	StrategyFactory            StrategyFactoryFunc
	UpdateNodeMetricsFrequency time.Duration
	HeartbeatFrequency         time.Duration
	HeartbeatTimeout           time.Duration
	OfflineCheckFrequency      time.Duration
	OfflineTimeout             time.Duration
	NeighboursCheckTimeout     time.Duration
	WaitForDependenciesTimeout time.Duration
	Middlewares                []Middlewares
	Namespace                  string
	RequestTimeout             time.Duration
	MCallTimeout               time.Duration
	RetryPolicy                *RetryPolicy
	MaxCallLevel               int
	Metrics                    bool
	MetricsRate                float32
	DisableInternalServices    bool
	DisableInternalMiddlewares bool
	DontWaitForNeighbours      bool
	WaitForNeighboursInterval  time.Duration
	Created                    func()
	Started                    func()
	Stopped                    func()

	Services map[string]interface{}

	// TCP transporter options (map-based for user flexibility)
	TCPOptions map[string]interface{}
}

type Context 

type Context interface {
	//context methods used by services
	MCall(map[string]map[string]interface{}) chan map[string]Payload
	Call(actionName string, params interface{}, opts ...Options) chan Payload
	Emit(eventName string, params interface{}, groups ...string)
	Broadcast(eventName string, params interface{}, groups ...string)
	Logger() *log.Entry

	Payload() Payload
	Meta() Payload
}

type CreatedFunc 

type CreatedFunc func(ServiceSchema, *log.Entry)

type EmitEventFunc 

type EmitEventFunc func(context BrokerContext)

type Event 

type Event struct {
	Name    string
	Group   string
	Handler EventHandler
}

type EventHandler 

type EventHandler func(context Context, params Payload)

type ForEachFunc 

type ForEachFunc func(iterator func(key interface{}, value Payload) bool)

type ForEachNodeFunc added in v0.3.4 

type ForEachNodeFunc func(node Node) bool

type InstanceIDFunc added in v0.3.0 

type InstanceIDFunc func() string

type LifecycleFunc 

type LifecycleFunc func(BrokerContext, ServiceSchema)

type LocalNodeFunc 

type LocalNodeFunc func() Node

type LoggerFunc 

type LoggerFunc func(name string, value string) *log.Entry

type Middleware 

type Middleware interface {
	CallHandlers(name string, params interface{}) interface{}
}

type MiddlewareHandler 

type MiddlewareHandler func(params interface{}, next func(...interface{}))

type MiddlewareHandlerFunc 

type MiddlewareHandlerFunc func(name string, params interface{}) interface{}

type Middlewares 

type Middlewares map[string]MiddlewareHandler

type Mixin 

type Mixin struct {
	Name         string
	Dependencies []string
	Settings     map[string]interface{}
	Metadata     map[string]interface{}
	Hooks        map[string]interface{}
	Actions      []Action
	Events       []Event
	Created      CreatedFunc
	Started      LifecycleFunc
	Stopped      LifecycleFunc
}

type MultActionDelegateFunc 

type MultActionDelegateFunc func(callMaps map[string]map[string]interface{}) chan map[string]Payload

type Node 

type Node interface {
	GetID() string
	GetHost() string
	ExportAsMap() map[string]interface{}
	IsAvailable() bool
	GetIpList() []string
	GetPort() int
	Available()
	Unavailable()
	IsExpired(timeout time.Duration) bool
	Update(id string, info map[string]interface{}) (bool, []map[string]interface{})
	UpdateInfo(info map[string]interface{}) []map[string]interface{}
	IncreaseSequence()
	HeartBeat(heartbeat map[string]interface{})
	Publish(service map[string]interface{})
	GetUdpAddress() string
	GetSequence() int64
	GetCpuSequence() int64
	GetCpu() int64
	IsLocal() bool
	UpdateMetrics()
	GetHostname() string
}

type ObjectSchema 

type ObjectSchema struct {
	Source interface{}
}

type Options added in v0.1.1 

type Options struct {
	Meta   Payload
	NodeID string
}

type Payload 

type Payload interface {
	First() Payload
	Sort(field string) Payload
	Remove(fields ...string) Payload
	AddItem(value interface{}) Payload
	Add(field string, value interface{}) Payload
	AddMany(map[string]interface{}) Payload
	MapArray() []map[string]interface{}
	RawMap() map[string]interface{}
	Bson() bson.M
	BsonArray() bson.A
	Map() map[string]Payload
	Exists() bool
	IsError() bool
	Error() error
	ErrorPayload() Payload
	Value() interface{}
	ValueArray() []interface{}
	Int() int
	IntArray() []int
	Int64() int64
	Int64Array() []int64
	Uint() uint64
	UintArray() []uint64
	Float32() float32
	Float32Array() []float32
	Float() float64
	FloatArray() []float64
	String() string
	StringArray() []string
	Bool() bool
	BoolArray() []bool
	ByteArray() []byte
	Time() time.Time
	TimeArray() []time.Time
	Array() []Payload
	At(index int) Payload
	Len() int
	Get(path string, defaultValue ...interface{}) Payload
	//Only return a payload containing only the field specified
	Only(path string) Payload
	IsArray() bool
	IsMap() bool
	ForEach(iterator func(key interface{}, value Payload) bool)
	MapOver(tranform func(in Payload) Payload) Payload
}

Payload contains the data sent/return to actions. I has convinience methods to read action parameters by name with the right type.

type PublishFunc 

type PublishFunc func(...interface{})

type Registry added in v0.3.4 

type Registry interface {
	GetNodeByID(nodeID string) Node
	AddOfflineNode(nodeID, hostname, ipAddress string, port int) Node
	ForEachNode(ForEachNodeFunc)
	DisconnectNode(nodeID string)
	RemoteNodeInfoReceived(message Payload)
	GetLocalNode() Node
	GetNodeByAddress(host string) Node
}

type RetryPolicy 

type RetryPolicy struct {
	Enabled  bool
	Retries  int
	Delay    int
	MaxDelay int
	Factor   int
	Check    func(error) bool
}

type ServiceForActionFunc 

type ServiceForActionFunc func(string) []*ServiceSchema

type ServiceSchema 

type ServiceSchema struct {
	Name         string
	Version      string
	Dependencies []string
	Settings     map[string]interface{}
	Metadata     map[string]interface{}
	Hooks        map[string]interface{}
	Mixins       []Mixin
	Actions      []Action
	Events       []Event
	Created      CreatedFunc
	Started      LifecycleFunc
	Stopped      LifecycleFunc
}

type StrategyFactoryFunc added in v0.2.2 

type StrategyFactoryFunc func() interface{}

type TCPConfig added in v0.3.7 

type TCPConfig struct {
	// Enable UDP discovery
	UdpDiscovery bool
	// Reusing UDP server socket
	UdpReuseAddr bool

	// UDP port for listening and discovery (defaults to 4445 for compatibility)
	UdpPort int
	// UDP bind address (if null, bind on all interfaces)
	UdpBindAddress string
	// UDP sending period (seconds)
	UdpPeriod time.Duration

	UdpMaxDiscovery int

	// Memory management options
	// Worker pool size for connection handling (default: 20)
	WorkerPoolSize int
	// Connection timeout duration (default: 30 seconds)
	ConnectionTimeout time.Duration
	// Idle connection cleanup interval (default: 60 seconds)
	IdleConnectionTimeout time.Duration

	// Multicast address.
	UdpMulticast string
	// Multicast TTL setting
	UdpMulticastTTL int

	// Send broadcast (Boolean, String, Array<String>)
	UdpBroadcast      []string
	UdpBroadcastAddrs []string
	// TCP server port. 0 means random port
	Port int
	// Static remote nodes address list (when UDP discovery is not available)
	Urls []string
	// Use hostname as preffered connection address
	UseHostname bool

	// Gossip sending period in seconds
	GossipPeriod int
	// Maximum enabled outgoing connections. If reach, close the old connections
	MaxConnections int
	// Maximum TCP packet size
	MaxPacketSize int

	Prefix      string
	NodeId      string
	Namespace   string
	Logger      *log.Entry
	Serializer  interface{} // Will be set to the actual serializer type
	ValidateMsg interface{} // Will be set to the actual validate function type
}

TCPConfig holds TCP transporter configuration options

type TransporterFactoryFunc 

type TransporterFactoryFunc func() interface{}

type WaitForFunc added in v0.1.3 

type WaitForFunc func(...string) error

Source Files

View all Source files

Directories

Path Synopsis
cli
cmd
examples
amqp_transporter command
math
math2 command
standalone command
amqp
kafka
memory
nats
pubsub
redis
tcp

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.