core

package

Go to main page

Versions in this module

v0

v0.6.0-alpha1

Aug 22, 2018

v0.5.4

Apr 17, 2019

v0.5.3

Apr 23, 2018

v0.5.2

Mar 30, 2018

v0.5.1

Jan 24, 2018

v0.5.0

Dec 21, 2017

Changes in this version

+ const FilterResultMessageAccept

+ const InvalidStream

+ const MessageQueueDiscard

+ const MessageQueueOk

+ const MessageQueueTimeout

+ const MetricActiveWorkers

+ const MetricMessagesRoutedAvg

+ const MetricPluginsActive

+ const MetricPluginsDead

+ const MetricPluginsInit

+ const MetricPluginsPrepareStop

+ const MetricPluginsStopping

+ const MetricPluginsWaiting

+ const ModulateResultContinue

+ const ModulateResultDiscard

+ const ModulateResultFallback

+ const PluginStatePrepareStop

+ const PluginStructTagDefault

+ const PluginStructTagMetric

+ const TraceInternalStream

+ var GeneratedRouterPrefix = "_GENERATED_"

+ var MessageDataPool = tcontainer.NewBytePoolWithSize(2)

+ var MessageTrace = func(msg *Message, pluginID string, comment string)

+ var TraceInternalStreamID = GetStreamID(TraceInternalStream)

+ var TypeRegistry = treflect.NewTypeRegistry()

+ func ActivateMessageTrace()

+ func CountConsumers()

+ func CountFallbackRouters()

+ func CountMessageDiscarded()

+ func CountMessageRouted()

+ func CountMessagesEnqueued()

+ func CountProducers()

+ func CountRouters()

+ func DeactivateMessageTrace()

+ func DiscardMessage(msg *Message, pluginID string, comment string)

+ func GetVersionNumber() int64

+ func GetVersionString() string

+ func Route(msg *Message, router Router) error

+ func RouteOriginal(msg *Message, router Router) error

+ type BatchedProducer struct

+ Batch MessageBatch

+ func (prod *BatchedProducer) BatchMessageLoop(workers *sync.WaitGroup, onBatchFlush func() AssemblyFunc)

+ func (prod *BatchedProducer) Configure(conf PluginConfigReader)

+ func (prod *BatchedProducer) DefaultClose()

+ func (prod *BatchedProducer) Enqueue(msg *Message, timeout time.Duration)

+ type BufferedProducer struct

+ func (prod *BufferedProducer) CloseMessageChannel(handleMessage func(*Message)) (empty bool)

+ func (prod *BufferedProducer) Configure(conf PluginConfigReader)

+ func (prod *BufferedProducer) DefaultClose()

+ func (prod *BufferedProducer) DefaultDrain()

+ func (prod *BufferedProducer) DrainMessageChannel(handleMessage func(*Message), timeout time.Duration) bool

+ func (prod *BufferedProducer) Enqueue(msg *Message, timeout time.Duration)

+ func (prod *BufferedProducer) GetQueueTimeout() time.Duration

+ func (prod *BufferedProducer) MessageControlLoop(onMessage func(*Message))

+ func (prod *BufferedProducer) TickerMessageControlLoop(onMessage func(*Message), interval time.Duration, onTimeOut func())

type Config

+ func ReadConfigFromFile(path string) (*Config, error)

+ func (conf *Config) GetConsumers() []PluginConfig

+ func (conf *Config) GetProducers() []PluginConfig

+ func (conf *Config) GetRouters() []PluginConfig

+ func (conf *Config) Validate() error

+ type Configurable interface

+ Configure func(PluginConfigReader)

type Consumer

+ GetShutdownTimeout func() time.Duration

+ type DirectProducer struct

+ func (prod *DirectProducer) Configure(conf PluginConfigReader)

+ func (prod *DirectProducer) Enqueue(msg *Message, timeout time.Duration)

+ func (prod *DirectProducer) MessageControlLoop(onMessage func(*Message))

+ func (prod *DirectProducer) TickerMessageControlLoop(onMessage func(*Message), interval time.Duration, onTimeOut func())

type Filter

+ ApplyFilter func(msg *Message) (FilterResult, error)

+ type FilterArray []Filter

+ func (filters FilterArray) ApplyFilter(msg *Message) (FilterResult, error)

+ type FilterModulator struct

+ Filter Filter

+ func NewFilterModulator(filter Filter) *FilterModulator

+ func (filterModulator *FilterModulator) ApplyFilter(msg *Message) (FilterResult, error)

+ func (filterModulator *FilterModulator) Modulate(msg *Message) ModulateResult

+ type FilterResult uint64

+ func FilterResultMessageReject(stream MessageStreamID) FilterResult

+ func (f FilterResult) GetStreamID() MessageStreamID

type Formatter

+ ApplyFormatter func(msg *Message) error

+ CanBeApplied func(msg *Message) bool

+ type FormatterArray []Formatter

+ func (formatters FormatterArray) ApplyFormatter(msg *Message) error

