broker

Documentation

Overview

Example (Full_replay)

uuid := setup()

w, _ := NewWriter(uuid)
w.Write([]byte("busl"))
w.Write([]byte(" hello"))
w.Write([]byte(" world"))

r, _ := NewReader(uuid)
defer r.(io.Closer).Close()

buf := make([]byte, 16)
io.ReadAtLeast(r, buf, 16)

fmt.Printf("%s", buf)

Output:

busl hello world

Example (Half_replay_half_subscribed)

uuid := setup()

w, _ := NewWriter(uuid)
w.Write([]byte("busl"))

r, _ := NewReader(uuid)

pub := make(chan bool)
done := make(chan bool)

go func() {
	<-pub
	io.Copy(os.Stdout, r)
	done <- true
}()

go func() {
	pub <- true

	w.Write([]byte(" hello"))
	w.Write([]byte(" world"))
	w.Close()
}()

<-done

Output:

busl hello world

Example (Pub_sub)

uuid := setup()

r, _ := NewReader(uuid)
defer r.(io.Closer).Close()

pub := make(chan bool)
done := make(chan bool)

go func() {
	<-pub
	io.Copy(os.Stdout, r)
	done <- true
}()

go func() {
	pub <- true

	w, _ := NewWriter(uuid)
	w.Write([]byte("busl"))
	w.Write([]byte(" hello"))
	w.Write([]byte(" world"))
	w.Close()
}()

<-done

Output:

busl hello world

Example (Subscribe_concurrent)

r, w := newReaderWriter()

pub := make(chan bool)
done := make(chan bool)

go func() {
	pub <- true
	w.Write([]byte("busl"))
	w.Close()
}()

go func() {
	<-pub
	io.Copy(os.Stdout, r)
	done <- true
}()

<-done

Output:

busl

Index

• Variables
• func Get(key string) ([]byte, error)
• func Len(wd io.WriteCloser) (int64, error)
• func NewReader(key string) (io.ReadCloser, error)
• func NewWriter(key string) (io.WriteCloser, error)
• func RenewExpiry(rd io.Reader)
• type Conn
  • func (c Conn) Close() error
• type RedisRegistrar
  • func NewRedisRegistrar() *RedisRegistrar
  • func (rr *RedisRegistrar) IsRegistered(channelName string) (registered bool, err error)
  • func (rr *RedisRegistrar) Register(channelName string) (err error)
• type Registrar

Examples

• Package (Full_replay)
• Package (Half_replay_half_subscribed)
• Package (Pub_sub)
• Package (Subscribe_concurrent)

Variables

var (
	ErrNotRegistered = errors.New("Channel is not registered.")
	ErrClosed        = errors.New("Channel is closed.")
)

known errors

Functions

func Get

func Get(key string) ([]byte, error)

Get returns a key value

func Len

func Len(wd io.WriteCloser) (int64, error)

Len returns the length of data already send to the reader

func NewReader

func NewReader(key string) (io.ReadCloser, error)

NewReader creates a new redis channel reader

func NewWriter

func NewWriter(key string) (io.WriteCloser, error)

NewWriter creates a new redis channel writer

func RenewExpiry

func RenewExpiry(rd io.Reader)

RenewExpiry renews the channel expiration

Types

type Conn

type Conn struct {
	redis.Conn
	// contains filtered or unexported fields
}

func (Conn) Close

func (c Conn) Close() error

type RedisRegistrar

type RedisRegistrar struct{}

RedisRegistrar is a channel storing data on redis

func NewRedisRegistrar

func NewRedisRegistrar() *RedisRegistrar

NewRedisRegistrar creates a new registrar instance

func (*RedisRegistrar) IsRegistered

func (rr *RedisRegistrar) IsRegistered(channelName string) (registered bool, err error)

IsRegistered checks whether a channel name is registered

func (*RedisRegistrar) Register

func (rr *RedisRegistrar) Register(channelName string) (err error)

Register registers the new channel

type Registrar

type Registrar interface {
	Register(key string) error
	IsRegistered(key string) bool
}

Registrar is a basic broker interface