haraqa

README

HARAQA - High Availability Routing And Queueing Application

haraqa is designed to be a developer friendly, scalable message queue for data persistence and real-time data streaming between microservices. Haraqa provides high-throughput, low-latency, fault-tolerant pipelines for architectures of any size.

Table of Contents

• About the Project
  • Overview
  • Persistence and Replication
• Getting Started
  • API Docs
  • Server
  • Client
• Contributing
• License

About the Project

Overview

Haraqa is meant for handling and persisting data in a distributed system. One or more servers can be used to send and receive messages. Each server has a set of 'topics', a set of messages stored in the order received.

A Haraqa client can produce and/or consume from a server's topics. Clients can produce messages one at a time or in batches. Clients consume messages by making a request for a specific offset and limit. The messages can be consumed one at a time or in batches.

Persistence and Replication

Each server, after receiving a message from a producer, can save the message to multiple volumes. These volumes are meant to be distributed in the architecture, such as having multiple PersistentVolumes in a Kubernetes cluster, EBS in AWS, or Persistent Disks in Google Cloud. The server reads messages from the last volume when sending to consumer clients.

If a volume is removed or corrupted during a restart the server repopulates the data from the other volumes.

Getting started

API Docs

• Swagger API Documentation
• Redocs API Documentation
• Swagger yaml

The docker server also includes local api documentation at the /docs/redocs and /docs/swagger endpoints.

Server

The recommended deployment strategy is to use Docker

docker run -it -p 4353:4353 -v $PWD/vol1:/vol1 haraqa/haraqa /vol1

docker run -it [port mapping] [volume mounts] haraqa/haraqa [flags] [volumes]

To run from source, navigate to cmd/server and run the main.go file.

cd cmd/server
go run main.go vol1

Flags:

  -http    uint    Port to listen on (default 4353)
  -cache   boolean Enable queue file caching (default true)
  -cors    boolean Enable CORS (default true)
  -docs    boolean Enable Docs pages (default true)
  -entries integer The number of msg entries per queue file before creating a new file (default 5000)
  -limit   integer Default batch limit for consumers (default -1)
  -ballast integer Garbage collection memory ballast size in bytes (default 1073741824)
  -prometheus boolean Enable prometheus metrics (default true)

Volumes:

Volumes will be written to in the order given and recovered from in the reverse order. Consumer requests are read from the last volume. For this reason it's recommended to use a local volume last.

For instance, given

docker run haraqa/haraqa /vol1 /vol2 /vol3

When a client produces a message, it will be written to /vol1, then /vol2, then /vol3. When a client consumes a message, it will be read from /vol3.

During recovery, if data exists in /vol3 it will be replicated to volumes /vol1 and /vol2. If /vol3 is empty, /vol2 will be replicated to /vol1 and /vol3.

Client

go get github.com/haraqa/haraqa

Contributing

We want this project to be the best it can be and all feedback, feature requests or pull requests are welcome.

License

MIT © 2019 haraqa and contributors. See LICENSE for more information.

Documentation

Index

• Variables
• type Client
  • func NewClient(opts ...Option) (*Client, error)
  • func (c *Client) Close() error
  • func (c *Client) Consume(topic string, id int64, limit int) (io.ReadCloser, []int64, error)
  • func (c *Client) ConsumeMsgs(topic string, id int64, limit int) ([][]byte, error)
  • func (c *Client) CreateTopic(topic string) error
  • func (c *Client) DeleteTopic(topic string) error
  • func (c *Client) ListTopics(regex string) ([]string, error)
  • func (c *Client) NewWriter(topic string, delimiter []byte) io.Writer
  • func (c *Client) Produce(topic string, sizes []int64, r io.Reader) error
  • func (c *Client) ProduceMsgs(topic string, msgs ...[]byte) error
  • func (c *Client) WatchTopics(ctx context.Context, topics []string, ch chan<- string) error
• type Option
  • func WithConsumerGroup(group string) Option
  • func WithHTTPClient(client *http.Client) Option
  • func WithURL(url string) Option
• type Writer
  • func (w *Writer) Write(b []byte) (int, error)

Variables

var (
	ErrTopicAlreadyExists = headers.ErrTopicAlreadyExists
	ErrNoContent          = headers.ErrNoContent
	ErrInvalidTopic       = headers.ErrInvalidTopic
)

Types

type Client

type Client struct {
	// contains filtered or unexported fields
}

Client is a lightweight client around the haraqa http api, use NewClient() to create a new client

func NewClient

func NewClient(opts ...Option) (*Client, error)

NewClient creates a new client instance. Any options given override the local defaults

func (*Client) Close

func (c *Client) Close() error

Close closes any active topic watchers

func (*Client) Consume

func (c *Client) Consume(topic string, id int64, limit int) (io.ReadCloser, []int64, error)

Consume reads messages off of a topic starting from id, no more than the given limit is returned. If limit is less than 1, the server sets the limit.

func (*Client) ConsumeMsgs

func (c *Client) ConsumeMsgs(topic string, id int64, limit int) ([][]byte, error)

ConsumeMsgs reads messages off of a topic starting from id, no more than the given limit is returned. If limit is less than 1, the server sets the limit.

func (*Client) CreateTopic

func (c *Client) CreateTopic(topic string) error

CreateTopic Creates a new topic. It returns an error if the topic already exists

func (*Client) DeleteTopic

func (c *Client) DeleteTopic(topic string) error

DeleteTopic Delete a topic

func (*Client) ListTopics

func (c *Client) ListTopics(regex string) ([]string, error)

ListTopics Lists all topics, filter by prefix, suffix, and/or a regex expression

func (*Client) NewWriter

func (c *Client) NewWriter(topic string, delimiter []byte) io.Writer

NewWriter creates a new producer consistent with io.Writer

func (*Client) Produce

func (c *Client) Produce(topic string, sizes []int64, r io.Reader) error

Produce sends messages from a reader to the designated topic

func (*Client) ProduceMsgs

func (c *Client) ProduceMsgs(topic string, msgs ...[]byte) error

ProduceMsgs sends the messages to the designated topic

func (*Client) WatchTopics

func (c *Client) WatchTopics(ctx context.Context, topics []string, ch chan<- string) error

WatchTopics opens a websocket to the server to listen for changes to the given topics. It writes the name of any modified topics to the given channel until a context cancellation or an error occurs

type Option

type Option func(*Client) error

Option represents a optional function argument to NewClient

func WithConsumerGroup

func WithConsumerGroup(group string) Option

func WithHTTPClient

func WithHTTPClient(client *http.Client) Option

WithHTTPClient replaces the default http client config when calling NewClient

func WithURL

func WithURL(url string) Option

WithURL replaces the default URL when calling NewClient

type Writer

type Writer struct {
	// contains filtered or unexported fields
}

func (*Writer) Write

func (w *Writer) Write(b []byte) (int, error)

Write produces messages to the haraqa instance separated by the delimiter. This is thread safe