+ func (formatters FormatterArray) CanBeApplied(msg *Message) bool

+ type FormatterModulator struct

+ Formatter Formatter

+ func NewFormatterModulator(formatter Formatter) *FormatterModulator

+ func (formatterModulator *FormatterModulator) ApplyFormatter(msg *Message) error

+ func (formatterModulator *FormatterModulator) CanBeApplied(msg *Message) bool

+ func (formatterModulator *FormatterModulator) Modulate(msg *Message) ModulateResult

+ type GetAppliedContent func(msg *Message) []byte

+ func GetAppliedContentGetFunction(applyTo string) GetAppliedContent

type LogConsumer

+ func (cons *LogConsumer) Fire(logrusEntry *logrus.Entry) error

+ func (cons *LogConsumer) GetID() string

+ func (cons *LogConsumer) GetShutdownTimeout() time.Duration

+ func (cons *LogConsumer) IsBlocked() bool

+ func (cons *LogConsumer) Levels() []logrus.Level

type Message

+ func (msg *Message) Clone() *Message

+ func (msg *Message) CloneOriginal() *Message

+ func (msg *Message) ExtendPayload(size int) []byte

+ func (msg *Message) FreezeOriginal()

+ func (msg *Message) GetCreationTime() time.Time

+ func (msg *Message) GetMetadata() Metadata

+ func (msg *Message) GetOrigRouter() Router

+ func (msg *Message) GetOrigStreamID() MessageStreamID

+ func (msg *Message) GetPayload() []byte

+ func (msg *Message) GetPrevRouter() Router

+ func (msg *Message) GetPrevStreamID() MessageStreamID

+ func (msg *Message) GetRouter() Router

+ func (msg *Message) GetSource() MessageSource

+ func (msg *Message) GetStreamID() MessageStreamID

+ func (msg *Message) ResizePayload(size int) []byte

+ func (msg *Message) SetStreamID(streamID MessageStreamID)

+ func (msg *Message) SetlStreamIDAsOriginal(streamID MessageStreamID)

+ func (msg *Message) StorePayload(data []byte)

+ func (msg *Message) TryGetMetadata() Metadata

+ type MessageData struct

+ type MessageQueue chan *Message

+ func NewMessageQueue(capacity int) MessageQueue

+ func (channel MessageQueue) Close()

+ func (channel MessageQueue) GetNumQueued() int

+ func (channel MessageQueue) IsEmpty() bool

+ func (channel MessageQueue) Pop() (*Message, bool)

+ func (channel MessageQueue) PopWithTimeout(maxDuration time.Duration) (*Message, bool)

+ func (channel MessageQueue) Push(msg *Message, timeout time.Duration) (state MessageQueueResult)

+ type MessageQueueResult int

type MessageSource

+ GetID func() string

type MessageStreamID

+ func (streamID MessageStreamID) GetName() string

+ type Metadata map[string][]byte

+ func (meta Metadata) Clone() (clone Metadata)

+ func (meta Metadata) Delete(key string)

+ func (meta Metadata) GetValue(key string) []byte

+ func (meta Metadata) GetValueString(key string) string

+ func (meta Metadata) SetValue(key string, value []byte)

+ func (meta Metadata) TryGetValue(key string) ([]byte, bool)

+ func (meta Metadata) TryGetValueString(key string) (string, bool)

+ func (meta Metadata) TrySetValue(key string, value []byte) bool

+ type ModulateResult int

+ type ModulateResultError struct

+ func NewModulateResultError(message string, values ...interface{}) ModulateResultError

+ func (p ModulateResultError) Error() string

+ type Modulator interface

+ Modulate func(msg *Message) ModulateResult

+ type ModulatorArray []Modulator

+ func (modulators ModulatorArray) Modulate(msg *Message) ModulateResult

type Plugin

+ func NewPluginWithConfig(config PluginConfig) (Plugin, error)

type PluginConfig

+ func NewNestedPluginConfig(defaultTypename string, values tcontainer.MarshalMap) (PluginConfig, error)

+ type PluginConfigReader struct

+ Errors *tgo.ErrorStack

+ WithError PluginConfigReaderWithError

+ func NewPluginConfigReader(config *PluginConfig) PluginConfigReader

+ func NewPluginConfigReaderFromReader(reader PluginConfigReaderWithError) PluginConfigReader

+ func (reader *PluginConfigReader) Configure(item interface{}) error

+ func (reader *PluginConfigReader) GetArray(key string, defaultValue []interface{}) []interface{}

+ func (reader *PluginConfigReader) GetBool(key string, defaultValue bool) bool

+ func (reader *PluginConfigReader) GetFilterArray(key string, logger logrus.FieldLogger, defaultValue FilterArray) FilterArray

+ func (reader *PluginConfigReader) GetFormatterArray(key string, logger logrus.FieldLogger, defaultValue FormatterArray) FormatterArray

+ func (reader *PluginConfigReader) GetID() string

+ func (reader *PluginConfigReader) GetInt(key string, defaultValue int64) int64

+ func (reader *PluginConfigReader) GetLogger() logrus.FieldLogger

