mapper

type Executor ¶

type Executor interface {
	// BrokerConn returns a broker connection used by the executor.
	BrokerConn() *sarama.Broker

	// Stop synchronously stops the executor.
	Stop()
}

Executor represents an entity that executes requests of workers via a particular broker connection.

type Resolver ¶

type Resolver interface {
	// ResolveBroker returns a broker connection that should be used to
	// determine an executor assigned to the specified worker.
	ResolveBroker(worker Worker) (*sarama.Broker, error)

	// SpawnExecutor spawns an executor for the specified connection.
	SpawnExecutor(brokerConn *sarama.Broker) Executor
}

Resolver defines an interface to resolve a broker connection that should serve requests of a particular worker, and to create an executor for a broker connection.

type T ¶

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

T maintains mapping of workers that generate requests to executors. An executor is associated with a particular Kafka broker. It aggregates worker requests, marshals them to a Kafka protocol packet, sends the packet to the associated broker, waits for response and fans replies out to the workers. An external resolver is used to determine worker to broker-executor assignments.

Mapper triggers reassignment whenever one of the following events happen:

• it is signaled that a new worker has been spawned via `OnWorkerSpawned()`;
• it is signaled that an existing worker has stopped via `OnWorkerStopped()`;
• a worker explicitly requested reassignment via `TriggerReassign()`
• an executor reported connection error via `BrokerFailed()`.

Executors are spawned on demand when a broker is resolved to a worker for the first time. It is guaranteed that a executor is stopped only after all workers that used to be assigned to it have either been stopped or assigned another to other executors.

func Spawn(parentActDesc *actor.Descriptor, cfg *config.Proxy, resolver Resolver) *T

func (m *T) OnWorkerSpawned(w Worker)

func (m *T) OnWorkerStopped(w Worker)

func (m *T) Stop()

func (m *T) TriggerReassign(w Worker)

type Worker ¶

type Worker interface {
	// assignment returns a channel that the worker expects broker assignments
	// at. Implementations have to ensure that the channel has a non zero buffer
	// and that they read from this channel as soon as the value becomes
	// available, for mapper will drop assignments in case the write to the
	// channel may block.
	Assignment() chan<- Executor
}

Worker represents an entity that makes requests via an assigned broker executor.