dispatcher

Documentation

Index

• type Factory
• type Request
• type Response
• type T
  • func New(namespace *actor.ID, factory Factory, cfg *config.Proxy) *T
  • func (d *T) Requests() chan<- Request
  • func (d *T) Start()
  • func (d *T) Stop()
• type Tier

Types

type Factory

type Factory interface {
	// dispatchKey returns a key that a child created for the specified `req`
	// should have.
	KeyOf(req Request) string

	// NewTier creates a new dispatch tier to handle requests with the
	// specified dispatch key.
	NewTier(key string) Tier
}

Factory defines an interface to create Tiers.

type Request

type Request struct {
	Timestamp  time.Time
	Group      string
	Topic      string
	ResponseCh chan<- Response
}

type Response

type Response struct {
	Msg consumer.Message
	Err error
}

type T

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

T reads consume requests submitted to the `Requests()` channel and dispatches them to downstream dispatch tiers based on the dispatch key of requests. Dispatcher uses `dispatchTierFactory` to resolve a dispatch key from a consume request and create `dispatchTier` instances on demand. When a down stream dispatch tier is created it is cached in case more requests resolving to it will come in the nearest future.

func New

func New(namespace *actor.ID, factory Factory, cfg *config.Proxy) *T

func (*T) Requests

func (d *T) Requests() chan<- Request

func (*T) Start

func (d *T) Start()

func (*T) Stop

func (d *T) Stop()

type Tier

type Tier interface {
	// Key returns the dispatch key of the tier.
	Key() string

	// Requests returns a channel to send requests dispatched to the tier to.
	Requests() chan<- Request

	// Start spins up the tier's goroutine(s).
	Start(stoppedCh chan<- Tier)

	// Stop makes all tier goroutines stop and releases all resources.
	Stop()
}

Tier defines a consume request handling tier interface.