+ func (reader *PluginConfigReader) GetMap(key string, defaultValue tcontainer.MarshalMap) tcontainer.MarshalMap

+ func (reader *PluginConfigReader) GetModulatorArray(key string, logger logrus.FieldLogger, defaultValue ModulatorArray) ModulatorArray

+ func (reader *PluginConfigReader) GetPlugin(key string, defaultType string, defaultConfig tcontainer.MarshalMap) Plugin

+ func (reader *PluginConfigReader) GetPluginArray(key string, defaultValue []Plugin) []Plugin

+ func (reader *PluginConfigReader) GetStreamArray(key string, defaultValue []MessageStreamID) []MessageStreamID

+ func (reader *PluginConfigReader) GetStreamID(key string, defaultValue MessageStreamID) MessageStreamID

+ func (reader *PluginConfigReader) GetStreamMap(key string, defaultValue string) map[MessageStreamID]string

+ func (reader *PluginConfigReader) GetStreamRoutes(key string, defaultValue map[MessageStreamID][]MessageStreamID) map[MessageStreamID][]MessageStreamID

+ func (reader *PluginConfigReader) GetString(key string, defaultValue string) string

+ func (reader *PluginConfigReader) GetStringArray(key string, defaultValue []string) []string

+ func (reader *PluginConfigReader) GetStringMap(key string, defaultValue map[string]string) map[string]string

+ func (reader *PluginConfigReader) GetSubLogger(subScope string) logrus.FieldLogger

+ func (reader *PluginConfigReader) GetTypename() string

+ func (reader *PluginConfigReader) GetURL(key string, defaultValue string) *url.URL

+ func (reader *PluginConfigReader) GetUint(key string, defaultValue uint64) uint64

+ func (reader *PluginConfigReader) GetValue(key string, defaultValue interface{}) interface{}

+ func (reader *PluginConfigReader) HasValue(key string) bool

+ type PluginConfigReaderWithError struct

+ func NewPluginConfigReaderWithError(config *PluginConfig) PluginConfigReaderWithError

+ func (reader PluginConfigReaderWithError) GetArray(key string, defaultValue []interface{}) ([]interface{}, error)

+ func (reader PluginConfigReaderWithError) GetBool(key string, defaultValue bool) (bool, error)

+ func (reader PluginConfigReaderWithError) GetFilterArray(key string, logger logrus.FieldLogger, defaultValue FilterArray) (FilterArray, error)

+ func (reader PluginConfigReaderWithError) GetFloat(key string, defaultValue float64) (float64, error)

+ func (reader PluginConfigReaderWithError) GetFormatterArray(key string, logger logrus.FieldLogger, defaultValue FormatterArray) (FormatterArray, error)

+ func (reader PluginConfigReaderWithError) GetID() string

+ func (reader PluginConfigReaderWithError) GetInt(key string, defaultValue int64) (int64, error)

+ func (reader PluginConfigReaderWithError) GetLogger() logrus.FieldLogger

+ func (reader PluginConfigReaderWithError) GetMap(key string, defaultValue tcontainer.MarshalMap) (tcontainer.MarshalMap, error)

+ func (reader PluginConfigReaderWithError) GetModulatorArray(key string, logger logrus.FieldLogger, defaultValue ModulatorArray) (ModulatorArray, error)

+ func (reader PluginConfigReaderWithError) GetPlugin(key string, defaultType string, defaultConfig tcontainer.MarshalMap) (Plugin, error)

+ func (reader PluginConfigReaderWithError) GetPluginArray(key string, defaultValue []Plugin) ([]Plugin, error)

+ func (reader PluginConfigReaderWithError) GetStreamArray(key string, defaultValue []MessageStreamID) ([]MessageStreamID, error)

+ func (reader PluginConfigReaderWithError) GetStreamID(key string, defaultValue MessageStreamID) (MessageStreamID, error)

+ func (reader PluginConfigReaderWithError) GetStreamMap(key string, defaultValue string) (map[MessageStreamID]string, error)

+ func (reader PluginConfigReaderWithError) GetStreamRoutes(key string, defaultValue map[MessageStreamID][]MessageStreamID) (map[MessageStreamID][]MessageStreamID, error)

+ func (reader PluginConfigReaderWithError) GetString(key string, defaultValue string) (string, error)

+ func (reader PluginConfigReaderWithError) GetStringArray(key string, defaultValue []string) ([]string, error)

+ func (reader PluginConfigReaderWithError) GetStringMap(key string, defaultValue map[string]string) (map[string]string, error)

+ func (reader PluginConfigReaderWithError) GetSubLogger(subScope string) logrus.FieldLogger

+ func (reader PluginConfigReaderWithError) GetTypename() string

+ func (reader PluginConfigReaderWithError) GetURL(key string, defaultValue string) (*url.URL, error)

+ func (reader PluginConfigReaderWithError) GetUint(key string, defaultValue uint64) (uint64, error)

+ func (reader PluginConfigReaderWithError) GetValue(key string, defaultValue interface{}) interface{}

