simple_triggered_stack

README

simple-triggered-stack

A simple stack that stacks items until some limit is reached, then executes a callback to flush all items

Example

package main

import (
	"fmt"
	"os"
	"time"

	"github.com/sayden/simple-triggered-stack"
	"github.com/timjchin/unpuzzled"
)

type config struct {
	port string
}

var conf config

func main() {
	app := unpuzzled.NewApp()

	app.Name = "Server"
	app.Usage = "See --help for info"
	app.Authors = []unpuzzled.Author{{Name: "Mario Castro", Email: "mariocaster@gmail.com"}}

	app.Command = &unpuzzled.Command{
		Name:  "Main",
		Usage: "Launches the server",
		Variables: []unpuzzled.Variable{
			&unpuzzled.StringVariable{
				Name:        "port",
				Destination: &conf.port,
				Default:     "8080",
				Description: "Server port",
			},
		},
		Action: launch,
	}
	app.Run(os.Args)
}

type stackableString struct {
	simple_triggered_stack.Stackable
	data string
}

func launch() {
	msgN := 10

	//Channel that the stack will use to notify us that flushing has complete
	quit := make(chan struct{})

	stackConfig := simple_triggered_stack.Config{
		MaxStack:           3,
		MaxIngestionBuffer: 10,
	}

	stackCallback := func(stack []simple_triggered_stack.Stackable) {
		for _, v := range stack {
			if s, ok := v.(*stackableString); ok {
				fmt.Printf("%s ", s.data	)
			}
		}

		fmt.Println()
	}

	stack := simple_triggered_stack.NewStack(quit, stackConfig, stackCallback)

	for i := 0; i < msgN; i++ {
		time.Sleep(time.Millisecond * 100)
		stack.IngestionCh() <- &stackableString{data: fmt.Sprintf("Hello %d", i)}
	}

	//We have finished pushing, close ingestion channel. Queue will notify of successful flushing by closing quit ch
	close(stack.IngestionCh())

	<-quit
}

Documentation

Index

• func NewStack(quit chan struct{}, c Config, cb func([]Stackable)) (s *stack)
• func NewWindowedStack(o ...*WindowedStackOptions) (w *windowedStack)
• type Config
• type GroupFlusher
• type Grouped
• type Stackable
• type WindowedStackOptions

Functions

func NewStack

func NewStack(quit chan struct{}, c Config, cb func([]Stackable)) (s *stack)

NewStack is the preferred way to create a new stack instance. quit channel will be 'notified' (by closing it) if the ingestion channel is closed AND the remaining data has been flushed using the callback

func NewWindowedStack

func NewWindowedStack(o ...*WindowedStackOptions) (w *windowedStack)

Types

type Config

type Config struct {
	//MaxStack represents the number of items that must be stacked before executing the callback
	MaxStack int

	//MaxIngestionBuffer is the buzzer size of the channel that receives the data
	MaxIngestionBuffer int
}

Config used to configure the stack

type GroupFlusher

type GroupFlusher interface {
	Flush([]Grouped)
}

type Grouped

type Grouped interface {
	Group() string
	SetGroup(string)
	Size() int64
	FlushAction() GroupFlusher
}

type Stackable

type Stackable interface{}

Stackable must be satisfied for any interface that wants to be stacked using this library

type WindowedStackOptions

type WindowedStackOptions struct {
	MaxBytesSize int64
	MaxStackSize int
	Timeout      time.Duration
}