qrpc

README

qrpc

A small distributed queue messaging service backed by grpc.

I've created the project mainly as an exercise because I wanted to play with grcp. This is not super secure queue service, rather a prototype implemented in go.

Note

For the purpose of the project I also "stole" some ideas :) QRPC stores data using Peter's awesome diskv library. Also multi-master architecture is something what I really like in disque, so I've tried to implement similar mechanism.

Install & Run

$ go get github.com/kuba--/qrpc/qrpc
$ qrpc --help
usage: ./qrpc [flags] [cluster peer(s) ...]
flags:
  -port int
  	port to listen on (default 9033)
  -data string
  	directory in which queue data is stored (default "/tmp/qrpc")
  -cache uint
  	max. cache size (bytes) before an item is evicted. (default 1048576)
  -interval duration
  	cluster watch timer interval. (default 1s)
  -timeout duration
  	cluster request (gossip) timeout. (default 3s)

# start standalone server :9091
$ qrpc -data /tmp/qrpc-9091 -port 9091

# ...join the cluster
$ qrpc -data /tmp/qrpc-9092 -port 9092 127.0.0.1:9091
$ qrpc -data /tmp/qrpc-9093 -port 9093 127.0.0.1:9092

Documentation

Index

• type Config
• type Server
  • func NewServer(cfg *Config) *Server
  • func (s *Server) Erase(ctx context.Context, req *internal.EraseRequest) (*internal.EraseResponse, error)
  • func (s *Server) Join(ctx context.Context, req *internal.JoinRequest) (*internal.JoinResponse, error)
  • func (s *Server) Ping(ctx context.Context, req *internal.PingRequest) (*internal.PingResponse, error)
  • func (s *Server) Receive(ctx context.Context, req *api.ReceiveRequest) (*api.ReceiveResponse, error)
  • func (s *Server) Send(ctx context.Context, req *api.SendRequest) (*api.SendResponse, error)
  • func (s *Server) Start() <-chan error
  • func (s *Server) Stop()
  • func (s *Server) Unjoin(ctx context.Context, req *internal.UnjoinRequest) (*internal.UnjoinResponse, error)
  • func (s *Server) Write(ctx context.Context, req *internal.WriteRequest) (*internal.WriteResponse, error)

Types

type Config

type Config struct {
	Port                  int
	MaxCacheSize          uint64
	DataBasePath          string
	ClusterRequestTimeout time.Duration
	ClusterWatchInterval  time.Duration
	ClusterPeers          []string
}

type Server

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

Server structure represents QRPC node.

func NewServer

func NewServer(cfg *Config) *Server

NewServer creates a qrpc server which has not started to accept requests yet.

func (*Server) Erase

func (s *Server) Erase(ctx context.Context, req *internal.EraseRequest) (*internal.EraseResponse, error)

Erase implements 'internal erase request endpoint'. The endpoint is used to delete the key from the server (becuase data were read from other peer).

func (*Server) Join

func (s *Server) Join(ctx context.Context, req *internal.JoinRequest) (*internal.JoinResponse, error)

Join implements 'internal join request endpoint'. Updates address for requesting peer. Returns list of already known peers.

func (*Server) Ping

func (s *Server) Ping(ctx context.Context, req *internal.PingRequest) (*internal.PingResponse, error)

Ping implements 'internal ping request endpoint'. The endpoint is mostly used by cluster's watcher for service discovery.

func (*Server) Receive

func (s *Server) Receive(ctx context.Context, req *api.ReceiveRequest) (*api.ReceiveResponse, error)

Receive implements 'receive request endpoint'. Data will be read and erased from diskv. The server will also try to erase the read key from cluster's peers.

func (*Server) Send

func (s *Server) Send(ctx context.Context, req *api.SendRequest) (*api.SendResponse, error)

Send implements 'send request endpoint'. Data will be stored in diskv and replicated across cluster's peers.

func (*Server) Start

func (s *Server) Start() <-chan error

Start starts a qrpc server on specified port and tries to connect to existing cluster (if peers were specified). Returns an error channel which will be populated when server fail.

func (*Server) Stop

func (s *Server) Stop()

Stop stops a qrpc server.

func (*Server) Unjoin

func (s *Server) Unjoin(ctx context.Context, req *internal.UnjoinRequest) (*internal.UnjoinResponse, error)

Unjoin implements 'internal unjoin request endpoint'. Deletes requesting peer from lookup table.

func (*Server) Write

func (s *Server) Write(ctx context.Context, req *internal.WriteRequest) (*internal.WriteResponse, error)

Write implements 'internal write request endpoint'. The endpoint is mostly used to replicate data across cluster's peers. If requesting req.Key already exists then data will be overwritten.