+ func (reader PluginConfigReaderWithError) HasValue(key string) bool

+ type PluginMetric struct

+ func (metric *PluginMetric) DecWorker()

+ func (metric *PluginMetric) IncWorker()

+ func (metric *PluginMetric) Init()

+ func (metric *PluginMetric) UpdateStateMetric(prevState, nextState string)

type PluginRunState

+ func (state *PluginRunState) GetStateString() string

+ type PluginStructTag reflect.StructTag

+ func (tag PluginStructTag) GetBool() bool

+ func (tag PluginStructTag) GetByteArray() []byte

+ func (tag PluginStructTag) GetInt() int64

+ func (tag PluginStructTag) GetMetricScale() int64

+ func (tag PluginStructTag) GetStream() MessageStreamID

+ func (tag PluginStructTag) GetStreamArray() []MessageStreamID

+ func (tag PluginStructTag) GetString() string

+ func (tag PluginStructTag) GetStringArray() []string

+ func (tag PluginStructTag) GetUint() uint64

type Producer

+ GetShutdownTimeout func() time.Duration

+ type Router interface

+ AddProducer func(producers ...Producer)

+ Enqueue func(msg *Message) error

+ GetID func() string

+ GetStreamID func() MessageStreamID

+ GetTimeout func() time.Duration

+ Start func() error

+ type ScopedLogger interface

+ GetLogger func() logrus.FieldLogger

+ type ScopedModulator interface

+ SetLogger func(logger logrus.FieldLogger)

type SerializedMessage

+ OrigStreamID *uint64

+ Original *SerializedMessageData

+ func (*SerializedMessage) Descriptor() ([]byte, []int)

+ func (m *SerializedMessage) GetOrigStreamID() uint64

+ func (m *SerializedMessage) GetOriginal() *SerializedMessageData

+ type SerializedMessageData struct

+ Data []byte

+ Metadata map[string][]byte

+ XXX_unrecognized []byte

+ func (*SerializedMessageData) Descriptor() ([]byte, []int)

+ func (*SerializedMessageData) ProtoMessage()

+ func (m *SerializedMessageData) GetData() []byte

+ func (m *SerializedMessageData) GetMetadata() map[string][]byte

+ func (m *SerializedMessageData) Reset()

+ func (m *SerializedMessageData) String() string

+ type SetAppliedContent func(msg *Message, content []byte)

+ func GetAppliedContentSetFunction(applyTo string) SetAppliedContent

+ type SimpleConsumer struct

+ Logger logrus.FieldLogger

+ func (cons *SimpleConsumer) AddHealthCheckAt(path string, callback thealthcheck.CallbackFunc)

+ func (cons *SimpleConsumer) AddMainWorker(workers *sync.WaitGroup)

+ func (cons *SimpleConsumer) AddWorker()

+ func (cons *SimpleConsumer) Configure(conf PluginConfigReader)

+ func (cons *SimpleConsumer) Control() chan<- PluginControl

+ func (cons *SimpleConsumer) ControlLoop()

+ func (cons *SimpleConsumer) Enqueue(data []byte)

+ func (cons *SimpleConsumer) EnqueueWithMetadata(data []byte, metaData Metadata)

+ func (cons *SimpleConsumer) GetID() string

+ func (cons *SimpleConsumer) GetLogger() logrus.FieldLogger

+ func (cons *SimpleConsumer) GetShutdownTimeout() time.Duration

+ func (cons *SimpleConsumer) GetState() PluginState

+ func (cons *SimpleConsumer) IsActive() bool

+ func (cons *SimpleConsumer) IsActiveOrStopping() bool

+ func (cons *SimpleConsumer) IsBlocked() bool

+ func (cons *SimpleConsumer) IsStopping() bool

+ func (cons *SimpleConsumer) SetPrepareStopCallback(onPrepareStop func())

+ func (cons *SimpleConsumer) SetRollCallback(onRoll func())

+ func (cons *SimpleConsumer) SetStopCallback(onStop func())

+ func (cons *SimpleConsumer) SetWorkerWaitGroup(workers *sync.WaitGroup)

+ func (cons *SimpleConsumer) TickerControlLoop(interval time.Duration, onTick func())

+ func (cons *SimpleConsumer) WorkerDone()

+ type SimpleFilter struct

+ Logger logrus.FieldLogger

+ func (filter *SimpleFilter) Configure(conf PluginConfigReader) error

+ func (filter *SimpleFilter) GetFilterResultMessageReject() FilterResult

+ func (filter *SimpleFilter) GetLogger() logrus.FieldLogger

+ func (filter *SimpleFilter) SetLogger(logger logrus.FieldLogger)

+ type SimpleFormatter struct

+ GetAppliedContent GetAppliedContent

+ Logger logrus.FieldLogger

+ SetAppliedContent SetAppliedContent

+ SkipIfEmpty bool

+ func (format *SimpleFormatter) CanBeApplied(msg *Message) bool

+ func (format *SimpleFormatter) Configure(conf PluginConfigReader)

