README
¶
rbforwarder
rbforwarder is an extensible tool for processing data asynchronously. It allows you to create a custom pipeline in a modular fashion.
You can use rbforwarder to build a pipeline that decodes a JSON, filter
or add fields, encode the data again to JSON and send it using multiple
protocols HTTP, MQTT, AMQP, etc. It's easy to write a pipeline component.
Built-in features
- Support multiple workers for each components.
- Asynchronous report system. Get responses on a separate goroutine.
- Built-in message retrying. rbforwarder can retry messages on fail.
- Instrumentation, to have an overview of how the application is performing.
Components
- Send data to an endpoint
- MQTT
- HTTP
- Kafka
- Decoders / Encoders
- JSON
- Utility
- Limiter
- Batcher (supports deflate)
Road Map
The application is on hard development, breaking changes expected until 1.0.
| Milestone | Feature | Status |
|---|---|---|
| 0.1 | Pipeline | Done |
| 0.2 | Reporter | Done |
| 0.3 | HTTP component | Done |
| 0.4 | Batcher component | Done |
| 0.5 | Limiter component | Done |
| 0.6 | Instrumentation | Pending |
| 0.8 | JSON component | Pending |
| 0.9 | MQTT component | Pending |
| 1.0 | Kafka component | Pending |
Documentation
¶
Index ¶
- Variables
- type Component
- type Config
- type RBForwarder
- func (f *RBForwarder) Close()
- func (f *RBForwarder) GetOrderedReports() <-chan interface{}
- func (f *RBForwarder) GetReports() <-chan interface{}
- func (f *RBForwarder) Produce(data []byte, opts map[string]interface{}, opaque interface{}) error
- func (f *RBForwarder) PushComponents(components []interface{})
- func (f *RBForwarder) Run()
- type Report
Constants ¶
This section is empty.
Variables ¶
var Logger = logrus.NewEntry(log)
Logger for the package
var Version = "0.5"
Version is the current tag
Functions ¶
This section is empty.
Types ¶
type Component ¶
type Component struct {
// contains filtered or unexported fields
}
Component contains information about a pipeline component
type RBForwarder ¶
type RBForwarder struct {
// contains filtered or unexported fields
}
RBForwarder is the main objecto of the package. It has the main methods for send messages and get reports. It has a backend for routing messages between workers
func NewRBForwarder ¶
func NewRBForwarder(config Config) *RBForwarder
NewRBForwarder creates a new Forwarder object
func (*RBForwarder) GetOrderedReports ¶
func (f *RBForwarder) GetOrderedReports() <-chan interface{}
GetOrderedReports is the same as GetReports() but the reports are delivered in order
func (*RBForwarder) GetReports ¶
func (f *RBForwarder) GetReports() <-chan interface{}
GetReports is used by the source to get a report for a sent message. Reports are delivered on the same order that was sent
func (*RBForwarder) Produce ¶
func (f *RBForwarder) Produce(data []byte, opts map[string]interface{}, opaque interface{}) error
Produce is used by the source to send messages to the backend
func (*RBForwarder) PushComponents ¶
func (f *RBForwarder) PushComponents(components []interface{})
PushComponents adds a new component to the pipeline
Source Files
Directories