soheilhy

// TaskRing is an efficient, gob-compatible ring buffer for Tasks.
// TaskRing always wastes one element.
type TaskRing struct {
	...
	Stats Stats
}

// Stats represents the statistics of the task ring buffer.
type Stats struct {
	Deque  uint64 // Deque represents the total nubmer of dequeued tasks.

soheilhy

import (
	"github.com/kandoo/beehive"
)

// EnQHandler handles Enque messages.
type EnQHandler struct{}

func (h EnQHandler) Rcv(msg beehive.Msg, ctx beehive.RcvContext) error {...}

func (h EnQHandler) Map(msg beehive.Msg, ctx beehive.MapContext) beehive.MappedCells {...}

soheilhy

const (
	// Dictionary names.
	active = "active"
	dequed = "dequed"
)

func (h EnQHandler) Rcv(msg beehive.Msg, ctx beehive.RcvContext) error {
	// TODO: assign a unique ID.
	enq := msg.Data().(Enque)

soheilhy

func (h EnQHandler) Rcv(msg beehive.Msg, ctx beehive.RcvContext) error {
	enq := msg.Data().(Enque)

	dict := ctx.Dict(active)
	key := string(enq.Task.Queue)

	var tasks TaskRing
	if val, err := dict.Get(key); err == nil {
		tasks = val.(TaskRing)
	}

soheilhy

func (h EnQHandler) Map(msg beehive.Msg, ctx beehive.MapContext) beehive.MappedCells {
	// Send the Enque message to the bee that owns
	// the Queue's entry in the active dictionary.
	q := string(msg.Data().(Enque).Queue)
	return beehive.MappedCells{{active, q}}
}

soheilhy

package main

import (
	"fmt"

	"github.com/soheilhy/args"
)

var ListenOn = args.NewInt(args.Default(8080))
var BufferSize = args.NewUint64(args.Default(uint64(1024 * 1024)))

soheilhy

// DeQHandler handles Deque messages.
type DeQHandler struct{}

func (h DeQHandler) Rcv(msg beehive.Msg, ctx beehive.RcvContext) error {
	deq := msg.Data().(Deque)

	// (a) we deque the task from active tasks.
	adict := ctx.Dict(active)
	key := string(deq.Queue)

soheilhy

func (h DeQHandler) Map(msg beehive.Msg,
	ctx beehive.MapContext) beehive.MappedCells {

	// Send the Deque message to the bee that owns the Queue's entry in
	// the active and the dequed dictionaries.
	q := string(msg.Data().(Deque).Queue)
	return beehive.MappedCells{{active, q}, {dequed, q}}
}

soheilhy

// AckHandler handles Ack messages.
type AckHandler struct{}

func (h AckHandler) Rcv(msg beehive.Msg, ctx beehive.RcvContext) error {
	ack := msg.Data().(Ack)

	ddict := ctx.Dict(dequed)
	key := string(ack.Queue)

	if val, err := ddict.Get(key); err == nil {

soheilhy

// Timeout represents a timeout message.
type Timeout time.Time