+ func (format *SimpleFormatter) GetLogger() logrus.FieldLogger

+ func (format *SimpleFormatter) SetLogger(logger logrus.FieldLogger)

+ type SimpleProducer struct

+ Logger logrus.FieldLogger

+ func (prod *SimpleProducer) AddHealthCheck(callback thealthcheck.CallbackFunc)

+ func (prod *SimpleProducer) AddHealthCheckAt(path string, callback thealthcheck.CallbackFunc)

+ func (prod *SimpleProducer) AddMainWorker(workers *sync.WaitGroup)

+ func (prod *SimpleProducer) AddWorker()

+ func (prod *SimpleProducer) Configure(conf PluginConfigReader)

+ func (prod *SimpleProducer) Control() chan<- PluginControl

+ func (prod *SimpleProducer) ControlLoop()

+ func (prod *SimpleProducer) GetID() string

+ func (prod *SimpleProducer) GetLogger() logrus.FieldLogger

+ func (prod *SimpleProducer) GetShutdownTimeout() time.Duration

+ func (prod *SimpleProducer) GetState() PluginState

+ func (prod *SimpleProducer) HasContinueAfterModulate(msg *Message) bool

+ func (prod *SimpleProducer) IsActive() bool

+ func (prod *SimpleProducer) IsActiveOrStopping() bool

+ func (prod *SimpleProducer) IsBlocked() bool

+ func (prod *SimpleProducer) IsStopping() bool

+ func (prod *SimpleProducer) Modulate(msg *Message) ModulateResult

+ func (prod *SimpleProducer) SetPrepareStopCallback(onPrepareStop func())

+ func (prod *SimpleProducer) SetRollCallback(onRoll func())

+ func (prod *SimpleProducer) SetStopCallback(onStop func())

+ func (prod *SimpleProducer) SetWorkerWaitGroup(workers *sync.WaitGroup)

+ func (prod *SimpleProducer) Streams() []MessageStreamID

+ func (prod *SimpleProducer) TickerControlLoop(interval time.Duration, onTimeOut func())

+ func (prod *SimpleProducer) TryFallback(msg *Message)

+ func (prod *SimpleProducer) WorkerDone()

+ type SimpleRouter struct

+ Logger logrus.FieldLogger

+ Producers []Producer

+ func (router *SimpleRouter) AddHealthCheck(callback thealthcheck.CallbackFunc)

+ func (router *SimpleRouter) AddHealthCheckAt(path string, callback thealthcheck.CallbackFunc)

+ func (router *SimpleRouter) AddProducer(producers ...Producer)

+ func (router *SimpleRouter) Configure(conf PluginConfigReader)

+ func (router *SimpleRouter) GetID() string

+ func (router *SimpleRouter) GetLogger() logrus.FieldLogger

+ func (router *SimpleRouter) GetProducers() []Producer

+ func (router *SimpleRouter) GetStreamID() MessageStreamID

+ func (router *SimpleRouter) GetTimeout() time.Duration

+ func (router *SimpleRouter) Modulate(msg *Message) ModulateResult

+ type StreamMetric struct

+ func GetStreamMetric(streamID MessageStreamID) StreamMetric

+ func (metric *StreamMetric) CountMessageDiscarded()

+ func (metric *StreamMetric) CountMessageRouted()

v0.5.0-rc3.1

Aug 15, 2017

v0.5.0-rc3

Aug 15, 2017

v0.5.0-rc2

Aug 10, 2017

v0.5.0-rc1

Aug 7, 2017

v0.4.5

Apr 3, 2017

v0.4.4

Aug 16, 2016

v0.4.3

Apr 29, 2016

Changes in this version

+ var InvalidStreamID = MessageStreamID(0)

v0.4.2

Feb 16, 2016

Changes in this version

+ const PluginStateInitializing

type MessageBatch

+ func (batch *MessageBatch) AfterFlushDo(callback func() error) error

v0.4.1

Nov 10, 2015

Changes in this version

+ const PluginControlFuseActive

+ const PluginControlFuseBurn

type ConsumerBase

+ func (cons *ConsumerBase) IsFuseBurned() bool

+ func (cons *ConsumerBase) SetFuseActiveCallback(onFuseActive func())

+ func (cons *ConsumerBase) SetFuseBurnedCallback(onFuseBurned func())

+ func (cons *ConsumerBase) WaitOnFuse()

type ProducerBase

+ func (prod *ProducerBase) GetFuse() *shared.Fuse

+ func (prod *ProducerBase) SetCheckFuseCallback(onCheckFuse func() bool)

v0.4.0

Aug 28, 2015

Changes in this version

+ const MessageStateDiscard

+ const MessageStateOk

+ const MessageStateTimeout

+ const PluginControlStopConsumer

+ const PluginControlStopProducer

+ const PluginStateActive

+ const PluginStateDead

+ const PluginStateStopping

+ const PluginStateWaiting

+ var PluginRegistry = pluginRegistry

+ func CountDiscardedMessage()

+ func CountDroppedMessage()

