queue2

README

go-deq

Double Evented Queue

func main() {

  unsub := queue2.Subscribe(func(s queue2.Item) {
    fmt.Println("got message", s.(string))
  })

  go func() {
    time.Sleep(10 * time.Second)
    unsub()
    queue2.Subscribe(func(s queue2.Item) {
      fmt.Println("[resub] got message", s.(string))
    })
  }()

  queue2.PCap = 5
  // call on queue2.LCap limit
  queue2.PC = func(c int) {
    fmt.Println("Primary queue limit reached: ", c)
  }

  queue2.SCap = 3
  // call on queue2.GCap limit
  queue2.SC = func(c int) {
    fmt.Println("Secondary queue limit reached: ", c)
  }
 

 // capture realtime traffic
  go func() {
    for {
      time.Sleep(1 * time.Second)
      queue2.WriteSecondary(fmt.Sprintf("global-%d", time.Now().Unix()))
    }
  }()

  // read from local file
  go func() {
    for i := 0; i <= 10; i++ {
      time.Sleep(500 * time.Millisecond)
      queue2.WritePrimary(fmt.Sprintf("local-%d", i))

    }
    // local file parse done, switch to secondary queue
    queue2.Switch()
  }()

  // run forever
  runtime.Goexit()
}

func main() {
  var lt = queue2.New()

  lt.WritePrimary(func() { println("fn 1") })
  lt.WritePrimary(func() { println("fn 2") })
  lt.WritePrimary(func() { println("fn 3") })

  for i, e := lt.ProcessPrimary(); e == nil; i, e = lt.ProcessPrimary() {
    i.(func())()
  }
}

Documentation

Index

• type Item
• type Q
  • func New() *Q
  • func (t *Q) ProcessPrimary() (Item, error)
  • func (t *Q) ProcessPrimarySync(fn func(i Item))
  • func (t *Q) ProcessSecondary() (Item, error)
  • func (t *Q) ProcessSecondarySync(fn func(i Item))
  • func (t *Q) Subscribe(fn func(Item)) (ubsub func())
  • func (t *Q) Switch()
  • func (t *Q) WritePrimary(m Item)
  • func (t *Q) WriteSecondary(m Item)

Types

type Item

type Item interface{}

type Q

type Q struct {
	PC   func(int)
	SC   func(int)
	PCap int
	SCap int

	OnSwitch func()
	// contains filtered or unexported fields
}

func New

func New() *Q

func (*Q) ProcessPrimary

func (t *Q) ProcessPrimary() (Item, error)

func (*Q) ProcessPrimarySync

func (t *Q) ProcessPrimarySync(fn func(i Item))

func (*Q) ProcessSecondary

func (t *Q) ProcessSecondary() (Item, error)

func (*Q) ProcessSecondarySync

func (t *Q) ProcessSecondarySync(fn func(i Item))

func (*Q) Subscribe

func (t *Q) Subscribe(fn func(Item)) (ubsub func())

func (*Q) Switch

func (t *Q) Switch()

func (*Q) WritePrimary

func (t *Q) WritePrimary(m Item)

func (*Q) WriteSecondary

func (t *Q) WriteSecondary(m Item)