client

README

Go Client

Usage

Publish

package main

import (
	"log"
	"time"

	"github.com/asim/emque/client"
)

func main() {
	tick := time.NewTicker(time.Second)

	for _ = range tick.C {
		if err := client.Publish("foo", []byte(`bar`)); err != nil {
			log.Println(err)
			break
		}
	}
}

Subscribe

package main

import (
	"log"

	"github.com/asim/emque/client"
)

func main() {
	ch, err := client.Subscribe("foo")
	if err != nil {
		log.Println(err)
		return
	}

	for e := range ch {
		log.Println(string(e))
	}

	log.Println("channel closed")
}

New Client

// defaults to MQ server localhost:8081
c := client.New()

gRPC client

import "github.com/asim/emque/client/grpc"

c := grpc.New()

Clustering

Clustering is supported on the client side. Publish/Subscribe operations are performed against all servers.

c := client.New(
	client.WithServers("10.0.0.1:8081", "10.0.0.1:8082", "10.0.0.1:8083"),
)

Sharding

Sharding is supported via client much like gomemcache. Publish/Subscribe operations are performed against a single server.

import "github.com/asim/emque/client/selector"

c := client.New(
	client.WithServers("10.0.0.1:8081", "10.0.0.1:8082", "10.0.0.1:8083"),
	client.WithSelector(new(selector.Shard)),
)

Resolver

A name resolver can be used to discover the ip addresses of MQ servers

import "github.com/asim/emque/client/resolver"

c := client.New(
	// use the DNS resolver
	client.WithResolver(new(resolver.DNS)),
	// specify DNS name as server
	client.WithServers("mq.proxy.local"),
)

Documentation

Index

• Variables
• func Publish(topic string, payload []byte) error
• func Subscribe(topic string) (<-chan []byte, error)
• func Unsubscribe(ch <-chan []byte) error
• type Client
  • func New(opts ...Option) Client
• type Option
  • func WithResolver(r Resolver) Option
  • func WithRetries(i int) Option
  • func WithSelector(s Selector) Option
  • func WithServers(addrs ...string) Option
• type Options
• type Resolver
• type Selector

Variables

var (
	// The default client
	Default = New()
	// The default server list
	Servers = []string{"http://127.0.0.1:8081"}
	// The default number of retries
	Retries = 1
)

Functions

func Publish

func Publish(topic string, payload []byte) error

Publish via the default Client

func Subscribe

func Subscribe(topic string) (<-chan []byte, error)

Subscribe via the default Client

func Unsubscribe

func Unsubscribe(ch <-chan []byte) error

Unsubscribe via the default Client

Types

type Client

type Client interface {
	Close() error
	Publish(topic string, payload []byte) error
	Subscribe(topic string) (<-chan []byte, error)
	Unsubscribe(<-chan []byte) error
}

Client is the interface provided by this package

func New

func New(opts ...Option) Client

New returns a new Client

type Option

type Option func(o *Options)

func WithResolver

func WithResolver(r Resolver) Option

WithResolver sets the resolver used to get the server list

func WithRetries

func WithRetries(i int) Option

WithRetries sets the number of retry attempts

func WithSelector

func WithSelector(s Selector) Option

WithSelector sets the server selector used by the client

func WithServers

func WithServers(addrs ...string) Option

WithServers sets the servers used by the client

type Options

type Options struct {
	// Number of retry attempts
	Retries int
	// Resolver
	Resolver Resolver
	// Server list
	Servers []string
	// Selector
	Selector Selector
}

type Resolver

type Resolver interface {
	Resolve(name string) ([]string, error)
}

Resolver resolves a name to a list of servers

type Selector

type Selector interface {
	Get(topic string) ([]string, error)
	Set(servers ...string) error
}

Selector provides a server list to publish/subscribe to