+ func CountFilteredMessage()

+ func CountNoRouteForMessage()

+ func CountProcessedMessage()

+ type AssemblyFunc func([]Message)

type ConsumerBase

+ func (cons *ConsumerBase) ControlLoop()

+ func (cons *ConsumerBase) GetState() PluginState

+ func (cons *ConsumerBase) IsActive() bool

+ func (cons *ConsumerBase) IsActiveOrStopping() bool

+ func (cons *ConsumerBase) IsBlocked() bool

+ func (cons *ConsumerBase) IsStopping() bool

+ func (cons *ConsumerBase) SetRollCallback(onRoll func())

+ func (cons *ConsumerBase) SetStopCallback(onStop func())

+ type Distributor func(msg Message)

type LogConsumer

+ func (cons *LogConsumer) GetState() PluginState

type Message

+ PrevStreamID MessageStreamID

+ func DeserializeMessage(data []byte) (Message, error)

+ func (msg Message) Route(targetID MessageStreamID)

+ func (msg Message) Serialize() ([]byte, error)

type MessageBatch

+ func (batch *MessageBatch) AppendOrBlock(msg Message) bool

+ func (batch *MessageBatch) AppendOrFlush(msg Message, flushBuffer func(), canBlock func() bool, drop func(Message))

+ func (batch *MessageBatch) Close(assemble AssemblyFunc, timeout time.Duration)

+ func (batch *MessageBatch) Len() int

+ func (batch MessageBatch) IsClosed() bool

type MessageSource

+ IsActive func() bool

+ IsBlocked func() bool

+ type MessageState int

type PluginConfig

+ ID string

+ func (conf PluginConfig) GetStreamArray(key string, defaultValue []MessageStreamID) []MessageStreamID

type PluginRunState

+ func NewPluginRunState() *PluginRunState

+ func (state *PluginRunState) GetState() PluginState

+ func (state *PluginRunState) SetState(nextState PluginState)

+ type PluginState int32

+ type PluginWithState interface

+ GetState func() PluginState

type Producer

+ AddDependency func(Producer)

+ DependsOn func(Producer) bool

+ GetDropStreamID func() MessageStreamID

type ProducerBase

+ func (prod *ProducerBase) Accepts(msg Message) bool

+ func (prod *ProducerBase) AddDependency(dep Producer)

+ func (prod *ProducerBase) CloseMessageChannel(handleMessage func(msg Message)) bool

+ func (prod *ProducerBase) ControlLoop()

+ func (prod *ProducerBase) DependsOn(dep Producer) bool

+ func (prod *ProducerBase) Drop(msg Message)

+ func (prod *ProducerBase) GetDropStreamID() MessageStreamID

+ func (prod *ProducerBase) GetFilter() Filter

+ func (prod *ProducerBase) GetState() PluginState

+ func (prod *ProducerBase) IsActive() bool

+ func (prod *ProducerBase) IsActiveOrStopping() bool

+ func (prod *ProducerBase) IsBlocked() bool

+ func (prod *ProducerBase) IsStopping() bool

+ func (prod *ProducerBase) MessageControlLoop(onMessage func(Message))

+ func (prod *ProducerBase) SetRollCallback(onRoll func())

+ func (prod *ProducerBase) SetStopCallback(onStop func())

+ func (prod *ProducerBase) TickerMessageControlLoop(onMessage func(Message), interval time.Duration, onTimeOut func())

+ func (prod *ProducerBase) WaitForDependencies(waitForState PluginState, timeout time.Duration)

+ func (prod ProducerBase) GetShutdownTimeout() time.Duration

+ type SerializedMessage struct

+ Data []byte

+ PrevStreamID *uint64

+ Sequence *uint64

+ StreamID *uint64

+ Timestamp *int64

+ XXX_unrecognized []byte

+ func (*SerializedMessage) ProtoMessage()

+ func (m *SerializedMessage) GetData() []byte

+ func (m *SerializedMessage) GetPrevStreamID() uint64

+ func (m *SerializedMessage) GetSequence() uint64

+ func (m *SerializedMessage) GetStreamID() uint64

+ func (m *SerializedMessage) GetTimestamp() int64

+ func (m *SerializedMessage) Reset()

+ func (m *SerializedMessage) String() string

type Stream

+ Flush func()

+ GetBoundStreamID func() MessageStreamID

+ GetProducers func() []Producer

type StreamBase

+ Timeout *time.Duration

+ func (stream *StreamBase) Broadcast(msg Message)

+ func (stream *StreamBase) ConfigureStream(conf PluginConfig, distribute Distributor) error

+ func (stream *StreamBase) Flush()

+ func (stream *StreamBase) GetBoundStreamID() MessageStreamID

+ func (stream *StreamBase) GetProducers() []Producer

+ func (stream *StreamBase) Route(msg Message, targetID MessageStreamID)

+ type WriterAssembly struct

+ func NewWriterAssembly(writer io.Writer, flush func(Message), formatter Formatter) WriterAssembly

