transporter

README

The Transporter

Usage

Run the binary with the configuration options below

transporter server <CONFIG OPTIONS>

or run the docker container with the environment variables below

Configuration

Transporter can be configured with command line flags and environment variables.

Command Line Flags

Flag	Options	Multiple Allowed	Description	Environment Variable
--log.level	debug, info, warn, error	No	The log level to use.	TRANSPORTER_LOG_LEVEL
--stats		No	The stats dsn to connect to	TRANSPORTER_STATS
--stats.tags		Yes	The tags to attach to all metrics	TRANSPORTER_STATS_TAGS
--port		No	The address to bind to for the http server.	TRANSPORTER_PORT
--buffer.size		No	The size of each topics buffer.	TRANSPORTER_BUFFER_SIZE
--kafka.brokers		Yes	The kafka seed brokers connect to. Format: 'ip:port'.	TRANSPORTER_KAFKA_BROKERS
--kafka.group-id		No	The kafka group id to subscribe to.	TRANSPORTER_KAFKA_GROUP_ID
--kafka.topics		Yes	The kafka topics to subscribe to.	TRANSPORTER_KAFKA_TOPICS

Multi value environment variables

When using environment variables where multiple values are allowed, the values should be comma seperated. E.g. --kafka.topics=foo --kafka.topics=bar should become TRANSPORTER_KAFKA_TOPICS=foo,bar.

HTTP Endpoints

GET /health

Gets the current health status of Transporter. Returns a 200 status code if healthy, otherwise a 500 status code

GET /:topic

Gets the next item in the topic queue. A timeout of 100ms is used when waiting for items in the queue. After the timeout a 204 status code is returned. If the topic does not exist, 404 status code is returned

Contributors

We're supposed to tell you how to contribute to transporter here.
Since this is github: You know the drill - open issues, fork, create PRs, ...

License

MIT-License. As is. No warranties whatsoever. Mileage may vary. Batteries not included.

Documentation

Index

• type Application
  • func NewApplication() *Application
  • func (a *Application) Close()
  • func (a *Application) GetNextBatch(ctx context.Context, topic string, count int) ([][]byte, error)
  • func (a *Application) GetNextMessage(ctx context.Context, topic string) ([]byte, error)
  • func (a *Application) IsHealthy() error
• type Consumer

Types

type Application

type Application struct {
	Consumer Consumer
}

Application represents the transporter application.

func NewApplication

func NewApplication() *Application

NewApplication creates an instance of Application.

func (*Application) Close

func (a *Application) Close()

Close closes all application connections.

func (*Application) GetNextBatch

func (a *Application) GetNextBatch(ctx context.Context, topic string, count int) ([][]byte, error)

GetNextBatch gets the next count messages from the queue.

func (*Application) GetNextMessage

func (a *Application) GetNextMessage(ctx context.Context, topic string) ([]byte, error)

GetNextMessage gets the next message from the queue.

func (*Application) IsHealthy

func (a *Application) IsHealthy() error

IsHealthy checks the health of the Application.

type Consumer

type Consumer interface {
	// Close closes the Consumer.
	Close()
	// GetNextMessage gets the next message from the queue.
	GetNextMessage(ctx context.Context, topic string) ([]byte, error)
	// GetNextBatch gets the next count messages from the queue.
	GetNextBatch(ctx context.Context, topic string, count int) ([][]byte, error)
}

Consumer represents a queued Kafka consumer.