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

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

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

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

soheilhy

app := beehive.NewApp("taskq")
app.Handle(Enque{}, EnQHandler{})
app.Handle(Deque{}, DeQHandler{})
app.Handle(Ack{}, AckHandler{})
app.Handle(Timeout{}, TimeoutHandler{
	ExpDur: 60 * time.Second,
})

// Detached runs a local go-routine, and NewTimer 
// creates a "detached" handler that calls the function

soheilhy

// RegisterTaskQ registers the taskq application and all its handler in the
// hive.
func RegisterTaskQ(h beehive.Hive) {
	a := h.NewApp("taskq", beehive.Persistent(3))
	a.Handle(Enque{}, EnQHandler{})
	a.Handle(Deque{}, DeQHandler{})
	a.Handle(Ack{}, AckHandler{})
	a.Handle(Timeout{}, TimeoutHandler{
		ExpDur: 60 * time.Second,
	})

soheilhy

func main() {
	h := beehive.NewHive()
	taskq.RegisterTaskQ(h)
	glog.Info("taskq started")
	h.Start()
}

soheilhy

func main() {
	h := beehive.NewHive()
	taskq.RegisterTaskQ(h)
	go h.Start()
	defer h.Stop()

	q := taskq.Queue("MyQueue")
	b := "copy f1 to f2"

	enq := taskq.Enque{Task: taskq.Task{Queue: q, Body: []byte(b)}}

soheilhy

type httpHandler struct {
	Hive beehive.Hive // Hive represents the hive our handler is registered on.
}

// EnQHTTPHandler provides the HTTP endpoint for enqueuing tasks.
type EnQHTTPHandler httpHandler

func (h *EnQHTTPHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
	q, ok := mux.Vars(r)["queue"]
	if !ok {