+ func (asm *WriterAssembly) Flush(messages []Message)

+ func (asm *WriterAssembly) SetErrorHandler(handleError func(error) bool)

+ func (asm *WriterAssembly) SetFlush(flush func(Message))

+ func (asm *WriterAssembly) SetValidator(validate func() bool)

+ func (asm *WriterAssembly) SetWriter(writer io.Writer)

+ func (asm *WriterAssembly) Write(messages []Message)

v0.3.2

Jun 22, 2015

Changes in this version

+ const DroppedStream

+ const LogInternalStream

+ const PluginControlRoll

+ const PluginControlStop

+ const WildcardStream

+ var DroppedStreamID = GetStreamID(DroppedStream)

+ var LogInternalStreamID = GetStreamID(LogInternalStream)

+ var MessageCount = uint32(0)

+ var StreamTypes = StreamRegistry

+ var WildcardStreamID = GetStreamID(WildcardStream)

+ func EnableRetryQueue(size int)

+ func GetAndResetMessageCount() uint32

+ func GetRetryQueue() <-chan Message

+ type AsyncMessageSource interface

+ EnqueueResponse func(msg Message)

+ type Config struct

+ Plugins []PluginConfig

+ Values []map[string]shared.MarshalMap

+ func ReadConfig(path string) (*Config, error)

+ type Consumer interface

+ Consume func(*sync.WaitGroup)

+ Control func() chan<- PluginControl

+ Streams func() []MessageStreamID

+ type ConsumerBase struct

+ func (cons *ConsumerBase) Configure(conf PluginConfig) error

+ func (cons *ConsumerBase) Control() chan<- PluginControl

+ func (cons *ConsumerBase) DefaultControlLoop(onRoll func())

+ func (cons *ConsumerBase) Enqueue(data []byte, sequence uint64)

+ func (cons *ConsumerBase) EnqueueCopy(data []byte, sequence uint64)

+ func (cons *ConsumerBase) EnqueueMessage(msg Message)

+ func (cons *ConsumerBase) ProcessCommand(command PluginControl, onRoll func()) bool

+ func (cons *ConsumerBase) Streams() []MessageStreamID

+ func (cons *ConsumerBase) TickerControlLoop(interval time.Duration, onRoll func(), onTick func())

+ func (cons ConsumerBase) AddMainWorker(workers *sync.WaitGroup)

+ func (cons ConsumerBase) AddWorker()

+ func (cons ConsumerBase) SetWorkerWaitGroup(workers *sync.WaitGroup)

+ func (cons ConsumerBase) WorkerDone()

+ type ConsumerError struct

+ func NewConsumerError(args ...interface{}) ConsumerError

+ func (err ConsumerError) Error() string

+ type Filter interface

+ Accepts func(msg Message) bool

+ type Formatter interface

+ Format func(msg Message) ([]byte, MessageStreamID)

+ type LinkableMessageSource interface

+ IsLinked func() bool

+ Link func(pipe interface{})

+ type LogConsumer struct

+ func (cons *LogConsumer) Configure(conf PluginConfig) error

+ func (cons *LogConsumer) Consume(threads *sync.WaitGroup)

+ func (cons *LogConsumer) Control() chan<- PluginControl

+ func (cons *LogConsumer) Streams() []MessageStreamID

+ func (cons LogConsumer) IsPaused() bool

+ func (cons LogConsumer) Pause()

+ func (cons LogConsumer) Resume()

+ func (cons LogConsumer) Write(data []byte) (int, error)

+ type MappedStream struct

+ Stream Stream

+ StreamID MessageStreamID

+ type Message struct

+ Data []byte

+ Sequence uint64

+ Source MessageSource

+ StreamID MessageStreamID

+ Timestamp time.Time

+ func NewMessage(source MessageSource, data []byte, sequence uint64) Message

+ func (msg Message) Drop(timeout time.Duration)

+ func (msg Message) Enqueue(channel chan<- Message, timeout time.Duration)

+ func (msg Message) Retry(timeout time.Duration)

+ func (msg Message) String() string

+ type MessageBatch struct

+ func NewMessageBatch(size int, format Formatter) *MessageBatch

+ func (batch *MessageBatch) Append(msg Message) bool

+ func (batch *MessageBatch) Flush(resource io.Writer, validate func() bool, onError func(error) bool)

+ func (batch *MessageBatch) Touch()

+ func (batch *MessageBatch) WaitForFlush(timeout time.Duration)

+ func (batch MessageBatch) IsEmpty() bool

+ func (batch MessageBatch) ReachedSizeThreshold(size int) bool

+ func (batch MessageBatch) ReachedTimeThreshold(timeout time.Duration) bool

+ type MessageSource interface

+ type MessageStreamID uint64

+ func GetStreamID(stream string) MessageStreamID

+ type Plugin interface

+ Configure func(conf PluginConfig) error

+ func NewPlugin(config PluginConfig) (Plugin, error)

+ func NewPluginWithType(typename string, config PluginConfig) (Plugin, error)

+ type PluginConfig struct

+ Enable bool

