README
¶
Transistor 
Transistor allows you to run distributed workload on one or accross multiple hosts. It's a plugin based system that allows plugins to subscribe to multiple events and internal scheduler takes care of delivery. This allows multiple plugins to recieve same message do some work and respond with updated or different message.
We use a central file api.go that keeps all available messages in one place.
package plugins
import "github.com/codeamp/transistor"
func init() {
}
type Hello struct {
Action string
Message string
}
you also need to register all api events with transistor so that we are able to transform it to correct type when json event payload is recieved.
All plugins need to implement Start, Stop, Subscribe and Process methods.
type Plugin interface {
// Start starts the Plugin service, whatever that may be
Start(chan Event) error
// Stop stops the services and closes any necessary channels and connections
Stop()
// Subscribe takes in an event message and validates it for Process
Subscribe() []string
// Process takes in an event message and tries to process it
Process(Event) error
}
You can create new events
func (x *ExamplePlugin1) Start(e chan transistor.Event) error {
log.Info("starting ExamplePlugin")
event := Hello{
Action: "examplePlugin1",
Message: "Hello World from ExamplePlugin1",
}
e <- transistor.NewEvent(transistor.EventName("exampleplugin1"), transistor.Action("create"), nil)
return nil
}
or respond to existing one and keep track of parent event
func (x *ExamplePlugin2) Process(e transistor.Event) error {
if e.Name == "exampleplugin2:create" {
hello := e.Payload().(Hello)
log.Info("ExamplePlugin2 received a message:", hello)
}
return nil
}
Transistor can run on a multiple or single host. To run on multiple hosts you will need a Redis connection. This is a minimal example to set up 2 plugins:
func main() {
config := transistor.Config{
Server: "0.0.0.0:16379",
Database: "0",
Pool: "30",
Process: "1",
Queueing: true,
Plugins: map[string]interface{}{
"exampleplugin1": map[string]interface{}{
"hello": "world1",
"workers": 1,
},
"exampleplugin2": map[string]interface{}{
"hello": "world2",
"workers": 1,
},
},
EnabledPlugins: []string{"exampleplugin1", "exampleplugin2"},
}
t, err := transistor.NewTransistor(config)
if err != nil {
log.Fatal(err)
}
signals := make(chan os.Signal)
signal.Notify(signals, os.Interrupt, syscall.SIGTERM)
go func() {
sig := <-signals
if sig == os.Interrupt || sig == syscall.SIGTERM {
log.Info("Shutting down circuit. SIGTERM recieved!\n")
// If Queueing is ON then workers are responsible for closing Shutdown chan
if !t.Config.Queueing {
t.Stop()
}
}
}()
log.InfoWithFields("plugins loaded", log.Fields{
"plugins": strings.Join(t.PluginNames(), ","),
})
t.Run()
}
if you want to run on a single host and without redis you need to set Queueing: false in config. You can see and run a minimal example that uses Redis in example/ folder.
Transistor was build to power Checkr's deployment pipeline and it's used to build and deploy over 100 microservices to kubernetes.
Documentation
¶
Index ¶
- Variables
- func MapPayload(name string, event *Event) error
- func RandomString(strlen int) string
- func RegisterPlugin(name string, creator Creator, events ...interface{})
- func SliceContains(name string, list []string) bool
- type Action
- type Artifact
- type Caller
- type Config
- type Creator
- type Event
- func (e *Event) AddArtifact(key string, value interface{}, secret bool)
- func (e *Event) CreateEvent(action Action, state State, stateMessage string) Event
- func (e *Event) DeleteEvent(action Action, state State, stateMessage string) Event
- func (e *Event) Dump()
- func (e *Event) Event() string
- func (e *Event) GetArtifact(key string) (Artifact, error)
- func (e *Event) GetArtifactFromSource(key string, source string) (Artifact, error)
- func (e *Event) Matches(name string) bool
- func (e *Event) NewEvent(action Action, state State, stateMessage string) Event
- func (e *Event) SetPayload(payload interface{})
- func (e *Event) StatusEvent(action Action, state State, stateMessage string) Event
- func (e *Event) UpdateEvent(action Action, state State, stateMessage string) Event
- type EventName
- type Plugin
- type RunningPlugin
- type State
- type Transistor
Constants ¶
This section is empty.
Variables ¶
var EventRegistry = make(map[string]interface{})
var PluginRegistry = map[string]Creator{}
Functions ¶
func MapPayload ¶
func RandomString ¶
func RegisterPlugin ¶
func SliceContains ¶
Types ¶
type Artifact ¶
type Artifact struct {
Source string `json:"source,omitempty"`
Key string `json:"key"`
Value interface{} `json:"value"`
Secret bool `json:"secret"`
}
func (*Artifact) StringSlice ¶
func (a *Artifact) StringSlice() []interface{}
type Event ¶
type Event struct {
ID uuid.UUID `json:"id"`
ParentID uuid.UUID `json:"parentId"`
Name EventName `json:"name"`
Action Action `json:"action"`
State State `json:"state"`
StateMessage string `json:"stateMessage"`
Payload interface{} `json:"payload"`
PayloadModel string `json:"payloadModel"`
CreatedAt time.Time `json:"createdAt"`
Caller Caller `json:"caller"`
Artifacts []Artifact `json:"artifacts"`
}
func CreateEvent ¶
func DeleteEvent ¶
func UpdateEvent ¶
func (*Event) AddArtifact ¶
func (*Event) CreateEvent ¶
func (*Event) DeleteEvent ¶
func (*Event) GetArtifactFromSource ¶
func (*Event) SetPayload ¶
func (e *Event) SetPayload(payload interface{})
func (*Event) StatusEvent ¶
type Plugin ¶
type Plugin interface {
// Start starts the Plugin service, whatever that may be
Start(chan Event) error
// Stop stops the services and closes any necessary channels and connections
Stop()
// Subscribe takes in an event message and validates it for Process
Subscribe() []string
// Process takes in an event message and tries to process it
Process(Event) error
}
type RunningPlugin ¶
type Transistor ¶
type Transistor struct {
Config Config
Events chan Event
TestEvents chan Event
Shutdown chan struct{}
Plugins []*RunningPlugin
}
Transistor runs codeflow and collects datt based on the given config
func NewTestTransistor ¶
func NewTestTransistor(config Config) (*Transistor, error)
NewTestTransistor returns an Transistor struct based off the given Config
func NewTransistor ¶
func NewTransistor(config Config) (*Transistor, error)
NewTransistor returns an Transistor struct based off the given Config
func (*Transistor) GetTestEvent ¶
func (*Transistor) LoadPlugins ¶
func (t *Transistor) LoadPlugins() error
func (*Transistor) PluginNames ¶
func (t *Transistor) PluginNames() []string
Returns t list of strings of the configured plugins.
Source Files
Directories