md

Documentation

Overview

Package md Core library for mq dispatcher.

Index

• type BootManager
• type ConsumerContainer
• type ConsumerManager
• type ConsumerWorker
• type ProducerBucket
• type ProducerManager
• type RemoterManager
• type RetryKind
• type RetryManager

Types

type BootManager

type BootManager interface {
	// Consumer
	// return consumer manager interface.
	//
	//   x := md.Boot.Consumer()
	//   x.Container().Flush(ctx)
	Consumer() ConsumerManager

	// Processor
	// return boot processor interface.
	//
	//   x := md.Boot.Processor()
	//   x.Start(ctx)
	Processor() process.Processor

	// Producer
	// return producer manager interface.
	//
	//   x := md.Boot.Producer()
	//   x.Publish(payload)
	Producer() ProducerManager

	// Retry
	// return retry manager interface.
	//
	//   x := md.Boot.Retry()
	//   x.Publish()
	Retry() RetryManager

	// Remoter
	// return remoter manager interface.
	//
	//   x := md.Boot.Remoter()
	//   x.Build(task)
	Remoter() RemoterManager
}

var Boot BootManager

type ConsumerContainer

type ConsumerContainer interface {
	// Flush
	// start adapter if not started. Stop adapter if task
	// disabled or deleted or parallels changed down.
	Flush(ctx context.Context)

	// IsEmpty
	// return container has adapter or not.
	//
	// Return true means no adapter in container, otherwise false
	// returned.
	IsEmpty() bool

	// Reload
	// call memory manager then flush adapters.
	Reload(ctx context.Context) error

	// Worker
	// return consumer worker interface.
	Worker() ConsumerWorker
}

type ConsumerManager

type ConsumerManager interface {
	// Container
	// return consumer container interface.
	Container() ConsumerContainer

	// Processor
	// return consumer processor interface.
	Processor() process.Processor

	// Reload
	// send consumer reload process.
	Reload()
}

type ConsumerWorker

type ConsumerWorker interface {
	// Do
	// worker process.
	//
	// - Dispatch message
	// - Send notification if enabled.
	// - Store received message
	Do(t *base.Task, m *base.Message) (retry bool)

	// IsIdle
	// return idle status.
	//
	// Return false if any job is not completed, otherwise true returned.
	IsIdle() bool
}

type ProducerBucket

type ProducerBucket interface {
	// IsEmpty
	// return bucket is empty or not.
	//
	// Return true if no payload in bucket, otherwise false
	// returned.
	IsEmpty() (yes bool)

	// IsFull
	// return bucket is full or not.
	//
	// Return true if payload count is greater or equal to size,
	// otherwise false returned.
	IsFull() (yes bool)

	// Length
	// return payloads count in bucket.
	Length() (count int)

	// Pop
	// get one payload from left cached.
	Pop() (payload *base.Payload)

	// Popn
	// get specified count payloads from left cached.
	Popn(n int) (list []*base.Payload, count int)

	// Push
	// add payload to right cached.
	Push(ps ...*base.Payload) error
}

type ProducerManager

type ProducerManager interface {
	// Bucket
	// return producer bucket interface.
	Bucket() ProducerBucket

	// Processor
	// return producer processor interface.
	Processor() process.Processor

	// Publish
	// send payloads to channel.
	//
	// Return immediately do not wait process completed.
	Publish(ps ...*base.Payload) (err error)

	// PublishDirect
	// send payload directly in sync coroutine.
	//
	// Return when process completed.
	PublishDirect(p *base.Payload) (err error)
}

type RemoterManager

type RemoterManager interface {
	// Adapter
	// return remoter adapter interface.
	Adapter() adapters.RemoterAdapter

	// Processor
	// return remoter processor interface.
	//
	//   x := md.Boot.Remoter().Processor()
	//   x.Start(ctx)
	Processor() process.Processor
}

type RetryKind

type RetryKind int

const (
	RetryKindMessage RetryKind
	RetryKindPayload
)

type RetryManager

type RetryManager interface {
	// Message
	// read waiting messages in database then call consume.
	//
	// - Waiting messages:
	//   SELECT * FROM `message` WHERE `status` = 3 LIMIT 10
	//
	// - Call retry:
	//   x := md.Boot.Retry()
	//   x.Message()
	Message()

	// Payload
	// read waiting payloads in database then call publish.
	//
	// - Waiting payloads:
	//   SELECT * FROM `payload` WHERE `status` = 3 LIMIT 10
	//
	// - Call:
	//   x := md.Boot.Retry()
	//   x.Payload()
	Payload()

	// Processor
	// return retry processor interface.
	//
	//   x := md.Boot.Retry().Processor()
	//   x.Start(ctx)
	Processor() process.Processor
}