+ Instances int

+ Settings shared.MarshalMap

+ Stream []string

+ Typename string

+ func NewPluginConfig(typename string) PluginConfig

+ func (conf *PluginConfig) Read(values shared.MarshalMap)

+ func (conf PluginConfig) GetBool(key string, defaultValue bool) bool

+ func (conf PluginConfig) GetInt(key string, defaultValue int) int

+ func (conf PluginConfig) GetStreamMap(key string, defaultValue string) map[MessageStreamID]string

+ func (conf PluginConfig) GetStreamRoutes(key string) map[MessageStreamID][]MessageStreamID

+ func (conf PluginConfig) GetString(key string, defaultValue string) string

+ func (conf PluginConfig) GetStringArray(key string, defaultValue []string) []string

+ func (conf PluginConfig) GetStringMap(key string, defaultValue map[string]string) map[string]string

+ func (conf PluginConfig) GetValue(key string, defaultValue interface{}) interface{}

+ func (conf PluginConfig) HasValue(key string) bool

+ func (conf PluginConfig) Override(key string, value interface{})

+ func (conf PluginConfig) Validate() bool

+ type PluginControl int

+ type PluginRunState struct

+ func (state *PluginRunState) AddWorker()

+ func (state *PluginRunState) IsPaused() bool

+ func (state *PluginRunState) Pause()

+ func (state *PluginRunState) Resume()

+ func (state *PluginRunState) SetWorkerWaitGroup(workers *sync.WaitGroup)

+ func (state *PluginRunState) WorkerDone()

+ type Producer interface

+ Control func() chan<- PluginControl

+ Enqueue func(msg Message)

+ Produce func(workers *sync.WaitGroup)

+ Streams func() []MessageStreamID

+ type ProducerBase struct

+ func (prod *ProducerBase) Close(onMessage func(msg Message))

+ func (prod *ProducerBase) Configure(conf PluginConfig) error

+ func (prod *ProducerBase) Control() chan<- PluginControl

+ func (prod *ProducerBase) DefaultControlLoop(onMessage func(msg Message), onRoll func())

+ func (prod *ProducerBase) Enqueue(msg Message)

+ func (prod *ProducerBase) Format(msg Message) ([]byte, MessageStreamID)

+ func (prod *ProducerBase) GetFormatter() Formatter

+ func (prod *ProducerBase) Messages() chan<- Message

+ func (prod *ProducerBase) PauseAllStreams(capacity int)

+ func (prod *ProducerBase) ProcessCommand(command PluginControl, onRoll func()) bool

+ func (prod *ProducerBase) ResumeAllStreams()

+ func (prod *ProducerBase) Streams() []MessageStreamID

+ func (prod *ProducerBase) TickerControlLoop(interval time.Duration, onMessage func(msg Message), onRoll func(), ...)

+ func (prod ProducerBase) AddMainWorker(workers *sync.WaitGroup)

+ func (prod ProducerBase) AddWorker()

+ func (prod ProducerBase) GetTimeout() time.Duration

+ func (prod ProducerBase) Next() (Message, bool)

+ func (prod ProducerBase) NextNonBlocking(onMessage func(msg Message)) bool

+ func (prod ProducerBase) SetWorkerWaitGroup(workers *sync.WaitGroup)

+ func (prod ProducerBase) WorkerDone()

+ type ProducerError struct

+ func NewProducerError(args ...interface{}) ProducerError

+ func (err ProducerError) Error() string

+ type SerialMessageSource interface

+ ResponseDone func()

+ type Stream interface

+ AddProducer func(producers ...Producer)

+ Enqueue func(msg Message)

+ Pause func(capacity int)

+ Resume func()

+ type StreamBase struct

+ Distribute func(msg Message)

+ Filter Filter

+ Format Formatter

+ Producers []Producer

+ func (stream *StreamBase) AddProducer(producers ...Producer)

+ func (stream *StreamBase) Configure(conf PluginConfig) error

+ func (stream *StreamBase) Enqueue(msg Message)

+ func (stream *StreamBase) Pause(capacity int)

+ func (stream *StreamBase) Resume()

+ type StreamRegistry struct

+ func (registry *StreamRegistry) GetStreamOrFallback(streamID MessageStreamID) Stream

+ func (registry *StreamRegistry) Register(stream Stream, streamID MessageStreamID)

+ func (registry *StreamRegistry) RegisterWildcardProducer(producers ...Producer)

+ func (registry StreamRegistry) AddWildcardProducersToStream(stream Stream)

+ func (registry StreamRegistry) ForEachStream(callback func(streamID MessageStreamID, stream Stream))

+ func (registry StreamRegistry) GetStream(id MessageStreamID) Stream

+ func (registry StreamRegistry) GetStreamByName(name string) Stream

+ func (registry StreamRegistry) GetStreamName(streamID MessageStreamID) string

+ func (registry StreamRegistry) IsStreamRegistered(id MessageStreamID) bool

?	: This menu
/	: Search site
f or F	: Jump to
y or Y	: Canonical URL