lineage

Documentation

Index

• type Message
  • func (msg *Message) Deserialize(data []byte) (err error)
  • func (msg *Message) Serialize() (data []byte, err error)
• type MqTracker
  • func NewMqTracker(processName, pipelineHash, brokerURL string, ...) MqTracker
  • func (tracker MqTracker) Start(wg *sync.WaitGroup)
  • func (tracker MqTracker) StartBlocking()

Types

type Message

type Message struct {
	ProcessName string                  `json:"transformation_step"`
	Fragment    desc.RemoteFragmentDesc `json:"description"`
}

Message is the structure holding the lineage information of a fragment

func (*Message) Deserialize

func (msg *Message) Deserialize(data []byte) (err error)

Deserialize tries to decode a json encoded byte array into `msg`. Errors on failure

func (*Message) Serialize

func (msg *Message) Serialize() (data []byte, err error)

Serialize tries to transform `msg` into a json encoded bytearray. Errors on failure

type MqTracker

type MqTracker struct {
	TransformationName string
	PipelineHash       string
	ToLineate          <-chan transmit.Serializable // desc.RemoteFragmentDesc
	Sender             messageq.SimpleSender
}

MqTracker posts lineage information to a designated message queue queue

func NewMqTracker

func NewMqTracker(processName, pipelineHash, brokerURL string, toLineate chan transmit.Serializable) MqTracker

NewMqTracker instantiates a new MqTracker

func (MqTracker) Start

func (tracker MqTracker) Start(wg *sync.WaitGroup)

Start asychronously calls StartBlocking via Goroutine

func (MqTracker) StartBlocking

func (tracker MqTracker) StartBlocking()

StartBlocking starts the main loop of the Tracker