soheilhy

func (h TimeoutHandler) Map(msg beehive.Msg,
	ctx beehive.MapContext) beehive.MappedCells {
	// Broadcast the timeout message to all local bees.
	return beehive.MappedCells{}
}

soheilhy

var port = args.NewInt(args.Flag("example.typed.port", 1234, "the port"))
var roundTripper = args.New(args.Default(http.DefaultTransport))
var timeout = args.NewDuration()

type ServerOpt args.V

func Port(p int) ServerOpt                 	  { return ServerOpt(port(p)) }
func RoundTripper(r http.RoundTrippper) ServerOpt { return ServerOpt(roundTripper(r)) }
func Timeout(d time.Duration) ServerOpt           { return ServerOpt(timeout(d)) }

soheilhy

// TimeoutHandler handles Ack messages.
type TimeoutHandler struct {
	// ExpDur is the duration after which a dequed and unacknowledged task
	// is returned to the active dictionary.
	ExpDur time.Duration
}

func (h TimeoutHandler) Rcv(msg beehive.Msg, ctx beehive.RcvContext) error {
	tout := time.Time(msg.Data().(Timeout))
	ddict := ctx.Dict(dequed)

soheilhy

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

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

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

// 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

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

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

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)
	}