moleculer

README

Moleculer Go

🚀 Progressive microservices framework for Go

Inspired and compatible with Moleculer JS

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

Get Started

• http://gomicro.services (Official site and Documentation)
• Database examples

Example

package main

import (
	"fmt"

	"gitee.com/anystreaming/moleculer-go"
	"gitee.com/anystreaming/moleculer-go/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()
}

Roadmap

v0.1.0 (MVP)

Contents:

• 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
• JSON Serializer
• Examples :)

v0.2.0 (Beta RC1)

• Action validators
• Support for streams
• More Load balancing implementations (cpu-usage, latency)
• Fault tolerance features (Circuit Breaker, Bulkhead, Retry, Timeout, Fallback)
• Built-in caching solution (memory, Redis)
• More transporters (gRPC, TCP, Redis, Kafka)
• More serializers (Avro, MsgPack, Protocol Buffer, Thrift)

v0.3.0 (Beta)

• Performance and Optimization
• More DB Adaptors (Firebase, MySQL)
• CLI for Project Seed Generation

v0.4.0 (Alpha)

• Event Sourcing Mixins

v0.5.0 (Release)

Installation

$ go get gitee.com/anystreaming/moleculer-go

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

• Variables
• type Action
• type ActionDelegateFunc
• type ActionHandler
• type ActionSchema
• type BrokerContext
• type BrokerContextFunc
• type BrokerDelegates
• type BusFunc
• type Config
• type Context
• type CreatedFunc
• type EmitEventFunc
• type Event
• type EventHandler
• type ForEachFunc
• type LifecycleFunc
• type LocalNodeFunc
• type LoggerFunc
• type Middleware
• type MiddlewareHandler
• type MiddlewareHandlerFunc
• type Middlewares
• type Mixin
• type MultActionDelegateFunc
• type Node
• type ObjectSchema
• type Options
• type Payload
• type PublishFunc
• type RedisConfig
• type RetryPolicy
• type ServiceForActionFunc
• type ServiceSchema
• type StrategyFactoryFunc
• type TransporterFactoryFunc
• type WaitForFunc

Variables

var DefaultConfig = Config{
	LogLevel:                   "INFO",
	LogFormat:                  "TEXT",
	DiscoverNodeID:             discoverNodeID,
	Transporter:                "MEMORY",
	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:            1 * time.Minute,
	MCallTimeout:              5 * time.Second,
	WaitForNeighboursInterval: 200 * time.Millisecond,
	RedisConfig:               nil,
}

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

	//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 {
	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
	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()
	RedisConfig                *RedisConfig
	Services                   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 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
	ExportAsMap() map[string]interface{}
	IsAvailable() bool
	Available()
	Unavailable()
	IsExpired(timeout time.Duration) bool
	Update(id string, info map[string]interface{}) bool

	IncreaseSequence()
	HeartBeat(heartbeat map[string]interface{})
	Publish(service map[string]interface{})
}

type ObjectSchema

type ObjectSchema struct {
	Source interface{}
}

type Options

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 RedisConfig

type RedisConfig struct {
	Addr     string
	PassWord string
	DB       int
}

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

type StrategyFactoryFunc func() interface{}

type TransporterFactoryFunc

type TransporterFactoryFunc func() interface{}

type WaitForFunc

type WaitForFunc func(...string) error