mb

README

Message batching queue

Docs

https://godoc.org/gopkg.in/cheggaaa/mb.v1

Installation

go get -u gopkg.in/cheggaaa/mb.v1

Example

package main

import (
	"fmt"
	"time"

	"gopkg.in/cheggaaa/mb.v1"
)

func main() {
	// create the queue with 10 items capacity
	q := mb.New(10)

	// create the channel for showing when all work will be done
	done := make(chan bool)

	// start two workers
	go worker("first", q, done)
	go worker("second", q, done)

	// start two publishers
	go publisher("first", q)
	go publisher("second", q)

	// give time to work
	time.Sleep(time.Second)

	// close the queue
	q.Close()

	// and wait until all sent messages will be processed
	for i := 0; i < 2; i++ {
		<-done
	}
}

func publisher(name string, q *mb.MB) {
	fmt.Printf("Publisher %s: started\n", name)
	var i int
	for {
		// will sending name and counter
		msg := fmt.Sprintf("%s - %d", name, i)
		// add
		if err := q.Add(msg); err != nil {
			// non-nil err mean that queue is closed
			break
		}
		// 10 messages per second
		time.Sleep(time.Second / 10)
		i++
	}
	fmt.Printf("Publisher %s: closed\n", name)
}

func worker(name string, q *mb.MB, done chan bool) {
	fmt.Printf("Worker %s: started\n", name)
	for {
		// getting messages
		msgs := q.Wait()

		if len(msgs) == 0 {
			// 0 messages mean that queue is closed
			break
		}

		msgsForPrint := ""
		for _, msg := range msgs {
			msgsForPrint += fmt.Sprintf("\t%s\n", msg)
		}
		fmt.Printf("Worker %s: %d messages recieved\n%s", name, len(msgs), msgsForPrint)

		// doing working, for example, send messages to remote server
		time.Sleep(time.Second / 3)
	}
	fmt.Printf("Worker %s: closed\n", name)
	done <- true
}

Documentation

Overview

Package mb - queue with message batching feature

Index

• Variables
• type MB
  • func New(size int) *MB
  • func (mb *MB) Add(msgs ...interface{}) (err error)
  • func (mb *MB) Close() (err error)
  • func (mb *MB) GetAll() (msgs []interface{})
  • func (mb *MB) Len() (l int)
  • func (mb *MB) Stats() (addCount, addMsgsCount, getCount, getMsgsCount int64)
  • func (mb *MB) Wait() (msgs []interface{})
  • func (mb *MB) WaitMax(max int) (msgs []interface{})
  • func (mb *MB) WaitMin(min int) (msgs []interface{})
  • func (mb *MB) WaitMinMax(min, max int) (msgs []interface{})

Variables

var ErrClosed = errors.New("MB closed")

ErrClosed is returned when you add message to closed queue

var ErrTooManyMessages = errors.New("Too many messages")

ErrTooManyMessages means that adding more messages (at one call) than the limit

Types

type MB

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

MB - message batching object Implements queue. Based on condition variables

func New

func New(size int) *MB

New returns a new MB with given queue size. size <= 0 means unlimited

func (*MB) Add

func (mb *MB) Add(msgs ...interface{}) (err error)

Add - adds new messages to queue. When queue is closed - returning ErrClosed When count messages bigger then queue size - returning ErrTooManyMessages When the queue is full - wait until will free place

func (*MB) Close

func (mb *MB) Close() (err error)

Close closes the queue All added messages will be available for Wait

func (*MB) GetAll

func (mb *MB) GetAll() (msgs []interface{})

GetAll return all messages and flush queue Works on closed queue

func (*MB) Len

func (mb *MB) Len() (l int)

Len returning current size of queue

func (*MB) Stats

func (mb *MB) Stats() (addCount, addMsgsCount, getCount, getMsgsCount int64)

Stats returning current statistic of queue usage addCount - count of calls Add addMsgsCount - count of added messages getCount - count of calls Wait getMsgsCount - count of issued messages

func (*MB) Wait

func (mb *MB) Wait() (msgs []interface{})

Wait until anybody add message Returning array of accumulated messages When queue will be closed length of array will be 0

func (*MB) WaitMax

func (mb *MB) WaitMax(max int) (msgs []interface{})

WaitMax it's Wait with limit of maximum returning array size

func (*MB) WaitMin

func (mb *MB) WaitMin(min int) (msgs []interface{})

WaitMin it's Wait with limit of minimum returning array size

func (*MB) WaitMinMax

func (mb *MB) WaitMinMax(min, max int) (msgs []interface{})

WaitMinMax it's Wait with limit of minimum and maximum returning array size value < 0 means no limit