dbq

README

dbq

a small distributed queue using database as backend

Features

• Create a new message queue.
• Delete an existing message queue.
• Push a new message to a queue.
• Pull messages from a queue.
• Retrieve a specific message from a queue.
• Update a message in a queue.
• Clear a queue (delete all messages).

Installation

go build cmd/httpserver.go 

Configuration

The program uses a configuration file in TOML format to specify the TiDB connection and server settings. By default, the program looks for a file named config.toml in the current directory. You can specify a different configuration file using the -c command-line flag.

Go to tidb.cloud to get a free database cluster

The default configuration file (config.toml) has the following structure:

[tidb]
username = "root"
password = ""
host = ""
port = 4000
database = "dbq"
tls_enabled = true

[tidb.tls_config]
server_name = ""

[server]
port = 8080
host = "localhost"
debug = true

• tidb: TiDB database configuration.

  • username: TiDB username.
  • password: TiDB password.
  • host: TiDB host.
  • port: TiDB port.
  • database: TiDB database name.
  • tls_enabled: Enable TLS connection to TiDB (true/false).
  • tls_config: TLS configuration.
    • server_name: Server name for TLS handshake.

• server: API server configuration.

  • port: API server port.
  • host: API server host.
  • debug: Enable debug mode (true/false).

Usage

1. Start the program:

   ./httpserver -c config.toml
   

   or

   AUTH_TOKEN=<your-auth-token> ./httpserver -c config.toml
   

   This will start the API server using the configuration specified in the config.toml file.

2. The API server will listen for incoming requests at the configured host and port (e.g., localhost:8080).

3. You can interact with the message queue API using any HTTP client or tool, such as cURL or Postman.

4. Refer to the API endpoints below for the available operations:

   • POST /q/:name: Create a new message queue with the specified name.
   • DELETE /q/:name: Delete an existing message queue with the specified name.
   • POST /q/:name/push: Push a new message to the queue with the specified name.
   • GET /q/:name/pull?limit=n: Pull up to n messages from the queue with the specified name.
   • GET /q/:name/msg/:id: Retrieve the message with the specified ID from the queue with the specified name.
   • PUT /q/:name/msg/:id: Update the message with the specified ID in the queue with the specified name.
   • DELETE /q/:name/truncate: Clear (delete all messages) from the queue with the specified name.

Example

1. Create a new message queue:

   http POST http://localhost:8080/q/myqueue
   

2. Delete an existing message queue:

   http DELETE http://localhost:8080/q/myqueue
   

3. Push a new message to a queue:

   echo '{"message": "Hello, World!"}' | http POST http://localhost:8080/q/myqueue/push
   

4. Pull messages from a queue:

   http GET http://localhost:8080/q/myqueue/pull?limit=10
   

5. Retrieve a specific message from a queue:

   http GET http://localhost:8080/q/myqueue/msg/12345
   

6. Update a message in a queue:

   echo '{"data": "Updated message"}' | http PUT http://localhost:8080/q/myqueue/msg/12345
   

7. Clear a queue (delete all messages):

   http DELETE http://localhost:8080/q/myqueue/truncate
   

Documentation

Index

• Constants
• func DB() *sql.DB
• func InitDB(db *sql.DB)
• type Msg
  • func NewMsg(data []byte) *Msg
• type MsgStatus
  • func (s MsgStatus) String() string
• type Q
  • func (q *Q) Clear() error
  • func (q *Q) Create() error
  • func (q *Q) Drop() error
  • func (Q) Exists(name string) (bool, error)
  • func (q *Q) GetMsgByID(id int64) (*Msg, error)
  • func (Q) New(name string) (*Q, error)
  • func (q *Q) NumPending() (int, error)
  • func (q *Q) Pull(limit int) ([]*Msg, error)
  • func (q *Q) Push(msgs []*Msg) error
  • func (q *Q) TableName() string
  • func (q *Q) UpdateMsg(msg *Msg) error

Constants

const MAX_PAYLOAD_SIZE = 1024 * 1024 * 10

Functions

func DB

func DB() *sql.DB

func InitDB

func InitDB(db *sql.DB)

Types

type Msg

type Msg struct {
	ID         int64     `json:"id"` // auto increment
	Data       []byte    `json:"data"`
	Status     MsgStatus `json:"status"`
	RetCode    *int      `json:"ret_code"`
	Progress   []byte    `json:"progress"`
	Ret        []byte    `json:"ret"`
	Error      []byte    `json:"error"`
	CreatedAt  time.Time `json:"created_at"`
	ScheduleAt time.Time `json:"schedule_at"`
	UpdatedAt  time.Time `json:"updated_at"`

	Q *Q `json:"-"`
}

func NewMsg

func NewMsg(data []byte) *Msg

type MsgStatus

type MsgStatus int

const (
	MsgStatusNil MsgStatus = iota
	MsgStatusPending
	MsgStatusDispatched
	MsgStatusRunning
	MsgStatusCanceled
	MsgStatusFinished
	MsgStatusFailed
)

func (MsgStatus) String

func (s MsgStatus) String() string

type Q

type Q struct {
	Name string `json:"name"`
}

func (*Q) Clear

func (q *Q) Clear() error

func (*Q) Create

func (q *Q) Create() error

func (*Q) Drop

func (q *Q) Drop() error

func (Q) Exists

func (Q) Exists(name string) (bool, error)

func (*Q) GetMsgByID

func (q *Q) GetMsgByID(id int64) (*Msg, error)

func (Q) New

func (Q) New(name string) (*Q, error)

func (*Q) NumPending

func (q *Q) NumPending() (int, error)

func (*Q) Pull

func (q *Q) Pull(limit int) ([]*Msg, error)

func (*Q) Push

func (q *Q) Push(msgs []*Msg) error

func (*Q) TableName

func (q *Q) TableName() string

func (*Q) UpdateMsg

func (q *Q) UpdateMsg(msg *Msg) error