fer

README

fer

fer is a simple reimplementation of FairMQ in Go.

License

fer is released under the BSD-3 license.

Installation

fer is installable via go get:

$> go get github.com/alice-go/fer/...

NOTE: you need at least go1.7.

Documentation

Documentation is available on godoc.

Examples

Testing example-2 from FairMQ tutorial

## terminal 1
$> fer-ex-sink --id sink1 --mq-config ./_example/cmd/testdata/ex2-sampler-processor-sink.json

## terminal 2
$> fer-ex-processor --id processor --mq-config ./_example/cmd/testdata/ex2-sampler-processor-sink.json

## terminal 3
$> fer-ex-sampler --id sampler1 --mq-config ./_example/cmd/testdata/ex2-sampler-processor-sink.json

This will run 3 devices, using the ZeroMQ transport.

To run with nanomsg as a transport layer, add --transport nanomsg to the invocations.

Documentation

Overview

Package fer provides a basic framework to run FairMQ-like processes. Clients create fer Devices that exchange data via fer message queue sockets.

A client device might look like so:

import "github.com/alice-go/fer"
import "github.com/alice-go/fer/config"
type myDevice struct {
    cfg  config.Device
    imsg chan fer.Msg
    omsg chan fer.Msg
}

A device needs to implement the fer.Device interface:

func (dev *myDevice) Run(ctl fer.Controller)   error { ... }

Optionnally, the following methods may be also implemented:

func (dev *myDevice) Configure(cfg config.Device) error { ... }
func (dev *myDevice) Init(ctl fer.Controller)  error { ... }
func (dev *myDevice) Pause(ctl fer.Controller) error { ... }
func (dev *myDevice) Reset(ctl fer.Controller) error { ... }

Typically, the Configure method is used to retrieve the configuration associated with the client's device. The Init method is used to retrieve the channels of input/output data messages. The Run method is an infinite for-loop, selecting on these input/output data messages. This infinite for-loop will also NEED to listen for the Controller.Done() channel to exit that for-loop.

e.g.:

func (dev *myDevice) Init(ctl fer.Controller) error {
    imsg, err := ctl.Chan("data-1", 0)
    omsg, err := ctl.Chan("data-2", 0)
    dev.imsg = imsg
    dev.omsg = omsg
    return nil
}

func (dev *myDevice) Run(ctl fer.Controller) error {
    for {
        select {
        case data := <-dev.imsg:
            dev.omsg <- bytes.Repeat(data, 2)
        case <-ctl.Done():
            return nil
        }
    }
}

Then, to create an executable:

package main

func main() {
    err := fer.Main(&myDevice{})
    if err != nil {
        log.Fatal(err)
    }
}

Build it as usual and run like so:

$> go build -o my-device
$> ./my-device --help
Usage of my-device:
  -control string
    	starts device in interactive/static mode (default "interactive")
  -id string
    	device ID
  -mq-config string
    	path to JSON file holding device configuration
  -transport string
    	transport mechanism to use (zeromq, nanomsg, go-chan, ...) (default "zeromq")
$> ./my-device --id my-id --mq-config ./path/to/config.json

Index

• func Main(dev Device) error
• func RunDevice(ctx context.Context, cfg config.Config, dev Device, r io.Reader, w io.Writer) error
• type Cmd
  • func (cmd Cmd) String() string
• type Controller
• type DevConfigurer
• type DevIniter
• type DevPauser
• type DevReseter
• type Device
• type Logger
• type Msg

Functions

func Main

func Main(dev Device) error

Main configures and runs a device's execution, managing its state.

func RunDevice

func RunDevice(ctx context.Context, cfg config.Config, dev Device, r io.Reader, w io.Writer) error

RunDevice runs a device's execution, managing its state.

Types

type Cmd

type Cmd byte

Cmd describes commands to be sent to a device, via a channel.

const (
	// CmdInitDevice is the command sent to initialize a device
	CmdInitDevice Cmd = iota
	// CmdInitTask is the command sent to initialize the tasks of a device
	CmdInitTask
	// CmdRun is the command sent to run a device
	CmdRun
	// CmdPause is the command sent to pause the execution of a device
	CmdPause
	// CmdStop is the command sent to stop the execution of a device
	CmdStop
	// CmdResetTask is the command sent to reset the state of the tasks of a device
	CmdResetTask
	// CmdResetDevice is the command sent to reset the state of a device
	CmdResetDevice
	// CmdEnd is the command sent to end a device
	CmdEnd
	// CmdError is the command sent to notify of an error
	CmdError
)

func (Cmd) String

func (cmd Cmd) String() string

type Controller

type Controller interface {
	Logger
	Chan(name string, i int) (chan Msg, error)
	Done() chan Cmd
	// contains filtered or unexported methods
}

Controller controls devices execution and gives a device access to input and output data channels.

type DevConfigurer

type DevConfigurer interface {
	// Configure hands a device its configuration.
	Configure(cfg config.Device) error
}

DevConfigurer configures a fer device.

type DevIniter

type DevIniter interface {
	// Init gives a chance to the device to initialize internal
	// data structures, retrieve channels to input/output data.
	Init(ctl Controller) error
}

DevIniter initializes a fer device.

type DevPauser

type DevPauser interface {
	// Pause pauses the device's execution.
	Pause(ctl Controller) error
}

DevPauser pauses the execution of a fer device.

type DevReseter

type DevReseter interface {
	// Reset resets the device's internal state.
	Reset(ctl Controller) error
}

DevReseter resets a fer device.

type Device

type Device interface {
	// Run is where the device's main activity happens.
	// Run should loop forever, until the Controller.Done() channel says
	// otherwise.
	Run(ctl Controller) error
}

Device is a handle to what users get to run via the Fer toolkit.

Devices are configured according to command-line flags and a JSON configuration file. Clients need to implement the Run method to receive and send data via the Controller data channels.

type Logger

type Logger interface {
	Fatalf(format string, v ...interface{})
	Printf(format string, v ...interface{})
}

Logger gives access to printf-like facilities

type Msg

type Msg struct {
	Data []byte // Data is the message payload.
	Err  error  // Err indicates whether an error occured.
}

Msg is a quantum of data being exchanged between devices.