contentconsumer

README

contentconsumer

Go Lang module - Ring Buffer based timed consumer of content

Documentation

Overview

Package contentconsumer - Consumes content from channel. (e.g.) TimedContentConsumer

Index

• Variables
• type ConsumerEvent
• type ConsumerEventConsumer
• type ConsumerEventPtr
• type TimedContent
• type TimedContentConsumer
  • func NewTimedContentConsumer(id string, interval time.Duration, slots int) *TimedContentConsumer
  • func (cc *TimedContentConsumer) Channel() chan<- TimedContentPtr
  • func (cc *TimedContentConsumer) CloseChannel()
  • func (cc *TimedContentConsumer) DrainChunks(ctx context.Context, wg *sync.WaitGroup)
  • func (cc *TimedContentConsumer) Run(wg *sync.WaitGroup)
• type TimedContentPtr
  • func NewTimedContent(duration time.Duration, data interface{}) TimedContentPtr

Variables

var ErrConsumerChannelClosed error = errors.New("ChannelClosed")

ErrConsumerChannelClosed - AddChunk to closed channel

var ErrConsumerDrainedEmpty error = errors.New("EmptyBuffer")

ErrConsumerDrainedEmpty - Empty Consumer Buffer

Types

type ConsumerEvent

type ConsumerEvent struct {
	ID  string
	Err error
	// contains filtered or unexported fields
}

ConsumerEvent - Events

type ConsumerEventConsumer

type ConsumerEventConsumer interface {
	//Channel - Return Channel to write to.  nil if channel is closed
	Channel() chan<- ConsumerEventPtr
	//Signal Close of channel when writing is finished
	CloseChannel()
}

ConsumerEventConsumer -

type ConsumerEventPtr

type ConsumerEventPtr *ConsumerEvent

ConsumerEventPtr - Ptr to ConsumerEvent

type TimedContent

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

TimedContent - Content with duration duration - content duration data - actual content

type TimedContentConsumer

type TimedContentConsumer struct {

	//id - User name for the TimedContentConsumer
	// used in the logging
	ID string
	//Interval to drain buffer for content worth the interval
	ConsumptionInterval time.Duration

	//downstream Consumer of content drained
	Downstream chan<- interface{}
	//downstream Consumer of events
	EventDownstream chan<- ConsumerEventPtr
	// contains filtered or unexported fields
}

TimedContentConsumer -

• Maintains a Cache of given ConsumptionInterval
• Channel may have more
• Drains the Cache in fixed intervals
• While Draining can forward data using PostDataFunc(duration, data)
• If CacheBuffer below Min Level - calls BelowMinLevelFunc()

func NewTimedContentConsumer

func NewTimedContentConsumer(id string, interval time.Duration, slots int) *TimedContentConsumer

NewTimedContentConsumer - Create

id - User Name
interval - Duration of poll
slots - intial ring buffer size estimate content to fill
        No worries it will be doubled if required

func (*TimedContentConsumer) Channel

func (cc *TimedContentConsumer) Channel() chan<- TimedContentPtr

Channel - Returns the channel to write to

func (*TimedContentConsumer) CloseChannel

func (cc *TimedContentConsumer) CloseChannel()

CloseChannel - Close the input channel This will make the Go Routine Run exit

func (*TimedContentConsumer) DrainChunks

func (cc *TimedContentConsumer) DrainChunks(ctx context.Context, wg *sync.WaitGroup)

DrainChunks - Drain the actual content

func (*TimedContentConsumer) Run

func (cc *TimedContentConsumer) Run(wg *sync.WaitGroup)

Run - Start Consuming Content

type TimedContentPtr

type TimedContentPtr *TimedContent

TimedContentPtr - Ptr to the TimedContext

func NewTimedContent

func NewTimedContent(duration time.Duration, data interface{}) TimedContentPtr

NewTimedContent - Create new TimedContent